大多数爬虫代码库都会生出一个代理管理器,不管是否有人特意设计过它。它通常先是一个构造代理 URL 的辅助函数,然后有人加了重试逻辑,接着为某个需要粘性会话的目标加了特殊处理,再加一个计数器来防止对一个持续失败的地址反复发起请求。十八个月后,这些逻辑分散在四个模块里,没人说得清当同一个会话中的请求连续失败两次时会发生什么。
值得刻意地把它构建出来,因为其中涉及的决策确实是共通的:这个请求用哪个出口、这个出口是否健康、失败时该怎么办,以及如何控制节奏。下面是这个组件的设计,以及它需要保存的状态。
管理器负责什么
把边界划得紧一些。代理管理器决定请求如何离开你的基础设施,以及失败时该怎么处理。它不解析页面,不知道商品列表长什么样,也不决定要访问哪些 URL。让它对业务逻辑一无所知,正是它能被每个任务共用的原因。
这样一来就剩下四项职责:按策略为每个请求选定一个出口,跟踪它所分发对象的健康状况,在出问题时执行失败处理,以及强制执行节奏控制,防止调用方各自压垮某个目标。其余的一切都不属于它。
选择:策略是按任务而非全局的
管理器首先需要的是调用方想要的出口类型的概念,而不是单一的全局模式。实际中有三种。
轮换(Rotating),每个请求获得一个全新的出口。这是批量采集的默认方式,通过省略会话标识来表达,也是轮换机制设计上应有的用法。
粘性(Sticky),调用方在一个序列中持有同一个出口,管理器必须在该序列的生命周期内返回同一个会话标识,然后再释放它。相关权衡见粘性代理与轮换代理对比。
地理定向(Geo-pinned),出口必须位于特定国家或城市,这对多市场工作来说与前两者是正交的:一个任务可以是”在德国轮换”或”在芝加哥粘性”。
把这些建模为请求范围内的租约(lease),而不是全局设置,让调用方按需请求,由管理器来满足:
from dataclasses import dataclass
from typing import Optional
import itertools, time
@dataclass
class Lease:
country: str
session: Optional[str] = None # None means rotate per request
ttl: Optional[int] = None # only meaningful with a session
def username(self, customer="USERNAME"):
parts = [f"customer-{customer}", f"country-{self.country}"]
if self.session:
parts.append(f"sid-{self.session}")
if self.ttl:
parts.append(f"ttl-{self.ttl}")
return "-".join(parts)
def proxies(self, password="PASSWORD", host="p.shifter.io:443"):
url = f"http://{self.username()}:{password}@{host}"
return {"http": url, "https": url}管理器的工作就是产出一个 Lease,交给调用方,并观察它接下来的情况。
健康状况:跟踪会话,而非地址
这正是大多数自研管理器出错的地方。在共享网关上,你不会持有一份要标记好坏的 IP 地址列表,因为你从未选择过它们,也无法保留它们。你能跟踪的是会话的健康状况,以及路由(即国家与目标的组合)的整体健康状况。
因此要维护两样东西。对每个活跃的粘性会话,保存一份关于连续失败次数和验证挑战响应的小记录,这样一个明显已经变差的会话可以被淘汰并换成新的标识。对每条路由,维护一个滚动成功率,它能告诉你是整个国家-目标组合都变差了,还是仅仅是某个会话运气不好。
这个体系里的健康检查不是周期性的 ping。ping 一个出口只能告诉你它能到达某个测试端点,而这不是关键问题;关键问题是它能否到达你的目标并获取真实内容。因此要用被动健康检查:每一次真实请求就是健康检查本身,其经过验证的结果会更新记录。主动探测只在针对每个目标做一个小规模、低成本的哨兵检测时才值得,这有助于区分”这个目标对所有人都挂了”和”我们通往它的路由已经变差了”。
关键的是,要依据经过验证的响应而非状态码来判断健康状况。一个返回 200 的验证挑战页面,从健康检测的角度看是一次失败,而如果管理器把它算作成功,就会持续复用一个目标早已判定过的会话。这与检测被封锁或伪造内容中所讲的验证原则是一致的。
class RouteHealth:
def __init__(self, window=50):
self.window, self.results = window, []
def record(self, ok: bool):
self.results.append(ok)
if len(self.results) > self.window:
self.results.pop(0)
@property
def success_rate(self):
return sum(self.results) / len(self.results) if self.results else 1.0
@property
def degraded(self):
return len(self.results) >= 10 and self.success_rate < 0.7失败处理:先分类,再行动
管理器拥有对失败含义的判定权,这正是防止这套逻辑在每个任务中被重新发明的原因。三个类别足以覆盖。
速率信号意味着放慢速度:429 或 Retry-After。正确的响应是等待,而具体来说不要更换到一个新出口,以便同样的节奏能够继续下去,因为正是这种更换会把一个被限速的路由变成一个被封杀的路由。
身份信号意味着这个出口在这个目标上已经用完了:一个封锁页面、持续的 403,或同一会话上反复出现的验证挑战。正确的响应是淘汰这个会话,取一个新的,然后继续,这正是故障切换模式。
终止性错误意味着停止:404、格式错误的 URL,或 407(一种认证问题,每次重试都会同样出错,应该响亮地失败而不是循环重试),详见修复 407 与凭据错误。
其余的都是暂时性的,应使用带抖动的指数退避,并以尝试次数上限为界,参见重试与退避。这里重要的架构要点是,重试逻辑存在于管理器内部,这样每个调用方都会继承同样的行为,而且管理器可以在所有调用方之间强制执行一个统一的重试预算,而不是让每个任务各自独立地对同一个陷入困境的目标进行重试。
顺手为每条路由加上一个断路器。当一条路由的健康状况跌破某个阈值时,停止向它发送请求,进入一个冷却期,然后允许少量请求通过来测试是否恢复。这既能保护目标不被你的集中冲击所压垮,也能保护你的账号不因针对一个对谁都不响应的站点而不断累积失败记录。
节奏控制也属于这里
由于管理器能看到每一个外发请求,它是强制执行按目标限速与并发上限的自然位置,这正是限速与请求节流中所描述的机制。把它放在这里有一个特定的好处:重试会自动遵守限速器,因为它们和其他任何请求走的是同一条路径。绕过节奏控制的重试,正是一个陷入困境的目标变成被封锁目标的方式。
整合起来
整个接口面很小,这正是要点所在:
class ProxyManager:
def __init__(self, limiter_factory, health_factory):
self.limiters = {} # host -> TargetLimiter
self.health = {} # (country, host) -> RouteHealth
self.limiter_factory, self.health_factory = limiter_factory, health_factory
def get(self, url, host, country, session=None, attempts=4):
route = self.health.setdefault((country, host), self.health_factory())
limiter = self.limiters.setdefault(host, self.limiter_factory(host))
sid = session
for attempt in range(attempts):
if route.degraded:
raise RouteUnavailable(country, host) # circuit open
limiter.acquire() # pacing, retries included
lease = Lease(country=country, session=sid, ttl=600 if sid else None)
outcome = self._send(url, lease) # returns (klass, response)
route.record(outcome.klass == "ok")
if outcome.klass == "ok":
return outcome.response
if outcome.klass == "terminal":
raise TerminalError(url, outcome.response)
if outcome.klass == "identity" and sid:
sid = new_session_id() # retire, do not reuse
if outcome.klass == "rate":
time.sleep(outcome.retry_after or backoff(attempt))
else:
time.sleep(backoff(attempt)) # transient
raise Exhausted(url)两点设计说明。管理器返回响应对象,并抛出带类型的错误,而不是返回 None,这样调用方就不会把失败无声地当作空数据处理。而且它从不吞掉终止性错误,因为一个认证或配置问题应该终止一个任务,而不是被重试到悄无声息地消失。
运维方面的考量
让它可观测。 管理器能看到每一个请求,所以每条路由的成功率、重试比例和每请求字节数都应该由它来输出,这正是代理 KPI中的指标所依据的数据,也是流水线监控所消费的数据。
跨进程共享状态。 按进程配置管理器会让你的实际速率成倍放大(乘以工作进程数量),并使健康跟踪碎片化。在分布式部署中,限速器、断路器和路由健康状态都应放在共享存储中,这与用 Kubernetes 扩展中的考量是一样的。
让凭据远离它。 管理器负责拼装用户名;它应该从密钥存储中读取密码,而不是把密码放在配置里,轮换密码应该是一个部署步骤,而不是代码改动。
不要过度抽象。 不要为假想中的供应商去构建插件架构。一个网关,只要表达清晰,就比一个覆盖你根本不使用的厂商的抽象层更容易理解。
结论
代理管理器是决定请求如何离开你的系统、以及失败时该怎么办的组件,把这些决策集中起来,正是防止同一套逻辑在每个任务中被不一致地重复实现的方式。给调用方一个请求范围内的租约,让轮换、粘性和地理位置都是按任务而非全局设定的。在会话和路由层面跟踪健康状况,而不是假装在管理单个地址,并依据经过验证的响应来判断健康状况,这样一个验证挑战页面才会被算作它本该是的失败。把失败分类的职责放在一个地方:遇到速率信号就等待,遇到身份信号就淘汰会话,遇到终止性错误就响亮地失败,其余一切都带抖动地退避,并为每条路由配一个断路器。把节奏控制放在同一条路径上,这样重试就无法绕过它。然后从中输出指标,因为它是唯一能看到全局的组件。
在这一切之下是住宅代理网络:一个网关,国家、城市、会话和 TTL 全部通过用户名来表达,这正是让这样一个管理器成为一小段代码而不是一项集成工程的原因,再加上按 GB 计价,它带来的效率会直接体现在账单上。