Scrapy 代理轮换中间件：按 IP 而非按请求轮换的做法

为什么不用现成的轮换池

现成的代理池组件大多按请求轮换，也就是每个请求换一个出口 IP。这在很多场景下是错的。

按请求轮换会导致：同一个会话里的请求散落在不同 IP 上，目标站点看到的是”一个用户在多台设备上同时操作”；分页抓取时上一页和下一页来自不同 IP，风控很容易命中；而且每次换 IP 都意味着新的连接建立，延迟成本被浪费掉。

更合理的模型是按 IP 分配一批请求：给每个出口 IP 一个粘性会话，让这个会话处理若干请求，遇到失败再换。

中间件实现

下面这个中间件维护一个会话池，每个会话绑定一个粘性出口，并按并发上限分配请求：

# myproject/middlewares.py
import itertools
import random
from collections import defaultdict

from scrapy import signals
from scrapy.exceptions import IgnoreRequest


class RotatingProxyMiddleware:
    """会话池式代理轮换：每个出口 IP 承载若干请求，失败时换出口。"""

    def __init__(self, gateway, username, password,
                 sessions=20, max_per_session=40, max_fails=3):
        self.gateway = gateway
        self.username = username
        self.password = password
        self.max_fails = max_fails
        self.max_per_session = max_per_session
        # 每个槽位一个固定会话 ID，保证出口在槽位存活期内不变
        self.slots = [
            {"sid": f"s{random.randint(10**7, 10**8 - 1)}",
             "used": 0, "fails": 0}
            for _ in range(sessions)
        ]
        self._cycle = itertools.cycle(range(len(self.slots)))

    @classmethod
    def from_crawler(cls, crawler):
        s = crawler.settings
        mw = cls(
            gateway=s.get("PROXY_GATEWAY"),
            username=s.get("PROXY_USER"),
            password=s.get("PROXY_PASSWORD"),
            sessions=s.getint("PROXY_SESSIONS", 20),
            max_per_session=s.getint("PROXY_MAX_PER_SESSION", 40),
        )
        crawler.signals.connect(mw.spider_closed, signal=signals.spider_closed)
        return mw

    def _pick_slot(self):
        """轮转取一个还能用的槽位；用满或失败过多的槽位换新会话 ID。"""
        for _ in range(len(self.slots) * 2):
            idx = next(self._cycle)
            slot = self.slots[idx]
            if slot["used"] >= self.max_per_session or slot["fails"] >= self.max_fails:
                slot["sid"] = f"s{random.randint(10**7, 10**8 - 1)}"
                slot["used"] = 0
                slot["fails"] = 0
            slot["used"] += 1
            return slot
        return self.slots[0]

    def _proxy_url(self, sid):
        user = f"{self.username}-session-{sid}"
        return f"http://{user}:{self.password}@{self.gateway}"

    def process_request(self, request, spider):
        slot = self._pick_slot()
        request.meta["proxy_slot"] = slot
        request.meta["proxy"] = self._proxy_url(slot["sid"])

    def process_response(self, request, response, spider):
        slot = request.meta.get("proxy_slot")
        if slot and response.status in (403, 429):
            # 被风控：记失败，换一个出口重试
            slot["fails"] += 1
            slot["used"] = self.max_per_session
            return request.replace(dont_filter=True)
        return response

    def process_exception(self, request, exception, spider):
        slot = request.meta.get("proxy_slot")
        if slot:
            slot["fails"] += 1
        return None       # 交给重试中间件处理

    def spider_closed(self, spider):
        alive = sum(1 for s in self.slots if s["fails"] < self.max_fails)
        spider.logger.info("proxy slots alive: %d/%d", alive, len(self.slots))

对应的 settings.py：

PROXY_GATEWAY = "gateway.example.com:8080"
PROXY_USER = "用户名"
PROXY_PASSWORD = "密码"
PROXY_SESSIONS = 20
PROXY_MAX_PER_SESSION = 40

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.RotatingProxyMiddleware": 543,
}

# 重试交给 Scrapy 内置的，但不要把 403 当成成功
RETRY_ENABLED = True
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 408, 429, 403]

# 降低并发，避免单槽位突发过高
CONCURRENT_REQUESTS = 16
CONCURRENT_REQUESTS_PER_DOMAIN = 8
DOWNLOAD_TIMEOUT = 45

三个设计要点

第一，槽位而不是请求作为轮换单位。 PROXY_MAX_PER_SESSION 控制一个出口 IP 承载多少个请求。这个值需要按目标站点调：防护越强，值应该越小。从 40 开始试，观察到 403 就往下调。

第二，失败时立刻淘汰槽位。 收到 403 或 429 说明这个出口在这个站点上已经不可用了，继续用它只会产生更多失败。代码里把 used 直接拉满，让下次取槽位时重置会话 ID。

第三，成功率要能被观测。 没有统计的轮换是盲调。加一个扩展或者在 spider_closed 里输出按状态码的分布，才谈得上优化：

# 在 Spider 里记录响应分类，收尾时打印
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.stats = defaultdict(int)

def parse(self, response):
    if response.status in (403, 429):
        self.stats["blocked"] += 1
        return
    body = response.text[:3000].lower()
    if "captcha" in body or "验证码" in body:
        self.stats["challenged"] += 1
        return
    self.stats["ok"] += 1
    # ...正常解析逻辑

def closed(self, reason):
    total = sum(self.stats.values()) or 1
    self.logger.info("success rate = %.1f%%  %s",
                     self.stats["ok"] / total * 100, dict(self.stats))

不要踩的两个坑

不要给每个请求新建代理连接。 Scrapy 的 DOWNLOAD_TIMEOUT 是从请求发出算起的，如果每个请求都换代理，TCP 握手时间会被反复计入，吞吐会明显下降。

不要把 dont_filter 用错。 失败重试时带上 dont_filter=True 是对的，避免被去重中间件挡掉；但正常的分页请求不要开，否则会重复抓取浪费流量。

小结

代理轮换的正确单位是会话，不是请求。用槽位模型维护一批粘性出口，按目标站点的防护强度调整单槽位的请求配额，并把成功率做成可观测的指标，比套用现成的轮换池组件更能拿到稳定的结果。