代理IP在分布式爬虫架构中的落地方案：Redis+Scrapy实战

pip install scrapy scrapy-redis redis

import redis
import random

# Redis连接配置（生产环境务必开启密码、私有化部署）
REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
REDIS_DB = 0

# Redis键名统一规范，方便后续运维排查问题
PROXY_VALID_KEY = "spider:proxy:valid"   # 可用代理集合
PROXY_INVALID_KEY = "spider:proxy:invalid" # 失效代理集合

class RedisProxyClient:
    def __init__(self):
        self.redis_cli = redis.Redis(
            host=REDIS_HOST,
            port=REDIS_PORT,
            db=REDIS_DB,
            decode_responses=True,
            socket_timeout=5
        )

    # 添加可用代理IP到Redis池
    def add_valid_proxy(self, proxy):
        self.redis_cli.sadd(PROXY_VALID_KEY, proxy)

    # 随机获取一个可用代理
    def get_random_proxy(self):
        valid_proxies = self.redis_cli.smembers(PROXY_VALID_KEY)
        if not valid_proxies:
            return None
        return random.choice(list(valid_proxies))

    # 标记代理失效，转移到失效集合
    def mark_proxy_invalid(self, proxy):
        if not proxy:
            return
        self.redis_cli.srem(PROXY_VALID_KEY, proxy)
        self.redis_cli.sadd(PROXY_INVALID_KEY, proxy)

    # 重置失效IP（用于定时复活检测）
    def reset_invalid_proxy(self, proxy):
        self.redis_cli.srem(PROXY_INVALID_KEY, proxy)
        self.redis_cli.sadd(PROXY_VALID_KEY, proxy)

    # 获取当前可用代理总数
    def get_valid_count(self):
        return self.redis_cli.scard(PROXY_VALID_KEY)

from .redis_proxy import RedisProxyClient

class RedisProxyMiddleware:
    def __init__(self):
        self.proxy_client = RedisProxyClient()
        # 连续失败3次判定IP失效，可根据业务自行调整
        self.fail_limit = 3
        # 本地记录各IP失败次数，辅助判断失效
        self.proxy_fail_count = {}

    # 每次请求前自动分配代理
    def process_request(self, request, spider):
        # 跳过本地请求、手动指定代理的特殊场景
        if request.meta.get("proxy") or "127.0.0.1" in request.url:
            return
        proxy = self.proxy_client.get_random_proxy()
        if proxy:
            request.meta["proxy"] = proxy
            spider.logger.info(f"本次请求分配代理：{proxy}")

    # 请求异常处理，自动剔除劣质IP
    def process_exception(self, request, exception, spider):
        proxy = request.meta.get("proxy")
        if not proxy:
            return
        
        # 累计失败次数
        self.proxy_fail_count[proxy] = self.proxy_fail_count.get(proxy, 0) + 1
        
        # 达到阈值，剔除失效IP
        if self.proxy_fail_count[proxy] >= self.fail_limit:
            self.proxy_client.mark_proxy_invalid(proxy)
            spider.logger.warning(f"代理{proxy}多次请求失败，已移出可用池")
            del self.proxy_fail_count[proxy]
        
        # 重试请求，自动更换新IP
        return request.replace(dont_filter=True)

# 开启scrapy-redis分布式调度
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
# 重启爬虫不丢失任务，非常实用
SCHEDULER_PERSIST = True
# 分布式全局去重
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# Redis连接信息
REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379

# 中间件优先级配置，禁用原生代理中间件
DOWNLOADER_MIDDLEWARES = {
    'your_spider.middlewares.RedisProxyMiddleware': 110,
    'scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware': None,
    'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware': 400,
}

# 爬虫并发、延时配置，适配代理池承载力
CONCURRENT_REQUESTS = 32
DOWNLOAD_DELAY = 0.3
RANDOMIZE_DOWNLOAD_DELAY = True

# 失败重试配置，覆盖常见异常状态码
RETRY_TIMES = 2
RETRY_HTTP_CODES = [403, 429, 500, 502, 503, 504]

from redis_proxy import RedisProxyClient

if __name__ == "__main__":
    proxy_client = RedisProxyClient()
    # 替换成自己的代理IP，支持批量写入
    proxy_list = [
        "http://xxx.xxx.xxx.xxx:xxxx",
        "http://xxx.xxx.xxx.xxx:xxxx",
        "http://xxx.xxx.xxx.xxx:xxxx"
    ]
    # 批量存入Redis可用代理池
    for proxy in proxy_list:
        proxy_client.add_valid_proxy(proxy)
    print(f"成功初始化{len(proxy_list)}个代理IP至Redis代理池")