企业级采集的架构设计：代理IP的会话隔离与IP轮转

"我每个请求都换了 IP，为什么还是被限制？"

这是我在做数据采集时被问到最多的一个问题。问的人通常已经做了 IP 轮转——每个请求用不同的代理 IP——但成功率还是上不去。

最常见的原因不是 IP 不够多，而是会话层面没隔离。什么意思？

你有两个采集任务：任务 A 采集公开招投标信息，每小时 500 个请求；任务 B 采集公开舆情数据，每小时 2000 个请求。两个任务共用一个代理 IP 池。

表面上看，每个请求都用了不同的 IP。但如果池子不大（比如 500 个 IP），同一个 IP 很可能在 10 分钟内既被任务 A 用过、又被任务 B 用过。如果其中一个目标站对该 IP 做了访问频率标记，这个标记可能会影响同一 IP 在另一个目标站上的表现——尤其当两个目标站使用了同一套访问频率控制服务的时候。

更隐蔽的情况是：任务 B 的请求量远大于任务 A，它把池子里的高质量 IP 消耗得更快，导致任务 A 经常分到评分低的 IP——虽然任务 A 本身的请求量很小，完全不需要那么多 IP。

这些问题的根源是：两个业务任务共享了同一个代理池的状态空间。IP 轮转解决的是"同一个任务内不用同一个 IP 连续发请求"，会话隔离解决的是"不同任务之间不共享 IP 状态"。

会话隔离的三个层次

会话隔离不是一个开关，而是一个可以按粒度分层的架构决策。从粗到细分三层：

第一层：按任务隔离代理池（池级隔离）。 不同的采集任务使用不同的代理池。Redis 里建多个 Sorted Set——proxy:pool:task_bid、proxy:pool:task_sentiment——每个池子独立管理 IP 评分和淘汰。

好处是隔离最彻底：任务 B 把 IP 用废了不影响任务 A。代价是 IP 总需求量变大——如果你有 10 个任务、每个池子至少需要 200 个 IP，总共就要 2000 个。

适用场景：任务数量不多（3-5 个）、每个任务对成功率要求高、IP 预算充足。

第二层：共享池 + 使用记录隔离（域名级隔离）。 所有任务共享一个代理池，但在取 IP 时，检查这个 IP 最近 N 分钟内有没有被用于当前目标域名。如果用过，跳过，取下一个。

实现方式：在 Redis 里给每个 IP 维护一个"最近使用域名"的记录，用带 TTL 的 Key：

Key:    proxy:used:{ip:port}:{domain}
Value:  timestamp
TTL:    300 (5 分钟冷却)

取 IP 时，先查 proxy:used:{candidate}:{target_domain} 是否存在。存在说明这个 IP 5 分钟内访问过这个域名，换一个。

好处是 IP 利用率更高——同一个 IP 可以同时服务不同域名。代价是实现稍复杂，而且冷却时间需要调参。

适用场景：任务多、IP 预算有限、不同任务的目标站不重叠。

第三层：会话级隔离（粘性会话）。 某些采集任务需要在多个请求之间保持同一个 IP——比如需要先登录（用合规的公开 API 认证）再翻页的场景。这时不能每个请求换 IP，而是要把一组相关请求绑定到同一个 IP 上，形成一个"会话"。

实现方式：给每个会话生成一个 session_id，在 Redis 里维护 session_id 和 IP 的绑定关系：

Key:    proxy:session:{session_id}
Value:  ip:port
TTL:    600 (会话超时)

请求发出前，先查当前 session_id 有没有绑定的 IP。有就用绑定的，没有就从池子里借一个新的并绑定。

好处是支持有状态的采集流程。代价是被绑定的 IP 在会话期间不能被其他任务使用（等于被"锁"住了），如果会话很多、持续时间长，池子里可用的 IP 会明显减少。

架构设计：怎么在代码里同时支持这三层

一个工程上比较优雅的做法是把隔离策略抽象成接口，让调度层根据任务配置选择策略：

from abc import ABC, abstractmethod

class IsolationStrategy(ABC):
    @abstractmethod
    def acquire(self, pool, task_id, domain, session_id=None):
        """从池子里取一个满足隔离要求的IP"""
        pass

    @abstractmethod
    def release(self, pool, addr, task_id, domain, session_id=None, success=True):
        """归还IP，更新状态"""
        pass


class PoolIsolation(IsolationStrategy):
    """第一层：每个任务用独立的池子"""
    def acquire(self, pool, task_id, domain, session_id=None):
        pool_key = f"proxy:pool:{task_id}"
        return pool.borrow_from(pool_key)

    def release(self, pool, addr, task_id, domain, session_id=None, success=True):
        pool_key = f"proxy:pool:{task_id}"
        pool.return_to(pool_key, addr, success)


class DomainIsolation(IsolationStrategy):
    """第二层：共享池 + 域名冷却"""
    def acquire(self, pool, task_id, domain, session_id=None):
        candidates = pool.get_top_candidates("proxy:pool:shared", count=50)
        for addr in candidates:
            cooldown_key = f"proxy:used:{addr}:{domain}"
            if not pool.rdb.exists(cooldown_key):
                pool.rdb.setex(cooldown_key, 300, "1")
                return addr
        return None

    def release(self, pool, addr, task_id, domain, session_id=None, success=True):
        pool.return_to("proxy:pool:shared", addr, success)


class SessionIsolation(IsolationStrategy):
    """第三层：粘性会话"""
    def acquire(self, pool, task_id, domain, session_id=None):
        if not session_id:
            return DomainIsolation().acquire(pool, task_id, domain)
        session_key = f"proxy:session:{session_id}"
        bound = pool.rdb.get(session_key)
        if bound:
            return bound.decode()
        # 新会话，绑定一个IP
        addr = DomainIsolation().acquire(pool, task_id, domain)
        if addr:
            pool.rdb.setex(session_key, 600, addr)
        return addr

    def release(self, pool, addr, task_id, domain, session_id=None, success=True):
        if not success and session_id:
            # 会话中IP失败，解绑，让下次请求换新的
            pool.rdb.delete(f"proxy:session:{session_id}")
        pool.return_to("proxy:pool:shared", addr, success)

任务配置层面，你可以给每个任务指定隔离策略：

TASK_CONFIG = {
   
    "bid_monitor": {
   
        "strategy": "pool",        # 独立池
        "pool_min_size": 200,
    },
    "sentiment": {
   
        "strategy": "domain",      # 共享池 + 域名冷却
        "cooldown_seconds": 300,
    },
    "paginated_api": {
   
        "strategy": "session",     # 粘性会话
        "session_ttl": 600,
    },
}

这样不同任务可以按自己的需求选择隔离粒度，不用改调度层的核心代码。

IP 轮转策略：不只是"换一个"

会话隔离讲完了，回来讲 IP 轮转。轮转策略也不是只有"每个请求换一个"这一种，至少有三种常见模式：

逐请求轮转（Per-request rotation）。 每个请求用不同的 IP。适合目标站不做跨请求关联分析的场景，也就是目标站只看单次请求的 IP，不关心"这个 IP 之前有没有来过"。

定时轮转（Time-based rotation）。 每 N 秒换一次 IP，N 秒内的所有请求用同一个 IP。适合目标站有短期访问频率控制但不做长期封禁的场景——比如同一 IP 每分钟最多 10 个请求，你就设成每分钟换一次，每个 IP 在自己的 1 分钟窗口内发不超过 10 个请求。

按失败轮转（Failure-triggered rotation）。 不主动换 IP，只在当前 IP 发生失败（超时、403、429 等）时才换。适合目标站的访问频率控制不可预测、但一旦触发就立即生效的场景。好处是 IP 利用率最高——一个 IP 能用就一直用，不浪费。

这三种模式可以组合使用。比如"定时轮转 + 按失败轮转"：每 5 分钟换一次 IP，但如果 5 分钟内出现连续 3 次失败，立即换。

class RotationPolicy:
    def __init__(self, mode="per_request", interval=None, fail_threshold=3):
        self.mode = mode
        self.interval = interval  # 定时轮转的间隔（秒）
        self.fail_threshold = fail_threshold
        self._current_proxy = None
        self._assigned_at = 0
        self._consecutive_fails = 0

    def should_rotate(self, success=True):
        if self.mode == "per_request":
            return True

        if not success:
            self._consecutive_fails += 1
            if self._consecutive_fails >= self.fail_threshold:
                self._consecutive_fails = 0
                return True
        else:
            self._consecutive_fails = 0

        if self.mode == "time_based" and self.interval:
            if time.time() - self._assigned_at > self.interval:
                return True

        return False

一个容易踩的坑：Cookie 和 IP 的绑定关系

IP 轮转时很容易忽略一个问题：Cookie。

如果你的采集任务带 Cookie（比如需要保持登录态来访问公开 API），换 IP 但不换 Cookie，目标站可能会发现"同一个 Cookie 从不同 IP 发来"——这本身就是一个异常信号。

反过来，如果你换 IP 的同时也清 Cookie，但采集任务需要 Cookie 保持连续性（比如翻页），功能就断了。

所以：IP 轮转策略要和 Cookie 策略联动。

如果任务不需要 Cookie，每次换 IP 时清空 Cookie（或者干脆不发 Cookie）。

如果任务需要 Cookie，用粘性会话——同一个 session_id 绑定同一个 IP 和同一组 Cookie，会话结束后 IP 和 Cookie 一起释放。

在 Scrapy 里可以通过 request.meta['cookiejar'] 来隔离不同会话的 Cookie：

request.meta['cookiejar'] = session_id

不同的 cookiejar 值会被 Scrapy 的 CookieMiddleware 分开管理，不会互相污染。

怎么验证隔离有没有生效

隔离策略配好了，怎么验证它在生产环境里真的在工作？

检查 IP 复用率。 统计每个目标域名下，同一个 IP 在 N 分钟内被使用的次数。如果域名隔离生效了，同一个 IP 在同一个域名下的复用间隔应该大于你设的冷却时间。

# 简单的复用率检测脚本
def check_reuse(rdb, domain, window_minutes=5):
    pattern = f"proxy:used:*:{domain}"
    keys = rdb.keys(pattern)
    ttls = [rdb.ttl(k) for k in keys]
    active = sum(1 for t in ttls if t > 0)
    print(f"Domain {domain}: {active} IPs in cooldown (window={window_minutes}min)")

检查跨任务污染。 如果用的是池级隔离，检查两个任务的池子是否有 IP 交集。理论上应该完全不重叠（除非上游分配了重复 IP，那是上游的问题）。

检查会话完整性。 如果用了粘性会话，检查同一个 session_id 内的所有请求是否确实用了同一个 IP。可以在请求日志里加一个字段记录实际使用的 IP，事后聚合分析。

FAQ

Q：IP 不够用的时候，隔离策略会不会让情况更糟？

会。隔离本质上是用 IP 利用率换稳定性。如果你只有 200 个 IP，拆成 4 个池子每个只有 50 个，每个池子的调度空间都很小。所以隔离策略的前提是 IP 预算足够。IP 紧张的时候，建议先用域名级隔离（第二层），它的 IP 利用率比池级隔离高很多。

Q：会话隔离和业务隔离是一回事吗？

不完全一样。业务隔离是更大的概念：不同业务线的采集任务在代理资源、请求队列、调度策略上完全分开，互相不影响。会话隔离是业务隔离的一个子问题，专门解决"同一个 IP 的使用状态在不同任务之间不串扰"。你可以做了业务隔离但没做会话隔离（比如不同任务用不同队列，但共享代理池），也可以做了会话隔离但没做完整的业务隔离。

Q：粘性会话的 IP 挂了怎么办？

会话中 IP 失败时，解绑 session_id 和 IP 的映射（删 Redis Key），让下一个请求重新绑定一个新 IP。代价是会话状态可能断裂（如果目标站把 Cookie 和 IP 绑定了）。对于这种场景，IP 失败后 Cookie 也要清掉，整个会话重头来过。