基于 Redis Streams 与消费者组的分布式异步任务队列架构实践

引言

在分布式系统架构中，异步任务队列是实现服务解耦、流量削峰与后台耗时任务调度的核心组件。当业务规模尚小或对延迟敏感时，许多团队会优先选择 Redis 作为轻量级消息中间件。然而，传统的 Redis List（LPUSH / BLPOP）与 Pub/Sub 模式在可靠性、消息回溯及多消费者协作方面存在明显短板。Redis 5.0 引入的 Redis Streams 数据结构及其**消费者组（Consumer Groups）**机制，为构建高可靠的分布式异步任务队列提供了原生支持。

一、为什么 Redis Streams 优于 List (BLPOP) 与 Pub/Sub？

• 对比 List (BLPOP)：BLPOP 一旦将任务弹出，消息便从 Redis 中移除。如果 Worker 进程在处理中途发生 OOM 或宕机，任务就会永久丢失。虽然可以通过 RPOPLPUSH / LMOVE 模拟备份队列，但仍缺乏原生的超时重试与消费者状态追踪。
• 对比 Pub/Sub：Pub/Sub 是“发后即忘（Fire-and-Forget）”的内存广播模型，不存储历史消息。一旦订阅者断线重连，期间发布的所有任务都会丢失，且无法在多个 Worker 之间进行负载均衡。
• Redis Streams 的优势：Streams 采用只追加（Append-Only）的日志结构，每条消息拥有全局唯一且单调递增的 ID（如 1731560000000-0），结合消费者组能够同时满足持久化、广播、负载均衡、ACK 确认与故障转移等核心诉求。

二、核心机制：消费者组与死信恢复

1. 基础指令流转：XADD、XREADGROUP 与 XACK

1. 生产消息 (XADD)：生产者向指定 Stream 写入任务，通常使用 * 让 Redis 自动生成时间戳序列 ID：

   XADD tasks:email * user_id 10042 template welcome
   

2. 创建消费者组 (XGROUP CREATE)：指定从流头（0）或最新位置（$）开始消费：

   XGROUP CREATE tasks:email email_workers 0 MKSTREAM
   

3. 组内拉取任务 (XREADGROUP)：每个 Worker 使用唯一的 consumer_name 从组内读取未分配的新消息（>）：

   XREADGROUP GROUP email_workers worker-1 COUNT 10 BLOCK 5000 STREAMS tasks:email >
   

   被读取的消息会进入该消费者的 PEL（Pending Entries List，待处理列表），在收到确认前不会标记为完成。
4. 确认完成 (XACK)：任务执行成功后，Worker 调用 XACK 将消息从 PEL 中移除：

   XACK tasks:email email_workers 1731560000000-0
   

2. 故障恢复：XPENDING 与 XCLAIM

当某个 Worker 意外崩溃时，其已拉取但未 XACK 的消息会滞留在 PEL 中。恢复守护进程可以定期执行以下流程：

1. 排查超时消息 (XPENDING)：查询空闲时间（Idle Time）超过阈值的未确认消息：

   XPENDING tasks:email email_workers - + 10
   

2. 转移所有权 (XCLAIM / XAUTOCLAIM)：将超时（例如超过 60 秒）的消息认领给健康的消费者（如 worker-2）重新处理：

   XCLAIM tasks:email email_workers worker-2 60000 1731560000000-0
   

   若某条消息的投递次数（delivery_count）超过最大重试上限（如 5 次），则应将其转入专门的**死信队列（DLQ Stream）**并执行 XACK，避免“毒丸消息”阻塞整个队列。

三、Python redis-py 完整实现示例

以下示例展示了生产者写入、消费者组异步消费、以及基于 XAUTOCLAIM 的超时恢复与死信处理：

import json
import time
import redis

STREAM_KEY = "tasks:async_jobs"
DLQ_KEY = "tasks:async_jobs:dlq"
GROUP_NAME = "job_workers"
MAX_RETRIES = 3

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)


def init_group():
    try:
        r.xgroup_create(STREAM_KEY, GROUP_NAME, id="0", mkstream=True)
    except redis.exceptions.ResponseError as e:
        if "BUSYGROUP" not in str(e):
            raise


def produce_task(task_type: str, payload: dict) -> str:
    msg_id = r.xadd(
        STREAM_KEY,
        {"task_type": task_type, "payload": json.dumps(payload)},
        maxlen=100_000,
        approximate=True,
    )
    return msg_id


def process_message(msg_id: str, fields: dict):
    payload = json.loads(fields["payload"])
    print(f"[Process] id={msg_id}, type={fields['task_type']}, payload={payload}")
    # 模拟业务处理逻辑


def run_worker(consumer_name: str):
    init_group()
    while True:
        # 1. 优先通过 XAUTOCLAIM 恢复超过 30 秒未 ACK 的超时消息
        next_start, claimed, _ = r.xautoclaim(
            STREAM_KEY, GROUP_NAME, consumer_name, min_idle_time=30_000, start_id="0-0", count=5
        )
        for msg_id, fields in claimed:
            pending_info = r.xpending_range(STREAM_KEY, GROUP_NAME, msg_id, msg_id, 1)
            deliveries = pending_info[0]["times_delivered"] if pending_info else 1
            if deliveries > MAX_RETRIES:
                r.xadd(DLQ_KEY, {"origin_id": msg_id, **fields})
                r.xack(STREAM_KEY, GROUP_NAME, msg_id)
                print(f"[DLQ] Moved poison message {msg_id} after {deliveries} deliveries")
            else:
                process_message(msg_id, fields)
                r.xack(STREAM_KEY, GROUP_NAME, msg_id)

        # 2. 阻塞拉取新任务
        events = r.xreadgroup(
            GROUP_NAME, consumer_name, {STREAM_KEY: ">"}, count=5, block=3000
        )
        if not events:
            continue
        for _, messages in events:
            for msg_id, fields in messages:
                try:
                    process_message(msg_id, fields)
                    r.xack(STREAM_KEY, GROUP_NAME, msg_id)
                except Exception as exc:
                    print(f"[Error] Failed {msg_id}: {exc}")
                    time.sleep(0.5)

四、生产环境最佳实践

1. 使用 MAXLEN ~ 近似裁剪控制内存：Streams 不会在 XACK 后自动删除消息，长期运行会导致内存持续增长。生产环境应在 XADD 时携带 MAXLEN ~ 100000（对应 approximate=True），利用 Redis 基数树节点整块释放的特性，以极低的 CPU 开销将队列长度稳定在合理水位。
2. 消费端幂等性设计：由于网络抖动或 XCLAIM 接管的存在，分布式任务队列遵循 At-Least-Once（至少一次） 投递语义。业务侧必须基于任务唯一键（如 task_id 或业务订单号）配合数据库唯一索引或 Redis SET NX 实现消费幂等，防止重复扣款或重复通知。
3. 关键指标监控与告警：
   • 消费积压（Lag）：通过 XINFO GROUPS <stream> 监控 lag 与 pending 数量，及时发现 Worker 吞吐不足或异常挂起。
   • 死信队列告警：持续监控 XLEN tasks:async_jobs:dlq，一旦产生死信立即触发告警以便人工介入排查。
   • 僵尸消费者清理：容器弹性缩容后，定期通过 XINFO CONSUMERS 清理长期空闲的旧实例名称，避免元数据膨胀。

总结

借助 Redis Streams 与消费者组，我们无需引入重量级的独立消息中间件即可获得可靠的任务持久化、多组消费、ACK 确认及死信恢复能力。结合合理的 MAXLEN ~ 裁剪策略与幂等性保障，Redis Streams 足以胜任绝大多数中小规模至中高并发场景下的分布式异步任务调度需求。