企业级采集的架构设计要注意什么-核心信息和使用场景的重点在于把前置条件、操作顺序和容易误判的地方分清楚。

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

放在具体场景中，按"我每个请求都换了 IP，为什么还是被限制？企业级采集的架构设计要注意什么-核心信息和使用场景不能只看功能名称，更要看它在什么场景下能解决问题。"、会话隔离的三个层次、架构设计：怎么在代码里同时支持这三层等重点拆开说明，方便直接对照使用。

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

更直接地说，两个任务共用一个袋里 IP 池。从操作角度看，任务 B 采集公开舆情数据，每小时 2000 个请求。你有两个采集任务：任务 A 采集公开招投标信息，每小时 500 个请求；

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

放在具体场景中，它把池子里的高质量 IP 消耗得更快更隐蔽的情况是：任务 B 的请求量远大于任务 A导致任务 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:timestampTTL:300 (5 分钟冷却)

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

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

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

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

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

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

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

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

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

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

from abc import ABC, abstractmethodclass IsolationStrategy(ABC):@abstractmethoddef acquire(self, pool, task_id, domain, session_id=None):"""从池子里取一个满足隔离要求的IP"""pass@abstractmethoddef release(self, pool, addr, task_id, domain, session_id=None, success=True):"""归还IP，更新状态"""passclass 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 addrreturn Nonedef 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()# 新会话，绑定一个IPaddr = DomainIsolation().acquire(pool, task_id, domain)if addr:pool.rdb.setex(session_key, 600, addr)return addrdef 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。

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

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

放在具体场景中，这三种模式可以组合使用。以实际场景来看，"定时轮转 按失败轮转"：每 5 分钟换一次 IP，但如果 5 分钟内出现连续 3 次失败，立即换。

class RotationPolicy:def __init__(self, mode="per_request", interval=None, fail_threshold=3):self.mode = modeself.interval = interval# 定时轮转的间隔（秒）self.fail_threshold = fail_thresholdself._current_proxy = Noneself._assigned_at = 0self._consecutive_fails = 0def should_rotate(self, success=True):if self.mode == "per_request":return Trueif not success:self._consecutive_fails  = 1if self._consecutive_fails >= self.fail_threshold:self._consecutive_fails = 0return Trueelse:self._consecutive_fails = 0if self.mode == "time_based" and self.interval:if time.time() - self._assigned_at > self.interval:return Truereturn 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 不够用的时候，隔离策略会不会让情况更糟？

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

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

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

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

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