分布式爬虫稳爬方案：短效动态代理集群搭建实战教程

# 基础爬虫与分布式必备依赖
pip install requests scrapy redis fake-useragent threading
# 异步高并发可选
pip install aiohttp

import requests
import time

# 代理配置项
AUTH_KEY = "你的个人auth_key"
API_URL = "https://tunnel.zhandaye.com/api"
PROXY_NUM = 100
PROXY_TIME = 5
PROXY_TYPE = 2

def get_dynamic_proxy():
    """批量获取标准化短效代理IP"""
    params = {
        "auth_key": AUTH_KEY,
        "num": PROXY_NUM,
        "time": PROXY_TIME,
        "type": PROXY_TYPE
    }
    try:
        res = requests.get(API_URL, params=params, timeout=10)
        if res.status_code == 200 and res.json()["code"] == 200:
            proxy_list = res.json()["data"]["list"]
            format_proxies = [f"{p['ip']}:{p['port']}" for p in proxy_list]
            print(f"成功获取 {len(format_proxies)} 个短效代理IP")
            return format_proxies
        else:
            print("代理获取失败：", res.json().get("msg", "未知错误"))
            return []
    except Exception as e:
        print("代理API请求异常：", str(e))
        return []

if __name__ == "__main__":
    proxy_list = get_dynamic_proxy()
    print("代理示例：", proxy_list[:10])

def check_proxy(proxy):
    """校验单个代理可用性"""
    test_url = "https://www.baidu.com"
    proxies = {
        "http": f"http://{proxy}",
        "https": f"http://{proxy}"
    }
    try:
        res = requests.get(test_url, proxies=proxies, timeout=3)
        return res.status_code == 200
    except:
        return False

def get_valid_proxies():
    """获取清洗后的有效代理池"""
    raw_proxies = get_dynamic_proxy()
    valid_proxies = [p for p in raw_proxies if check_proxy(p)]
    print(f"校验完成，有效IP数量：{len(valid_proxies)}")
    return valid_proxies

import redis

# Redis配置
REDIS_HOST = "你的Redis服务器IP"
REDIS_PORT = 6379
REDIS_PASSWORD = "你的Redis密码"

def init_redis_pool():
    r = redis.Redis(
        host=REDIS_HOST,
        port=REDIS_PORT,
        password=REDIS_PASSWORD,
        decode_responses=True,
        db=0
    )
    # 清理旧无效代理
    r.delete("valid_proxy")
    # 写入新校验IP
    valid_proxies = get_valid_proxies()
    if valid_proxies:
        r.sadd("valid_proxy", *valid_proxies)
    print(f"Redis代理池初始化完成，可用IP：{r.scard('valid_proxy')}")
    return r

redis_client = init_redis_pool()

import random

def get_random_proxy():
    """随机获取有效代理，池空自动补充"""
    proxy = redis_client.spop("valid_proxy")
    if not proxy:
        print("代理池枯竭，自动补充新IP")
        new_proxies = get_valid_proxies()
        if new_proxies:
            redis_client.sadd("valid_proxy", *new_proxies)
            proxy = redis_client.spop("valid_proxy")
    return proxy

def proxy_recycle(proxy, is_valid):
    """代理回收机制"""
    if is_valid and proxy:
        redis_client.sadd("valid_proxy", proxy)

from fake_useragent import UserAgent
ua = UserAgent()

def spider_task(url):
    proxy = get_random_proxy()
    if not proxy:
        print("暂无可用代理，任务暂停")
        return False

    proxies = {
        "http": f"http://{proxy}",
        "https": f"http://{proxy}"
    }
    headers = {"User-Agent": ua.random}

    try:
        res = requests.get(url, headers=headers, proxies=proxies, timeout=5)
        if res.status_code in [200, 201]:
            print(f"成功抓取：{url} | 当前IP：{proxy}")
            proxy_recycle(proxy, True)
            return True
        else:
            proxy_recycle(proxy, False)
            return False
    except Exception as e:
        proxy_recycle(proxy, False)
        return False

def distributed_spider_run():
    """分布式节点持续消费任务"""
    while True:
        task_url = redis_client.lpop("spider_task")
        if not task_url:
            time.sleep(3)
            continue
        spider_task(task_url)
        time.sleep(1)

def push_task_to_redis(task_list):
    if not task_list:
        return
    redis_client.rpush("spider_task", *task_list)
    print(f"成功推送 {len(task_list)} 条任务")

if __name__ == "__main__":
    # 替换为自己的目标链接
    task_urls = [
        "https://www.example.com/1",
        "https://www.example.com/2",
        "https://www.example.com/3"
    ]
    push_task_to_redis(task_urls)
    distributed_spider_run()