Redis Stream用消费组实现可重试任务队列的方案

Redis Stream 用消费组实现可重试任务队列时，核心不是简单地把消息读出来，而是把“已投递但未完成”的状态留下来：XREADGROUP 负责分发，XACK 只在业务成功后确认，超时未确认的消息再由 XPENDING 和 XAUTOCLAIM 找回。这样得到的是至少一次投递语义，业务处理器必须具备幂等能力。

官方地址：https://redis.io/docs/latest/

最小可靠闭环是：生产者用 XADD 写入，消费组用 XREADGROUP ... > 取新消息，成功后执行 XACK，故障恢复时按空闲时间执行 XAUTOCLAIM。

建立 Stream 与消费组

先把任务设计成字段集合，例如 job_id、payload 和 attempt。消费组是 Stream 上的协作边界，同组消费者不会重复领取同一条新消息；不同组则各自保留消费进度，适合“任务处理”和“审计订阅”两条独立链路。

# 写入一条任务；* 让 Redis 生成单调递增的消息 ID
redis-cli XADD orders:stream * job_id order-1001 payload "ship" attempt 0

# 创建消费组；0-0 表示从已有历史消息开始，MKSTREAM 允许空 Stream 初始化
redis-cli XGROUP CREATE orders:stream order-workers 0-0 MKSTREAM

如果只希望处理消费组创建之后的新消息，可把起点改成 $。生产环境要把 Stream 名称、消费组名称和字段协议写进配置，避免多个服务误用同一个组名。

用 XREADGROUP 读取并保留待确认消息

消费者第一次拉取新消息时使用 >。Redis 将已投递、尚未确认的消息放入该组的 Pending Entries List（PEL），因此进程在执行业务时退出，消息不会因为“读过一次”而丢失。

# 只读取本组尚未投递给任何消费者的新消息
redis-cli XREADGROUP GROUP order-workers worker-a COUNT 10 BLOCK 5000 STREAMS orders:stream >

# 重新读取当前消费者名下的待确认消息；0 表示从该消费者的 PEL 开始
redis-cli XREADGROUP GROUP order-workers worker-a COUNT 10 STREAMS orders:stream 0

COUNT 控制一次拉取的上限，BLOCK 只影响等待新消息的时间。读取到消息后先解析 ID 和字段，再执行业务；不要在刚读到时就确认，否则消费者崩溃会造成任务丢失。消费者名称应稳定且可观测，常见做法是使用实例名加进程标识。

用 XACK、XPENDING 与 XAUTOCLAIM 组织重试

处理订单、通知或同步任务时，只有业务提交成功后才执行 XACK。确认会从 PEL 移除这条消息，但不会自动删除 Stream 中的原始记录，因此保留策略要单独设计。

# 业务事务成功后确认；消息 ID 必须使用本次读取到的原值
redis-cli XACK orders:stream order-workers 1710000000000-0

# 查看消费组中待确认的数量和空闲时间，用于发现卡住的消费者
redis-cli XPENDING orders:stream order-workers

# 把空闲超过 60 秒的消息转给 worker-recovery，并返回下一页游标
redis-cli XAUTOCLAIM orders:stream order-workers worker-recovery 60000 0-0 COUNT 10

XAUTOCLAIM 的最小空闲时间不是“重试次数”，而是判断原消费者可能失联的时间窗口。恢复消费者接手后仍然要再次执行同一套幂等业务逻辑，并在成功后用原消息 ID 执行 XACK。若任务本身允许长时间处理，空闲阈值要大于正常处理时长，否则会出现两个消费者同时处理同一条消息。

用幂等与清理控制至少一次语义

至少一次意味着同一任务可能被原消费者和恢复消费者各执行一次。可以用业务唯一键做去重，例如把 job_id 写入独立的去重集合，或让数据库唯一约束保护最终写入；去重记录和业务结果最好在同一个事务边界内完成。

排障时重点观察 Stream 长度、各消费者的空闲时间、PEL 数量和重试次数。对已经确认的历史消息可以按业务保留周期裁剪；裁剪前要确认审计或补偿流程不再需要它。Redis Stream 适合可靠的异步分发，但不替代业务数据库事务，也不保证“只执行一次”。

常见问题

为什么 XREADGROUP 读不到历史消息？

新消息读取使用 >，它只匹配尚未投递给任何消费者的消息。要处理当前消费者的 PEL，应使用消息 ID 起点 0；要让新组从历史开始，则在创建组时使用 0-0。

XAUTOCLAIM 后为什么还会重复处理？

原消费者可能只是暂时变慢，或者恢复动作与原处理并发发生。把空闲阈值设在正常处理时长之上，并用业务唯一键、数据库唯一约束或幂等写入保护结果。

确认消息后 Stream 记录还在吗？

在。XACK 处理的是消费组的待确认状态，不负责删除 Stream 记录。历史清理应根据审计、补偿和回放需求单独使用保留策略。

总结：用消费组分配任务，用 PEL 记录未完成状态，用 XACK 表示业务成功，用 XPENDING 和 XAUTOCLAIM 恢复失联消费者，再用幂等键承受至少一次投递带来的重复执行。