Redis Stream 怎样把超时消息转给新消费者

Redis Stream 消费组里，消息被某个消费者读走以后，会先进入 Pending Entries List（PEL）。如果旧消费者宕机、卡住或处理时间明显超过约定阈值，新消费者可以先查出空闲时间足够长的消息，再用 XCLAIM 把它们转到自己名下。关键点是：XCLAIM 只改变消费组中的归属，不代表业务已经成功；新消费者仍要用消息 ID 做幂等判断，业务成功后再执行 XACK。

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

可靠的接管闭环是“找出 PEL 中超时的 ID → XCLAIM 认领 → 按消息 ID 判重 → 处理成功 → XACK”。不要用“读取不到新消息”直接推断没有积压，也不要把认领操作当成业务确认。

先看清 Redis Stream 中发生了什么变化

假设订单通知由消费组 order-workers 处理。消费者 worker-a 通过 XREADGROUP 读到消息 1710000000000-0，Redis 会把这条消息加入消费组的 PEL，并记录当前消费者、空闲时间和投递次数。只要 worker-a 没有执行 XACK，消息就不会因为“被读过”而从待确认集合中消失。

当 worker-a 停止响应，worker-b 的恢复任务应先检查 PEL，再根据业务约定的空闲阈值认领。这个变化解决的是消费者归属问题：它让恢复任务有机会继续处理未确认消息，而不是重新把整个 Stream 当作普通列表扫描。

用 XPENDING 找出可以接管的消息

XPENDING 的作用是观察消费组的待处理消息。实际恢复程序通常需要先拿到消息 ID，再把 ID 交给 XCLAIM。不要只按消息进入 Stream 的时间判断超时，因为“消息创建时间”与“被某个消费者读走后的空闲时间”不是同一个概念。

XPENDING orders order-workers - + 20
# 第一个参数是 Stream，第二个参数是消费组。
# - 和 + 表示查看 PEL 的 ID 范围，20 表示最多取 20 条摘要。
# 生产程序还应读取每条消息的空闲时间和投递次数，再决定是否认领。

如果只想按某个消费者查看，也可以把范围和数量调整为面向该消费者的查询。工程上要把“可接管阈值”设得比正常处理时长更宽一些：太短会把仍在工作的消费者误判为失联，太长又会让积压恢复得过慢。

让新消费者用 XCLAIM 接手超时消息

XCLAIM 接收 Stream、消费组、新消费者名称、最小空闲时间和一个或多个消息 ID。只有目标消息在该消费组 PEL 中，并且空闲时间达到最小值时，认领才符合预期。Redis 官方命令参考还提供 IDLE、TIME、RETRYCOUNT、JUSTID 等选项；本文只把它们作为控制归属和重试信息的辅助能力，不把它们当作业务幂等。

XCLAIM orders order-workers worker-b 120000 1710000000000-0
# 120000 表示至少空闲 120 秒，避免过早抢走仍在处理的消息。
# worker-b 是新的归属消费者；这里只认领指定消息 ID。
# 返回消息内容后，worker-b 还必须执行幂等检查和业务处理。

如果恢复任务只需要先搬运 ID，可以使用 JUSTID，这样返回结果更轻；如果后续马上要处理消息体，则直接取得消息内容更方便。批量认领时要限制单次 ID 数量，让恢复任务的处理时间保持可控，否则一次接管过多消息会把故障从“无人处理”变成“新消费者再次积压”。

认领之后先用消息 ID 做幂等

旧消费者可能只是网络抖动，并没有真正退出。此时 worker-a 与 worker-b 可能先后执行同一条消息，甚至同时触发外部副作用。因此，认领不是“只执行一次”的保证，幂等键才是业务层的保护。

一个可落地的顺序是：以 Redis Stream 消息 ID 或业务事件 ID 作为幂等键；检查幂等记录；尚未处理时执行业务；只有业务成功后写入完成状态并确认消息。幂等记录的写入与外部系统调用如果不在同一个事务内，就要结合业务状态机、唯一约束或可重试的接口设计，不能仅靠一条普通的 Redis 标记声称绝对一次。

-- 伪代码中的 Redis 操作顺序，具体事务边界由业务存储决定
if not idempotency_store.exists(message_id) then
    -- 先建立“处理中”或占用状态，防止并发消费者同时执行。
    idempotency_store.reserve(message_id)
    ok = handle_business(message)
    if ok then
        -- 业务成功后才确认 Stream 消息。
        idempotency_store.mark_done(message_id)
        XACK orders order-workers message_id
    end
else
    -- 已完成或已占用的消息不能再次触发外部副作用。
    XACK orders order-workers message_id
end

上面的代码是流程示意，不是 Redis CLI 可直接执行的脚本。真实实现还要明确“处理中”超时如何恢复、“已完成”如何长期保存，以及业务失败时是否释放占用状态。对于支付、库存、发货等外部副作用，最好让下游接口也接受同一个幂等键。

处理成功后再 XACK，失败则保留重试机会

XACK 会把消息从消费组的 PEL 中确认移除。确认太早会造成消息已经不再待处理，但业务还没有完成；确认太晚则会让恢复任务反复看到它。正确的边界通常是：消息已经被当前消费者接管，幂等状态允许执行，业务处理和必要的持久化结果都成功之后，再确认。

XACK orders order-workers 1710000000000-0
# 只有业务结果已经落库或下游已接受幂等请求后才确认。
# 如果处理失败，不要用 XACK 掩盖失败；保留 PEL 记录并记录错误原因。

重试次数应参与退避和死信策略。若同一条消息不断被认领、处理、失败，再次被认领，说明问题可能不是消费者宕机，而是输入数据、下游依赖或业务规则持续失败。此时应把它转入人工可观察的异常队列或记录表，而不是无限增加认领频率。

一套可执行的接管清单

常见问题

XCLAIM 会自动把消息重新读取出来吗？

它会改变符合条件消息在消费组中的消费者归属，并可按选项返回消息或 ID。恢复程序仍要负责业务处理和最终确认。

为什么刚查到的消息不能立刻认领？

通常是因为它的空闲时间还没有达到 min-idle-time，或者它不在目标消费组的 PEL 中。阈值应反映真实处理时长，而不是随意设成很小的值。

用了 XCLAIM 还会重复处理吗？

会。旧消费者与新消费者可能发生并发或迟到重试，所以必须在业务层以消息 ID 或业务键做幂等，不能把 XCLAIM 当成 exactly-once 机制。

什么时候应该确认消息？

在业务结果已经可靠保存、必要的外部调用已经按幂等语义完成之后确认。失败时保留待处理状态，交给退避和重试策略。

总结

Redis Stream 的超时接管可以拆成三个边界：PEL 负责告诉你哪些消息已投递但未确认，XCLAIM 负责把符合空闲阈值的消息交给新消费者，业务幂等记录负责避免旧消费者迟到造成重复副作用。把 XACK 放在业务成功之后，并为持续失败设计退避或异常处理，才能让“新消费者接手”真正成为可恢复的消费机制。