Python Token 爬虫实战：如何优雅地搞定分布式采集与反爬策略
一、为什么「Token 爬虫」是块硬骨头普通的游客态爬虫（直接抓公开页面）已经很少见了。如今稍微有点价值的站点，几乎都走 前后端分离 + Token 鉴权：请求不带 Authorization: Bearer 直接 401；Token 有有效期（30min / 2h / 24h 不等），过期就采集中断；接口往往带签名校验或时间戳防重放，简单改个参数就被识破；同一 Token 高频请求会被标记为「机器行为」，连带封 IP。更麻烦的是，当你想用分布式把采集速度拉满时，这些麻烦会成倍放大：多个 Worker 各自刷新 Token 会互相挤掉、分散的 IP 各自为战、限流策略一旦不统一就被批量封。本文给你一套工程化、可落地的打法：Token 统一管理 + 分布式任务队列 + 反爬对抗三板斧 + 高质量代理 IP 池。代码基于 httpx(async) + Redis + pydantic，复制即可跑。二、Token 机制与刷新策略：别让 Worker 各自为政分布式爬虫最致命的坑，就是每个 Worker 自己管 Token。结果就是 A 刷新了，B 还拿着旧 Token 请求 → 401 重试 → 又触发刷新 → 服务端以为被盗号，直接封号。正确姿势：集中式 Token 管理，所有 Worker 从同一个「Token 服务」取用。 ``` # token_manager.py from __future__ import annotations import asyncio import time from typing import Callable import httpx class TokenManager: """集中式 Token 管理器：单点刷新，多 Worker 共享。 Args: auth_func: 异步函数，返回新 token 字符串（负责登录/换票）。 ttl: token 有效期（秒），默认 1800。 refresh_ahead: 提前多少秒刷新，避免临界点失效。默认 60。 """ def __init__( self, auth_func: Callable[[], str], ttl: int = 1800, refresh_ahead: int = 60, ) -> None: self._auth_func = auth_func self._ttl = ttl self._refresh_ahead = refresh_ahead self._token: str | None = None self._expire_at: float = 0.0 self._lock = asyncio.Lock() async def get_token(self) -> str: """获取有效 token，临近过期时加锁刷新，避免惊群效应。""" async with self._lock: # 双重检查：拿到锁后再判断一次，防止并发请求重复刷新 if self._token is None or time.time() >= (self._expire_at - self._refresh_ahead): self._token = await self._auth_func() self._expire_at = time.time() + self._ttl return self._token # 使用示例 async def login_and_get_token() -> str: """真实项目里这里放账号密码登录 / OAuth 换票逻辑。""" async with httpx.AsyncClient() as client: resp = await client.post( "https://api.example.com/auth/login", json={"username": "your_user", "password": "your_pass"}, ) resp.raise_for_status() return resp.json()["token"] token_manager = TokenManager(login_and_get_token) ``` 为什么这样做：asyncio.Lock + 双重检查，保证「同一时刻只有一个协程去刷新」，其余协程复用结果。分布式多进程场景下，则把 Token 缓存到 Redis（见第四节），用 SET NX 抢刷新权。三、反爬三板斧：UA、频率、IP服务端识别爬虫，无非三招。我们挨个拆解应对： |反爬手段|识别逻辑|优雅应对| | --- | --- | --- | |UA 指纹|固定/缺失 User-Agent、缺失常见请求头|维护 UA 池 + 补全 Accept/Referer 等头| |请求频率|单位时间请求数超阈值、无随机间隔|令牌桶限速 + 随机抖动（jitter）| |IP 信誉|同一 IP 高频访问、数据中心 IP 段|高质量代理 IP 池（关键，见第五节）| 四、分布式采集架构：Redis 当任务总线核心思路：任务入队（Redis List/Stream）→ 多个 Worker 抢任务 → 结果落库。Worker 之间只通过 Redis 通信，彼此无状态，想横向扩到 100 台机器只需加进程。 ``` # distributed_crawler.py from __future__ import annotations import asyncio import json import random from typing import Any import httpx import redis.asyncio as aioredis from token_manager import token_manager class DistributedCrawler: """基于 Redis 任务队列的分布式爬虫骨架。 Args: redis_url: Redis 连接串，Token 与任务都走它。 proxy_getter: 返回单个代理 URL 的异步函数（对接亿牛云等）。 """ def __init__(self, redis_url: str, proxy_getter: Any) -> None: self._redis = aioredis.from_url(redis_url, decode_responses=True) self._proxy_getter = proxy_getter self._ua_pool = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko)", ] async def push_task(self, url: str) -> None: """生产端：把待采 URL 推入队列。""" await self._redis.lpush("crawl:tasks", url) async def _build_headers(self, token: str) -> dict[str, str]: """随机 UA + 标准头，降低指纹命中率。""" return { "Authorization": f"Bearer {token}", "User-Agent": random.choice(self._ua_pool), "Accept": "application/json", "Referer": "https://www.example.com/", } async def worker(self, worker_id: int) -> None: """消费端：循环取任务、带代理与 Token 请求、结果写回。""" async with httpx.AsyncClient(timeout=15) as client: while True: # 阻塞式取任务，0 表示一直等 _, url = await self._redis.brpop("crawl:tasks", timeout=0) token = await token_manager.get_token() proxy = await self._proxy_getter() # 每次请求换出口 IP try: resp = await client.get( url, headers=await self._build_headers(token), proxies={"http://": proxy, "https://": proxy}, ) resp.raise_for_status() # 结果落库（这里简化为回写另一个队列） await self._redis.lpush("crawl:results", resp.text[:500]) print(f"[Worker-{worker_id}] OK {url} via {proxy}") except httpx.HTTPStatusError as e: # 401 说明 token 失效，交给上层重试；其余按失败处理 print(f"[Worker-{worker_id}] FAIL {url}: {e.response.status_code}") except Exception as e: # noqa: BLE001 print(f"[Worker-{worker_id}] ERROR {url}: {e}") # 限速：随机抖动，模拟人类节奏 await asyncio.sleep(random.uniform(0.5, 2.0)) # 亿牛云代理接入示例（Token/账密认证，动态 IP 池） async def yiniu_proxy_getter() -> str: """从亿牛云 API 动态提取一个代理，返回 httpx 可用的代理 URL。 亿牛云支持「订单号+密码」或「子授权」方式鉴权， 提取接口返回的 IP:PORT 直接用即可，无需在每台机器硬编码。 """ async with httpx.AsyncClient() as client: # 示例地址，实际以亿牛云后台提供的提取 API 为准 r = await client.get("https://proxy.16yun.cn/api/get_proxy") data = r.json() return f"http://{data['ip']}:{data['port']}" async def main() -> None: crawler = DistributedCrawler( redis_url="redis://localhost:6379/0", proxy_getter=yiniu_proxy_getter, ) # 生产端灌任务 for i in range(100): await crawler.push_task(f"https://api.example.com/items/{i}") # 启动 4 个 Worker 并发采集 await asyncio.gather(*(crawler.worker(i) for i in range(4))) if __name__ == "__main__": asyncio.run(main()) ``` 依赖：pip install httpx redis pydantic五、代理 IP 池实战：把亿牛云接进分布式前面三板斧里，IP 是最关键也最难自建的一环。把代理服务做成统一 getter 函数，Worker 无感知背后是哪家的池子——哪天换服务商只改一处。接入亿牛云时的三个工程要点：鉴权方式统一：亿牛云支持账密 / 子授权 Token 两种方式。分布式场景下推荐子授权 Token——每个 Worker 用不同子账号，万一某个出口 IP 被临时风控，不会连累整池。失败熔断：代理偶尔抽风是正常的。getter 里加重试 + 超时，连续 N 次失败则短暂降级（放慢采集），别让异常把 Worker 打挂。IP 复用策略：对「同一站点」适当复用出口 IP（避免每个请求都换 IP 反而触发「IP 跳跃」风控），对「不同站点」分散出口——这套策略在 getter 里用 LRU 缓存即可实现。 ``` # proxy_pool.py —— 带熔断与缓存的代理 getter 封装 from __future__ import annotations import asyncio import time from collections import OrderedDict from typing import Callable import httpx class ProxyPool: """代理池封装：熔断 + LRU 复用，屏蔽底层服务商差异。 Args: extractor: 异步函数，从服务商（如亿牛云）提取一个 ip:port。 cache_size: 同站点复用 IP 的缓存条数。 max_fail: 连续失败多少次触发熔断。 """ def __init__( self, extractor: Callable[[], str], cache_size: int = 16, max_fail: int = 3, ) -> None: self._extractor = extractor self._cache: "OrderedDict[str, str]" = OrderedDict() self._fail_count = 0 self._circuit_open_until = 0.0 async def get(self, site_key: str) -> str: """按站点取代理：同站点优先复用，否则拉新。""" now = time.time() if now = 3: self._circuit_open_until = now + 30 # 熔断 30s raise self._cache[site_key] = proxy if len(self._cache) > 16: self._cache.popitem(last=False) return proxy ``` 六、限速与抖动：像人一样请求分布式不是「无脑并发」。哪怕有代理池，单 IP 的请求节奏依然要像人：令牌桶限速：全局 QPS 上限，例如每 IP 每秒 ≤ 2 次；随机抖动：sleep(uniform(0.5, 2.0))，消除「机器般精确」的间隔；指数退避：遇到 429 / 503，先等 1s，再 2s、4s…… 别硬刚。 ``` # rate_limiter.py import asyncio import random import time class TokenBucket: """简单令牌桶：保证全局速率可控。""" def __init__(self, rate: float, capacity: float) -> None: self._rate = rate # 每秒补充令牌数 self._capacity = capacity # 桶容量 self._tokens = capacity self._last = time.monotonic() async def acquire(self) -> None: while True: now = time.monotonic() self._tokens = min(self._capacity, self._tokens + (now - self._last) * self._rate) self._last = now if self._tokens >= 1: self._tokens -= 1 # 加一点随机抖动，避免请求间隔过于规整 await asyncio.sleep(random.uniform(0, 0.3)) return await asyncio.sleep(0.1) ``` 七、工程化最佳实践（上线前 checklist）配置外置：Redis 地址、代理鉴权、账号密码全走环境变量 / pydantic-settings，绝不硬编码进仓库。可观测：每次请求记录 url / proxy / status / cost_ms 到日志，封禁率异常立刻告警。幂等消费：任务队列用 Redis Stream + 消费者组，Worker 崩溃重启不丢任务、不重复采。优雅退出：SIGTERM 时停止拉新任务、处理完在途请求再退出，别直接 kill 导致 Token 状态混乱。代理质量监控：定期统计各出口 IP 的成功率，低质量 IP 自动剔除——这也是选亿牛云这类带质量保障 SLA 服务商的价值所在。八、总结Token 爬虫的「优雅」，本质是把混乱收敛成几个清晰的职责边界：Token 集中管理，杜绝刷新内卷；任务用 Redis 解耦，Worker 无状态横向扩展；反爬三板斧（UA / 频率 / IP）层层设防，其中 IP 这一层交给专业代理服务限速 + 抖动 + 退避，让请求节奏像人。按这套骨架搭起来，你就能从「单机裸爬三天两头挂」，进化到「百 Worker 稳定运行、封禁率可控」的生产级采集系统。代码都已给出，缺的只是把 example.com 换成你的目标站、把代理 getter 换成你的服务商的提取 API。
