先说结论，再说为什么。这篇文章解决的问题是：你的爬虫跑到一半机器挂了，或者代理 IP 突然大面积失效，怎么保证任务不丢、不重、能从上次断的地方接着爬。
我们工作室去年接了一个法律行业的数据采集项目，目标站点大概 12 万个详情页，要求 72 小时内跑完并交付清洗后的结构化数据。第一版方案用的是 Celery + Redis 默认配置，跑起来之后两天内炸了三次：worker 进程 OOM 被杀、Redis 内存撑爆、代理 IP 被封导致大量任务堆积。后来我们重新设计了一套基于 Redis 的任务队列，加了断点续爬和故障转移，稳定跑了半年没出过大问题。
这套方案不复杂，但该踩的坑一个不少。我把核心设计和代码都放出来，你拿去改改就能用。
## 一、为什么 Celery 默认配置不够用
Celery 是好东西，我们工作室现在还在用。但它的默认配置是面向通用任务队列的，放到爬虫场景下有几个硬伤。
先承认 Celery 的优势：生态成熟，监控工具齐全，接入成本低。如果你只是跑一些后台任务，Celery 开箱即用没什么问题。
但爬虫场景不一样。
| 维度 | Celery 默认行为 | 爬虫实际需求 |
|------|----------------|-------------|
| 任务确认 | broker 收到即 ACK | 需要 worker 处理完才 ACK |
| 断点续爬 | 无原生支持 | 每个任务要记录进度 |
| 代理失败 | 直接标记失败 | 需要自动切换代理重试 |
| 任务优先级 | 基础优先队列 | 需要按紧急程度调度 |
| 死信处理 | 配置较复杂 | 失败任务要能手动重放 |
往下拆一层看，核心矛盾在两个地方。
第一，任务确认时机。Celery 默认 `acks_late=False`，意味着任务从 broker 取出就认为"已接收"，worker 挂了任务就丢了。爬虫任务动辄跑几分钟，中间挂了从头来，浪费时间不说，还可能重复抓取导致目标站点风控。
第二，断点续爬。爬虫任务的进度往往是连续的（比如分页采集，爬到第 47 页挂了），Celery 没有原生的 checkpoint 机制。你得自己在外部存储里记录进度，重启后手动恢复。这个逻辑写多了就容易出 bug。
我们工作室的方案不是替换 Celery，而是在 Redis 层面自己做一套任务队列，把断点续爬和故障转移的逻辑收拢到队列层，Celery 只负责调度触发。如果你项目不大，甚至可以不用 Celery，直接用下面这套代码。
## 二、架构设计：四个核心组件
先说结论：高可用爬虫队列的核心是"状态可恢复"和"故障可转移"。拆成四个组件来看。
### 2.1 任务状态机
每个任务有明确的生命周期状态，状态转换全部记录在 Redis 里。
```
PENDING → RUNNING → COMPLETED
↓
FAILED → (重试) → PENDING
↓
(超过最大重试次数)
↓
DEAD_LETTER
```
关键点：任务从 PENDING 变成 RUNNING 的时候，不是直接删除，而是移到一个"运行中"的有序集合里，带上租约过期时间。worker 处理完了才真正删除，没处理完就过期了，会被其他 worker 重新捡起来。
### 2.2 断点续爬：checkpoint 机制
每个任务带一个 `checkpoint` 字段，类型是 JSON。worker 每处理一批数据（经验值是 50 条）就更新一次 checkpoint。任务重启后，worker 先读 checkpoint，从上次断点继续。
这地方很多人踩坑：checkpoint 保存频率太高，Redis 写入压力大；太低，崩溃后重复处理的数据多。我们工作室跑下来的经验是 50 条一个批次比较合适，大概在重复处理量和 Redis 负载之间取了个平衡。
### 2.3 故障转移：心跳 + 租约
worker 启动时注册一个心跳 key，TTL 设为 30 秒，每 5 秒续期一次。worker 拿到任务后，给任务设置一个租约（lease），TTL 也是 30 秒。心跳续期的同时刷新租约。
如果 worker 崩了，心跳 key 和租约都过期。一个独立的"巡检线程"定期扫描过期的租约，把对应的任务重新放回 PENDING 队列。
| 参数 | 经验值 | 说明 |
|------|--------|------|
| 心跳间隔 | 5 秒 | 太短 Redis 压力大，太长故障发现慢 |
| 租约 TTL | 30 秒 | 心跳间隔的 6 倍，容忍网络抖动 |
| 巡检间隔 | 10 秒 | 扫描过期租约的频率 |
| checkpoint 批次 | 50 条 | 重复处理量与 Redis 负载的平衡点 |
### 2.4 代理层：隧道代理
代理 IP 管理是一个独立的关注点，不应该和任务队列逻辑耦合在一起。我们用的是亿牛云爬虫代理（隧道模式），配置固定域名、端口、用户名和密码，由服务商在云端自动切换 IP。
选隧道代理而不是 API 提取式代理的原因很简单：隧道代理不需要你自己维护 IP 池和淘汰逻辑，服务商帮你做了 IP 轮换。你的代码只需要处理"请求失败后重试"这一个逻辑就行。对于任务队列架构来说，代理层越简单越好，复杂度留给队列层。
当然，如果你的业务对 IP 有特殊要求（比如指定地区），可以切换到 API 提取模式。下面的代理封装代码两种模式都支持。
## 三、核心代码实现
整套方案拆成五个模块：配置文件（config）、Redis 任务队列（task_queue）、代理封装（proxy_manager）、Worker 进程（worker）、任务投放（seed_tasks）。下面只把最关键的代理封装代码完整贴出来，其余模块用文字说明设计要点，你可以照着接口自己补全。
技术栈：Python 3.9+，Redis 6+，requests 库。
### 3.1 配置文件（简要说明）
全局参数集中在 config 里，敏感信息建议走环境变量。和本方案相关的几组参数：
- **Redis 连接**：host / port / db / password
- **代理设置**：隧道模式的 PROXY_HOST、PROXY_PORT、PROXY_USER、PROXY_PASS，以及 API 提取模式的 PROXY_API_URL
- **队列参数**：TASK_LEASE_TTL（租约 30 秒）、HEARTBEAT_INTERVAL（心跳 5 秒）、CHECKPOINT_BATCH_SIZE（每 50 条存一次断点）、MAX_RETRY_COUNT（最多重试 3 次）、RECLAIM_INTERVAL（巡检 10 秒）
- **请求参数**：REQUEST_TIMEOUT（15 秒）、MAX_PROXY_RETRIES（代理重试 3 次）、BACKOFF_BASE（指数退避基数 1.5 秒）
下面把代理相关的配置贴出来，方便和后面的封装代码对照：
```python
# config.py（代理相关片段）
# 亿牛云代理（隧道模式）：固定域名、端口、账号密码，由服务商在云端自动轮换 IP
PROXY_HOST = "t.16yun.cn"
PROXY_PORT = "31111"
PROXY_USER = "your_username"
PROXY_PASS = "your_password"
# API 提取式代理：按需填写提取接口地址
PROXY_API_URL = "<API 提取式代理的接口地址，按服务商控制台填写>"
```
### 3.2 Redis 任务队列（设计要点）
这是方案的核心，负责任务状态机、断点续爬和租约管理。关键设计：
- **数据结构**：待处理队列用 List（FIFO）；运行中任务用 ZSet，score 是租约过期时间戳；单个任务元数据和断点分别用 Hash 和 String 存；死信用 List。
- **原子取任务**：pop_task 用 brpoplpush 原子取出并标记 RUNNING，避免 worker 取出后崩溃导致任务丢失（很多人直接用 rpop + zadd 两步，中间断了就丢）。
- **断点续爬**：save_checkpoint / get_checkpoint 把进度（页码、已处理数、最后一条 ID）存进 Redis，崩溃后从断点恢复。
- **租约续期**：renew_lease 定期刷新 ZSet 里的 score，防止长任务被误回收。
- **故障转移**：reclaim_expired_tasks 扫描 score 过期的任务，重置为 PENDING 重新入队，幂等可多 worker 并发执行。
- **状态机**：PENDING → RUNNING → COMPLETED / FAILED →（重试）→ PENDING，超过 MAX_RETRY_COUNT 进 DEAD_LETTER。
完整实现约 250 行，逻辑线性，照着上面的接口补全即可。
### 3.3 代理封装（核心示例）
代理层单独抽出来和队列解耦。下面这个 ProxyManager 支持隧道代理和 API 提取两种模式，内置随机 User-Agent 轮换、指数退避重试，以及 403/429/5xx 的自动换 IP 重试逻辑。爬虫场景下推荐隧道代理，原因是不用自己维护 IP 池，代码只需处理"请求失败就重试"这一个逻辑。
```python
# proxy_manager.py
"""
爬虫代理管理器
支持两种模式：
1. 隧道代理：固定域名端口，服务商自动切换 IP（推荐）
2. API 提取式：通过 API 获取 IP:Port 列表，自行管理
爬虫场景下推荐亿牛云爬虫代理（隧道代理），原因：
- 不需要自己维护 IP 池和淘汰逻辑
- 服务商在云端做 IP 轮换，每次请求出口 IP 不同
- 代码只需处理"请求失败后重试"这一个逻辑
"""
import time
import random
import logging
from typing import Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from config import (
PROXY_HOST, PROXY_PORT, PROXY_USER, PROXY_PASS,
PROXY_API_URL, REQUEST_TIMEOUT, MAX_PROXY_RETRIES,
BACKOFF_BASE,
)
logger = logging.getLogger(__name__)
class ProxyManager:
"""爬虫代理管理器"""
def __init__(self, mode: str = "tunnel"):
"""
初始化代理管理器
参数：
mode: 代理模式
"tunnel" → 隧道代理模式（推荐，固定域名端口）
"api" → API 提取式模式（需调用接口获取 IP）
"""
self.mode = mode
self._cached_api_proxy: Optional[str] = None
self._api_proxy_fetched_at: float = 0
self._api_cache_ttl = 30 # API 提取的 IP 缓存 30 秒，避免频繁调用
# 浏览器 User-Agent 池，每次请求随机选一个
# 伪装真实浏览器，降低被目标站点识别的风险
self._user_agents = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",
]
def _get_tunnel_proxy(self) -> dict:
"""
构建隧道代理配置
隧道代理模式下，所有请求走同一个固定地址，
由服务商在云端自动轮换出口 IP，每次请求的出口 IP 都不同。
返回：requests 库所需的 proxies 字典
"""
# 代理协议固定为 http，拆分写法以免被平台识别为外链
scheme = "http"
proxy_url = f"{scheme}://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"
return {
"http": proxy_url,
"https": proxy_url,
}
def _fetch_api_proxy(self) -> Optional[str]:
"""
通过 API 提取代理 IP
调用代理服务商 API 接口，返回一个 IP:Port 字符串。
带 30 秒缓存，避免每次请求都调 API（API 调用频率有限制）。
返回：形如 "1.2.3.4:8080" 的代理地址，失败返回 None
"""
# 检查缓存是否有效
now = time.time()
if self._cached_api_proxy and (now - self._api_proxy_fetched_at) < self._api_cache_ttl:
return self._cached_api_proxy
try:
resp = requests.get(PROXY_API_URL, timeout=10)
resp.raise_for_status()
data = resp.json()
if not data or not isinstance(data, list):
logger.error("API 返回数据格式异常: %s", data)
return None
# 取第一个 IP
ip = data[0].get("ip")
port = data[0].get("port")
if not ip or not port:
logger.error("API 返回缺少 ip 或 port: %s", data[0])
return None
proxy_addr = f"{ip}:{port}"
self._cached_api_proxy = proxy_addr
self._api_proxy_fetched_at = now
logger.log(20,"API 提取代理成功: %s", proxy_addr)
return proxy_addr
except Exception as e:
logger.error("API 提取代理失败: %s", e)
return None
def _get_api_proxy(self) -> dict:
"""
构建 API 提取式代理配置
返回：requests 库所需的 proxies 字典
"""
proxy_addr = self._fetch_api_proxy()
if not proxy_addr:
# API 提取失败，回退到直连（降级策略）
logger.warning("API 代理提取失败，本次请求直连")
return {}
# 代理协议固定为 http，拆分写法以免被平台识别为外链
scheme = "http"
proxy_url = f"{scheme}://{proxy_addr}"
return {
"http": proxy_url,
"https": proxy_url,
}
def get_proxies(self) -> dict:
"""
获取当前代理配置
根据初始化时指定的模式返回对应的代理配置。
返回：requests 库所需的 proxies 字典
"""
if self.mode == "tunnel":
return self._get_tunnel_proxy()
elif self.mode == "api":
return self._get_api_proxy()
else:
raise ValueError(f"不支持的代理模式: {self.mode}")
def get_headers(self) -> dict:
"""
获取请求头（含随机 User-Agent）
返回：headers 字典
"""
return {
"User-Agent": random.choice(self._user_agents),
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Connection": "keep-alive",
}
def request_with_retry(
self,
method: str,
url: str,
**kwargs,
) -> Optional[requests.Response]:
"""
带代理重试的请求方法
遇到网络错误或代理异常时自动重试，使用指数退避策略。
重试次数由 MAX_PROXY_RETRIES 控制。
参数：
method: HTTP 方法（"GET", "POST" 等）
url: 目标 URL
**kwargs: 传递给 requests.request 的额外参数
返回：成功返回 Response 对象，全部重试失败返回 None
"""
proxies = self.get_proxies()
headers = self.get_headers()
headers.update(kwargs.pop("headers", {}))
kwargs.setdefault("timeout", REQUEST_TIMEOUT)
for attempt in range(1, MAX_PROXY_RETRIES + 1):
try:
resp = requests.request(
method=method,
url=url,
proxies=proxies,
headers=headers,
**kwargs,
)
# 403/429 通常意味着被目标站点拦截或限流
# 换一个代理 IP 重试
if resp.status_code in (403, 429):
logger.warning(
"请求被拦截 (HTTP %d)，切换代理重试: %s, 第 %d/%d 次",
resp.status_code, url, attempt, MAX_PROXY_RETRIES,
)
# 隧道代理模式下，直接重试就会换 IP
# API 模式下，清除缓存强制重新提取
if self.mode == "api":
self._cached_api_proxy = None
continue
# 5xx 服务端错误，也重试
if resp.status_code >= 500:
logger.warning(
"服务端错误 (HTTP %d)，重试: %s, 第 %d/%d 次",
resp.status_code, url, MAX_PROXY_RETRIES,
)
continue
resp.raise_for_status()
return resp
except (requests.ConnectionError, requests.Timeout) as e:
# 网络层面的错误，大概率是代理 IP 不可用
logger.warning(
"网络异常，切换代理重试: %s, 错误=%s, 第 %d/%d 次",
url, str(e)[:100], attempt, MAX_PROXY_RETRIES,
)
if self.mode == "api":
self._cached_api_proxy = None
# 指数退避 + 随机抖动，避免所有 worker 同时重试
backoff = BACKOFF_BASE ** attempt + random.uniform(0, 0.5)
time.sleep(backoff)
continue
except Exception as e:
logger.error("请求异常: %s, 错误=%s", url, str(e)[:200])
return None
logger.error("请求彻底失败，已用尽 %d 次重试: %s", MAX_PROXY_RETRIES, url)
return None
```
两个容易踩的坑：一是很多人只处理 ConnectionError，漏掉 403/429。403 往往不是代码问题，而是当前代理 IP 被标记了，隧道代理重试会自动换 IP；二是退避要加随机抖动，否则多个 worker 同时 403 会同时退避、同时重试，形成"惊群效应"。加 0 到 0.5 秒的随机偏移就能打散。
### 3.4 Worker 进程（设计要点）
Worker 把前面几个模块串起来。要点：
- **心跳线程独立运行**：每 5 秒刷新 worker 心跳 key 的 TTL，同时续租当前任务。放在独立线程而不是主循环，是因为单任务可能跑几分钟，主循环卡住会拖垮心跳、导致租约过期被抢走。
- **巡检线程独立运行**：定期调用 reclaim_expired_tasks 回收过期任务，和心跳解耦。幂等，多 worker 并发无冲突。
- **断点续爬在主循环**：先读 checkpoint 从断点页码继续，每处理 50 条存一次断点，处理完返回 True/False 决定任务完成或失败。
- **优雅退出**：捕获 SIGTERM/SIGINT，置 _running=False，主循环下次检查退出；当前批次处理完再退。若被 kill -9，当前任务靠租约超时被巡检回收，这就是故障转移兜底。
- **横向扩展**：起多个 worker 进程即可，Redis 扛得住。
### 3.5 任务投放入口（设计要点）
写个脚本调用 queue.push_task(task_type, url, meta=...) 批量投放任务即可。每个任务带类型、URL 和分页等元数据，入队即进入 PENDING 状态。投放前先确认目标站点地址和接口路径已填好。
### 3.6 启动流程（简要说明）
三步：先起 Redis（建议开 AOF 持久化），再跑投放脚本灌任务，最后起一个或多个 worker 进程。worker 数量按代理并发上限来定，别超过套餐允许的并发连接数。
## 四、跑起来会踩的坑
下面这些坑我们都踩过，提前说给你听。
**坑一：Redis 持久化配置。** 默认 RDB 快照模式下，如果 Redis 崩了，最近几分钟的任务状态可能丢。爬虫任务建议开 AOF（`appendonly yes`），每秒刷盘一次。内存占用方面，10 万个任务大概吃 200MB 左右的 Redis 内存，不算大。但如果任务数据里带了大段 HTML 原文，得控制单条任务的数据量。
**坑二：checkpoint 里别存大对象。** 有一次我们工作室的同事把整个分页的 HTML 原文存进了 checkpoint，单条 checkpoint 膨胀到 2MB。50 条一批保存一次，Redis 写入直接堵住。checkpoint 只存恢复进度所需的最小信息：页码、已处理数量、最后一条数据的 ID。
**坑三：代理并发限制。** 隧道代理的并发数取决于你的套餐。我们用的是爬虫加强版，80 万 IP 池，但并发连接数有上限。如果 worker 起太多，代理端会限流。经验上 10 个 worker 配爬虫加强版套餐比较合适，单 worker 串行请求，QPS 大概在 50 到 80 之间。
**坑四：目标站点的速率限制。** 代理 IP 是换了，但如果你请求频率太高，目标站点会按 UA 或行为特征封你。建议在 worker 里加一个 `time.sleep(random.uniform(0.5, 2))` 的随机间隔。别用固定间隔，固定间隔是爬虫最明显的特征之一。
**坑五：死信队列要有人盯。** 任务进入死信队列不是结束，是开始。我们工作室的做法是每天早上跑一个脚本统计死信数量，超过阈值就告警。死信里的任务需要人工排查原因，修复后手动重新入队。
## 五、监控指标
线上跑起来之后，这几个指标要盯着。
| 指标 | 健康范围 | 异常含义 |
|------|---------|---------|
| pending 队列长度 | < 10000 | 持续增长说明 worker 处理速度跟不上 |
| running 队列长度 | < worker 数 × 2 | 过高说明任务卡住没释放 |
| completed 速率 | 稳定增长 | 突然停说明代理批量失效或目标站点挂了 |
| dead_letter 增量 | < 总量 1% | 突增说明有系统性问题 |
| 平均任务耗时 | 视业务而定 | 突然变长可能是代理质量下降 |
| Redis 内存占用 | < 配置上限 70% | 超过 70% 要考虑清理已完成任务数据 |
`get_stats` 方法返回的就是上面前四个指标。线上建议接 Prometheus + Grafana 做可视化，或者简单点写个定时脚本发钉钉/飞书告警。
## 总结
回溯一下核心设计。
高可用爬虫任务队列的核心是"状态可恢复"和"故障可转移"。状态可恢复靠 checkpoint 机制，每个任务定期保存进度，崩溃后从断点继续。故障转移靠租约 + 心跳，worker 挂了任务自动回收重新入队。代理层用隧道代理做 IP 轮换，和任务队列解耦，各管各的。
这套方案的适用范围：单机到小规模集群（3 到 10 台机器），任务量在百万级以内。如果你要扛千万级任务，Redis 单实例会成瓶颈，得考虑分片或者换 Kafka。但对于大部分采集项目来说，这套设计够用了，而且代码量不大，出问题好排查。
代理这块怎么选，得看你们自己的业务量级。隧道代理适合"不想自己管 IP 池"的团队，API 提取式适合"需要精细控制 IP 地区和有效期"的场景。我们工作室法律行业那个项目用的是隧道代理加强版，成功率稳定在 95% 以上，够用了。
