一、为什么「Token 爬虫」是块硬骨头
普通的游客态爬虫（直接抓公开页面）已经很少见了。如今稍微有点价值的站点，几乎都走 前后端分离 + Token 鉴权：
- 请求不带
**Authorization: Bearer <token>**直接 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 < self._circuit_open_until:
raise RuntimeError("代理池熔断中，请稍后重试")
if site_key in self._cache:
self._cache.move_to_end(site_key)
return self._cache[site_key]
try:
proxy = await self._extractor()
self._fail_count = 0
except Exception: # noqa: BLE001
self._fail_count += 1
if self._fail_count >= 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。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/Z_suger7/article/details/168026184
