如何设计一套高可用的爬虫任务队列，保证断点续爬与故障转移？

先说结论，再说为什么。这篇文章解决的问题是：你的爬虫跑到一半机器挂了，或者代理 IP 突然大面积失效，怎么保证任务不丢、不重、能从上次断的地方接着爬。

我们工作室去年接了一个法律行业的数据采集项目，目标站点大概 12 万个详情页，要求 72 小时内跑完并交付清洗后的结构化数据。第一版方案用的是 Celery + Redis 默认配置，跑起来之后两天内炸了三次：worker 进程 OOM 被杀、Redis 内存撑爆、代理 IP 被封导致大量任务堆积。后来我们重新设计了一套基于 Redis 的任务队列，加了断点续爬和故障转移，稳定跑了半年没出过大问题。

这套方案不复杂，但该踩的坑一个不少。我把核心设计和代码都放出来，你拿去改改就能用。

一、为什么 Celery 默认配置不够用

Celery 是好东西，我们工作室现在还在用。但它的默认配置是面向通用任务队列的，放到爬虫场景下有几个硬伤。

先承认 Celery 的优势：生态成熟，监控工具齐全，接入成本低。如果你只是跑一些后台任务，Celery 开箱即用没什么问题。

但爬虫场景不一样。

往下拆一层看，核心矛盾在两个地方。

第一，任务确认时机。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 队列。

2.4 代理层：隧道代理

代理 IP 管理是一个独立的关注点，不应该和任务队列逻辑耦合在一起。我们用的是爬虫代理（隧道模式），配置固定域名、端口、用户名和密码，由服务商在云端自动切换 IP。

选隧道代理而不是 API 提取式代理的原因很简单：隧道代理不需要你自己维护 IP 池和淘汰逻辑，服务商帮你做了 IP 轮换。你的代码只需要处理"请求失败后重试"这一个逻辑就行。对于任务队列架构来说，代理层越简单越好，复杂度留给队列层。

当然，如果你的业务对 IP 有特殊要求（比如指定地区），可以切换到 API 提取模式。下面的代码两种都给了。

三、代码实现

下面是完整可运行的项目模板。技术栈是 Python 3.9+，Redis 6+，requests 库。代理用的是隧道代理。

3.1 配置文件

这段定义全局配置。实际项目中建议用环境变量管理敏感信息，不要硬编码。

# config.py
"""爬虫任务队列全局配置"""

# Redis 配置
REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
REDIS_DB = 0
REDIS_PASSWORD = None  # 生产环境务必设置密码

# 代理（隧道模式）配置
# 替换为亿牛云代理提供的实际参数
PROXY_HOST = "t.16yun.cn"
PROXY_PORT = "31111"
PROXY_USER = "your_username"
PROXY_PASS = "your_password"

# 如果使用 API 提取式代理，配置提取接口
# API 提取地址在亿牛云控制台获取
PROXY_API_URL = "http://ip.16yun.cn:817/myip/pl/?s=YOUR_KEY&u=YOUR_USER&format=json"

# 任务队列参数
TASK_LEASE_TTL = 30          # 任务租约超时时间（秒）
HEARTBEAT_INTERVAL = 5       # worker 心跳间隔（秒）
CHECKPOINT_BATCH_SIZE = 50   # 每处理多少条数据保存一次 checkpoint
MAX_RETRY_COUNT = 3          # 单个任务最大重试次数
RECLAIM_INTERVAL = 10        # 过期任务巡检间隔（秒）

# 请求参数
REQUEST_TIMEOUT = 15         # 单次请求超时（秒）
MAX_PROXY_RETRIES = 3        # 单次请求最大代理重试次数
BACKOFF_BASE = 1.5           # 指数退避基数（秒）

3.2 Redis 任务队列核心

这是整个方案的核心。任务队列、状态机、checkpoint、租约管理都在这里。

# task_queue.py
"""
基于 Redis 的高可用爬虫任务队列

特性：
- 任务状态机：PENDING → RUNNING → COMPLETED / FAILED / DEAD_LETTER
- 断点续爬：每个任务带 checkpoint 字段，崩溃后可从断点恢复
- 故障转移：租约机制 + 心跳续期，worker 挂了任务自动回收
"""

import json
import time
import uuid
import logging
from enum import Enum
from typing import Optional, Any

import redis

from config import (
    REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD,
    TASK_LEASE_TTL, MAX_RETRY_COUNT, CHECKPOINT_BATCH_SIZE,
)

logger = logging.getLogger(__name__)


class TaskStatus(str, Enum):
    """任务状态枚举"""
    PENDING = "pending"          # 等待处理
    RUNNING = "running"          # 正在处理
    COMPLETED = "completed"      # 已完成
    FAILED = "failed"            # 失败（可重试）
    DEAD_LETTER = "dead_letter"  # 死信（超过最大重试次数）


class TaskQueue:
    """
    Redis 任务队列

    数据结构说明：
    - spider:tasks:pending    → List，待处理任务队列（FIFO）
    - spider:tasks:running    → ZSet，运行中任务（score = 租约过期时间戳）
    - spider:tasks:completed  → Set，已完成任务 ID
    - spider:tasks:dead       → List，死信队列
    - spider:task:{task_id}   → Hash，单个任务的元数据
    - spider:task:{task_id}:checkpoint → String，任务的断点信息
    - spider:worker:{worker_id}:heartbeat → String，worker 心跳（带 TTL）
    """

    # Redis key 前缀
    KEY_PENDING = "spider:tasks:pending"
    KEY_RUNNING = "spider:tasks:running"
    KEY_COMPLETED = "spider:tasks:completed"
    KEY_DEAD = "spider:tasks:dead"
    KEY_TASK_PREFIX = "spider:task:"
    KEY_CHECKPOINT_SUFFIX = ":checkpoint"
    KEY_WORKER_PREFIX = "spider:worker:"
    KEY_HEARTBEAT_SUFFIX = ":heartbeat"

    def __init__(self, redis_client: Optional[redis.Redis] = None):
        self.redis = redis_client or redis.Redis(
            host=REDIS_HOST,
            port=REDIS_PORT,
            db=REDIS_DB,
            password=REDIS_PASSWORD,
            decode_responses=True,
        )

    def push_task(
        self,
        task_type: str,
        url: str,
        priority: int = 0,
        meta: Optional[dict] = None,
    ) -> str:
        """
        推入一个新任务

        参数：
            task_type: 任务类型（如 "detail_page", "list_page"）
            url: 目标 URL
            priority: 优先级（数字越小越先执行，0 为默认）
            meta: 附加元数据（如分页参数、分类信息等）

        返回：任务 ID
        """
        task_id = str(uuid.uuid4())
        now = time.time()

        task_data = {
   
            "task_id": task_id,
            "task_type": task_type,
            "url": url,
            "priority": str(priority),
            "status": TaskStatus.PENDING.value,
            "retry_count": "0",
            "created_at": str(now),
            "started_at": "",
            "completed_at": "",
            "meta": json.dumps(meta or {
   }, ensure_ascii=False),
        }

        # 用 pipeline 批量写入，减少网络往返
        pipe = self.redis.pipeline()
        pipe.hset(f"{self.KEY_TASK_PREFIX}{task_id}", mapping=task_data)
        pipe.lpush(self.KEY_PENDING, task_id)
        pipe.execute()

        logger.info("任务已入队: task_id=%s, type=%s, url=%s", task_id, task_type, url)
        return task_id

    def pop_task(self, worker_id: str) -> Optional[dict]:
        """
        取出一个待处理任务

        取出后任务状态变为 RUNNING，同时设置租约。
        如果 worker 处理过程中崩溃，租约过期后任务会被回收。

        参数：
            worker_id: 当前 worker 的唯一标识

        返回：任务字典，没有可用任务时返回 None
        """
        # 从 pending 队列阻塞取出一个任务 ID（最多等 5 秒）
        result = self.redis.brpoplpush(
            self.KEY_PENDING,    # 源队列
            self.KEY_PENDING,    # 临时放回源队列（原子操作，防止丢失）
            timeout=5,
        )
        if not result:
            return None

        task_id = result
        task_key = f"{self.KEY_TASK_PREFIX}{task_id}"

        # 从源队列移除（因为 brpoplpush 会复制到目标队列，这里目标也是源队列）
        self.redis.lrem(self.KEY_PENDING, 1, task_id)

        # 获取任务元数据
        task_data = self.redis.hgetall(task_key)
        if not task_data:
            logger.warning("任务元数据不存在，可能已被清理: %s", task_id)
            return None

        # 更新状态为 RUNNING，记录开始时间
        now = time.time()
        lease_expires = now + TASK_LEASE_TTL

        pipe = self.redis.pipeline()
        pipe.hset(task_key, mapping={
   
            "status": TaskStatus.RUNNING.value,
            "started_at": str(now),
            "worker_id": worker_id,
        })
        # 加入 running 有序集合，score 是租约过期时间戳
        pipe.zadd(self.KEY_RUNNING, {
   task_id: lease_expires})
        pipe.execute()

        # 解析 JSON 字段
        task_data["status"] = TaskStatus.RUNNING.value
        task_data["started_at"] = str(now)
        task_data["worker_id"] = worker_id
        task_data["meta"] = json.loads(task_data.get("meta", "{}"))

        logger.info("任务已取出: task_id=%s, worker=%s", task_id, worker_id)
        return task_data

    def save_checkpoint(self, task_id: str, checkpoint: dict) -> None:
        """
        保存任务的断点信息

        在处理分页数据时，每处理完一批就调用此方法。
        如果 worker 崩溃，重启后可通过 get_checkpoint 恢复进度。

        参数：
            task_id: 任务 ID
            checkpoint: 断点数据（如 {"page": 47, "last_id": "abc123"}）
        """
        key = f"{self.KEY_TASK_PREFIX}{task_id}{self.KEY_CHECKPOINT_SUFFIX}"
        self.redis.set(key, json.dumps(checkpoint, ensure_ascii=False))
        logger.debug("断点已保存: task_id=%s, checkpoint=%s", task_id, checkpoint)

    def get_checkpoint(self, task_id: str) -> Optional[dict]:
        """
        获取任务的断点信息

        返回：断点字典，没有断点记录时返回 None
        """
        key = f"{self.KEY_TASK_PREFIX}{task_id}{self.KEY_CHECKPOINT_SUFFIX}"
        data = self.redis.get(key)
        if data:
            return json.loads(data)
        return None

    def renew_lease(self, task_id: str) -> bool:
        """
        续租任务

        worker 在处理任务期间应定期调用此方法刷新租约。
        如果租约已过期（任务可能已被回收），返回 False。

        返回：续租成功返回 True，租约已过期返回 False
        """
        # 检查任务是否还在 running 集合中
        if not self.redis.zscore(self.KEY_RUNNING, task_id):
            logger.warning("任务不在运行中集合，租约可能已过期: %s", task_id)
            return False

        # 刷新租约过期时间
        new_expires = time.time() + TASK_LEASE_TTL
        self.redis.zadd(self.KEY_RUNNING, {
   task_id: new_expires})
        return True

    def complete_task(self, task_id: str) -> None:
        """
        标记任务完成

        清理任务数据，从 running 集合移除，加入 completed 集合。
        checkpoint 数据也会被清理。
        """
        pipe = self.redis.pipeline()
        pipe.hset(f"{self.KEY_TASK_PREFIX}{task_id}", mapping={
   
            "status": TaskStatus.COMPLETED.value,
            "completed_at": str(time.time()),
        })
        pipe.zrem(self.KEY_RUNNING, task_id)
        pipe.sadd(self.KEY_COMPLETED, task_id)
        pipe.delete(f"{self.KEY_TASK_PREFIX}{task_id}{self.KEY_CHECKPOINT_SUFFIX}")
        pipe.execute()

        logger.info("任务已完成: %s", task_id)

    def fail_task(self, task_id: str, error: str) -> None:
        """
        标记任务失败

        如果重试次数未超上限，任务重新放回 pending 队列。
        超过最大重试次数，任务进入死信队列，需人工介入。

        参数：
            task_id: 任务 ID
            error: 错误信息
        """
        task_key = f"{self.KEY_TASK_PREFIX}{task_id}"
        retry_count = int(self.redis.hget(task_key, "retry_count") or "0")
        retry_count += 1

        pipe = self.redis.pipeline()
        pipe.zrem(self.KEY_RUNNING, task_id)
        pipe.hset(task_key, "retry_count", str(retry_count))
        pipe.hset(task_key, "last_error", error)

        if retry_count >= MAX_RETRY_COUNT:
            # 超过最大重试次数，进入死信队列
            pipe.hset(task_key, "status", TaskStatus.DEAD_LETTER.value)
            pipe.lpush(self.KEY_DEAD, task_id)
            logger.error(
                "任务进入死信队列: %s, 重试次数=%d, 错误=%s",
                task_id, retry_count, error,
            )
        else:
            # 重新入队等待重试
            pipe.hset(task_key, "status", TaskStatus.PENDING.value)
            pipe.lpush(self.KEY_PENDING, task_id)
            logger.warning(
                "任务重试: %s, 第 %d 次, 错误=%s",
                task_id, retry_count, error,
            )

        pipe.execute()

    def reclaim_expired_tasks(self) -> int:
        """
        回收过期任务（故障转移核心方法）

        扫描 running 集合中租约已过期的任务，将其重新放回 pending 队列。
        这个方法应该由一个独立的巡检线程定期调用。

        返回：本次回收的任务数量
        """
        now = time.time()

        # 找出所有租约已过期的任务（score < 当前时间戳）
        expired = self.redis.zrangebyscore(self.KEY_RUNNING, 0, now)

        if not expired:
            return 0

        pipe = self.redis.pipeline()
        for task_id in expired:
            task_key = f"{self.KEY_TASK_PREFIX}{task_id}"

            # 检查任务是否已经被其他 worker 完成或回收
            status = self.redis.hget(task_key, "status")
            if status and status != TaskStatus.RUNNING.value:
                # 任务状态已变更，从 running 集合清理即可
                pipe.zrem(self.KEY_RUNNING, task_id)
                continue

            # 租约过期，任务重新入队
            pipe.zrem(self.KEY_RUNNING, task_id)
            pipe.hset(task_key, "status", TaskStatus.PENDING.value)
            pipe.lpush(self.KEY_PENDING, task_id)

            logger.warning("回收过期任务: %s", task_id)

        pipe.execute()
        return len(expired)

    def get_stats(self) -> dict:
        """
        获取队列统计信息（用于监控）

        返回各状态任务数量
        """
        return {
   
            "pending": self.redis.llen(self.KEY_PENDING),
            "running": self.redis.zcard(self.KEY_RUNNING),
            "completed": self.redis.scard(self.KEY_COMPLETED),
            "dead_letter": self.redis.llen(self.KEY_DEAD),
        }

代码不短，但逻辑是线性的。几个关键设计要说一下。

pop_task 方法里用了 brpoplpush 做原子取出，防止 worker 取出任务后还没来得及设置租约就崩溃导致任务丢失。这一步很多人直接用 rpop + zadd 两步操作，中间断了任务就没了。

reclaim_expired_tasks 是故障转移的关键。它扫描 ZSet 里 score（租约过期时间）小于当前时间的任务，把状态重置为 PENDING 重新入队。注意这里检查了任务当前状态，防止重复回收。

3.3 代理封装

代理层单独抽出来，和任务队列解耦。下面封装了一个带自动重试和退避的代理请求方法。

# 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 字典
        """
        proxy_url = f"http://{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.info("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 {
   }

        proxy_url = f"http://{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, attempt, 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，不用你手动管理。

还有一个容易忽略的细节：退避策略要加随机抖动。如果你有 10 个 worker 同时碰到 403，它们会同时退避、同时重试，形成"惊群效应"。加一个 0 到 0.5 秒的随机偏移就把这个问题打散了。

3.4 Worker：心跳、租约、断点续爬

Worker 是把上面三个组件串起来的地方。心跳线程独立运行，处理任务的主循环里定期续租和保存 checkpoint。

# worker.py
"""
爬虫 Worker 进程

职责：
1. 从任务队列取任务
2. 通过代理请求目标页面
3. 处理过程中定期保存 checkpoint（断点续爬）
4. 定期发送心跳续租（故障转移）
5. 任务完成或失败后更新状态

启动方式：
    python worker.py
    python worker.py  # 起多个进程即可横向扩展
"""

import os
import sys
import time
import json
import uuid
import signal
import logging
import threading
from typing import Optional

import redis

from config import (
    REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD,
    HEARTBEAT_INTERVAL, CHECKPOINT_BATCH_SIZE,
    RECLAIM_INTERVAL, TASK_LEASE_TTL,
)
from task_queue import TaskQueue, TaskStatus
from proxy_manager import ProxyManager

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("worker")


class SpiderWorker:
    """爬虫 Worker"""

    def __init__(self, proxy_mode: str = "tunnel"):
        """
        初始化 worker

        参数：
            proxy_mode: 代理模式，"tunnel" 或 "api"
        """
        self.worker_id = f"worker-{os.getpid()}-{uuid.uuid4().hex[:8]}"
        self.redis = redis.Redis(
            host=REDIS_HOST,
            port=REDIS_PORT,
            db=REDIS_DB,
            password=REDIS_PASSWORD,
            decode_responses=True,
        )
        self.queue = TaskQueue(self.redis)
        self.proxy = ProxyManager(mode=proxy_mode)

        # 运行状态控制
        self._running = True
        self._heartbeat_thread: Optional[threading.Thread] = None
        self._reclaim_thread: Optional[threading.Thread] = None

        # 当前正在处理的任务 ID（用于心跳续租）
        self._current_task_id: Optional[str] = None
        self._task_lock = threading.Lock()

    def _heartbeat_loop(self):
        """
        心跳线程：定期续租当前任务

        每 HEARTBEAT_INTERVAL 秒执行一次：
        1. 更新 worker 心跳 key 的 TTL
        2. 如果有当前任务，续租
        """
        heartbeat_key = f"spider:worker:{self.worker_id}:heartbeat"

        while self._running:
            try:
                # 刷新 worker 心跳
                self.redis.setex(heartbeat_key, TASK_LEASE_TTL, "alive")

                # 续租当前任务
                with self._task_lock:
                    if self._current_task_id:
                        if not self.queue.renew_lease(self._current_task_id):
                            # 租约续期失败，任务可能已被回收
                            logger.warning(
                                "租约续期失败，任务可能已被回收: %s",
                                self._current_task_id,
                            )

            except Exception as e:
                logger.error("心跳异常: %s", e)

            time.sleep(HEARTBEAT_INTERVAL)

    def _reclaim_loop(self):
        """
        巡检线程：定期回收过期任务

        独立于心跳线程，避免心跳逻辑阻塞巡检。
        在多 worker 环境下，每个 worker 都跑巡检不会有问题，
        因为 reclaim_expired_tasks 是幂等操作。
        """
        while self._running:
            try:
                count = self.queue.reclaim_expired_tasks()
                if count > 0:
                    logger.info("本次回收过期任务: %d 个", count)
            except Exception as e:
                logger.error("巡检异常: %s", e)

            time.sleep(RECLAIM_INTERVAL)

    def _process_task(self, task: dict) -> bool:
        """
        处理单个任务

        这是业务逻辑的核心方法，需要根据你的实际采集需求改写。
        下面是一个分页采集的示例：
        - 检查是否有断点记录
        - 从断点页码开始继续采集
        - 每处理 CHECKPOINT_BATCH_SIZE 条数据保存一次断点

        参数：
            task: 任务字典

        返回：成功返回 True，失败返回 False
        """
        task_id = task["task_id"]
        url = task["url"]
        task_type = task["task_type"]
        meta = task.get("meta", {
   })

        # 读取断点信息（如果之前跑了一半挂了，从这里恢复）
        checkpoint = self.queue.get_checkpoint(task_id)

        if checkpoint:
            logger.info("从断点恢复: task_id=%s, checkpoint=%s", task_id, checkpoint)
            start_page = checkpoint.get("page", 1)
            processed_count = checkpoint.get("processed", 0)
        else:
            start_page = meta.get("start_page", 1)
            processed_count = 0

        # ---- 以下是业务逻辑示例 ----
        # 假设这是一个分页详情页采集任务
        current_page = start_page
        batch_counter = 0

        while self._running:
            # 构造分页 URL
            page_url = f"{url}?page={current_page}"

            # 通过代理发起请求
            resp = self.proxy.request_with_retry("GET", page_url)

            if resp is None:
                # 请求彻底失败（代理重试也救不了）
                return False

            # 解析响应数据（这里用模拟逻辑，实际项目替换为你的解析代码）
            try:
                # 假设目标站点返回 JSON 格式的列表数据
                data = resp.json()
                items = data.get("items", [])

                if not items:
                    # 没有更多数据了，采集完成
                    logger.info("任务采集完成，无更多数据: %s", task_id)
                    break

                # 处理每条数据
                for item in items:
                    # 这里写你的数据处理逻辑
                    # 比如：解析字段、去重、写入数据库等
                    processed_count += 1
                    batch_counter += 1

                    # 每处理一批数据就保存断点
                    if batch_counter >= CHECKPOINT_BATCH_SIZE:
                        self.queue.save_checkpoint(task_id, {
   
                            "page": current_page,
                            "processed": processed_count,
                            "last_item_id": item.get("id", ""),
                        })
                        batch_counter = 0
                        logger.info(
                            "断点已保存: page=%d, processed=%d",
                            current_page, processed_count,
                        )

                current_page += 1

            except (json.JSONDecodeError, KeyError) as e:
                logger.error("解析数据失败: %s, 错误=%s", page_url, e)
                return False

        # 最终再保存一次断点（确保最后的数据不丢）
        self.queue.save_checkpoint(task_id, {
   
            "page": current_page,
            "processed": processed_count,
            "status": "completed",
        })

        return True

    def run(self):
        """Worker 主循环"""
        logger.info("Worker 启动: %s", self.worker_id)

        # 注册信号处理，优雅退出
        def handle_signal(signum, frame):
            logger.info("收到退出信号 %d，准备优雅退出...", signum)
            self._running = False

        signal.signal(signal.SIGINT, handle_signal)
        signal.signal(signal.SIGTERM, handle_signal)

        # 启动心跳线程
        self._heartbeat_thread = threading.Thread(
            target=self._heartbeat_loop, daemon=True,
        )
        self._heartbeat_thread.start()

        # 启动巡检线程
        self._reclaim_thread = threading.Thread(
            target=self._reclaim_loop, daemon=True,
        )
        self._reclaim_thread.start()

        # 主循环：取任务、处理、更新状态
        while self._running:
            try:
                # 从队列取出任务（阻塞式，最多等 5 秒）
                task = self.queue.pop_task(self.worker_id)

                if task is None:
                    continue

                task_id = task["task_id"]

                with self._task_lock:
                    self._current_task_id = task_id

                # 处理任务
                success = self._process_task(task)

                with self._task_lock:
                    self._current_task_id = None

                # 更新任务状态
                if success:
                    self.queue.complete_task(task_id)
                else:
                    self.queue.fail_task(task_id, "处理失败，详见日志")

            except Exception as e:
                logger.error("Worker 主循环异常: %s", e)
                # 异常后短暂等待，避免死循环疯狂报错
                time.sleep(1)

        # 优雅退出：清理心跳 key
        heartbeat_key = f"spider:worker:{self.worker_id}:heartbeat"
        self.redis.delete(heartbeat_key)
        logger.info("Worker 已退出: %s", self.worker_id)


if __name__ == "__main__":
    # 默认使用隧道代理模式
    # 如果需要 API 提取模式，改为 SpiderWorker(proxy_mode="api")
    worker = SpiderWorker(proxy_mode="tunnel")
    worker.run()

这段代码有几个设计决策值得展开说。

心跳和巡检放在独立线程而不是主循环里。原因是任务处理可能比较慢（单个任务跑几分钟），如果心跳逻辑依赖主循环，任务卡住的时候心跳也卡住了，租约过期，任务被别的 worker 抢走，造成重复处理。

_process_task 里的业务逻辑是示例。你需要根据自己的采集场景改写。但框架是固定的：读 checkpoint、处理数据、定期存 checkpoint、处理完返回 True/False。

优雅退出靠信号处理。收到 SIGTERM 时设置 _running = False，主循环下一次检查就会退出。当前正在处理的任务不会立即中断，会处理完当前批次再退出。如果 worker 被直接 kill -9，那当前任务靠租约超时被巡检线程回收，这就是故障转移兜底。

3.5 任务投放入口

写个简单的脚本往队列里投放任务，方便测试。

# seed_tasks.py
"""
任务投放脚本

批量往队列里投放采集任务，用于初始化或补充任务。
"""

import logging
from task_queue import TaskQueue

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("seed")


def seed_tasks():
    """投放一批测试任务"""
    queue = TaskQueue()

    # 示例：投放 100 个列表页采集任务
    base_url = "https://example.com/api/list"
    total = 100

    for i in range(1, total + 1):
        task_id = queue.push_task(
            task_type="list_page",
            url=base_url,
            meta={
   "start_page": i, "category": "default"},
        )

    logger.info("已投放 %d 个任务", total)

    # 查看队列状态
    stats = queue.get_stats()
    logger.info("当前队列状态: %s", stats)


if __name__ == "__main__":
    seed_tasks()

3.6 完整启动流程

三步走：

# 第一步：启动 Redis（如果还没装）
docker run -d --name redis -p 6379:6379 redis:7

# 第二步：投放任务
python seed_tasks.py

# 第三步：启动 worker（可以开多个终端跑多个实例）
python worker.py
python worker.py
python worker.py

worker 可以横向扩展。起 5 个、10 个都行，Redis 自己扛得住。每个 worker 都带巡检线程，多个 worker 同时巡检也不会冲突，因为 reclaim_expired_tasks 是幂等的。

四、跑起来会踩的坑

下面这些坑我们都踩过，提前说给你听。

坑一：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)) 的随机间隔。别用固定间隔，固定间隔是爬虫最明显的特征之一。

坑五：死信队列要有人盯。 任务进入死信队列不是结束，是开始。我们工作室的做法是每天早上跑一个脚本统计死信数量，超过阈值就告警。死信里的任务需要人工排查原因，修复后手动重新入队。

五、监控指标

线上跑起来之后，这几个指标要盯着。

get_stats 方法返回的就是上面前四个指标。线上建议接 Prometheus + Grafana 做可视化，或者简单点写个定时脚本发钉钉/飞书告警。

总结

回溯一下核心设计。

高可用爬虫任务队列的核心是"状态可恢复"和"故障可转移"。状态可恢复靠 checkpoint 机制，每个任务定期保存进度，崩溃后从断点继续。故障转移靠租约 + 心跳，worker 挂了任务自动回收重新入队。代理层用隧道代理做 IP 轮换，和任务队列解耦，各管各的。

这套方案的适用范围：单机到小规模集群（3 到 10 台机器），任务量在百万级以内。如果你要扛千万级任务，Redis 单实例会成瓶颈，得考虑分片或者换 Kafka。但对于大部分采集项目来说，这套设计够用了，而且代码量不大，出问题好排查。

代理这块怎么选，得看你们自己的业务量级。隧道代理适合"不想自己管 IP 池"的团队，API 提取式适合"需要精细控制 IP 地区和有效期"的场景。我们工作室法律行业那个项目用的是隧道代理加强版，成功率稳定在 95% 以上，够用了。