Redis 消费者组 pending 消息的认领与恢复流程

Redis Streams 消费者取到消息后，如果进程在 XACK 之前退出，这条消息不会自动回到“新消息”队列，而是继续留在消费者组的 Pending Entries List（PEL）里。恢复的正确思路不是重复读取 >，而是先用 XPENDING 确认滞留情况，再按空闲时间用 XAUTOCLAIM 或 XCLAIM 转移所有权，幂等处理成功后执行 XACK。

Redis Streams 官方文档：https://redis.io/docs/latest/develop/data-types/streams/

先看 PEL，而不是立刻抢消息

我遇到过一次消费者重启后新消息仍在继续处理，但积压数始终不归零。原因不是 Stream 没数据，而是旧实例已经领取了一批消息，却没有确认。消费者组会为这类“已投递、未确认”的消息保留 PEL 记录；单纯再次执行 XREADGROUP ... > 读取的是从未投递的新消息，不会自动接管别人的 pending 消息。

先看组级摘要：

# 查看组内 pending 总数、最小/最大消息 ID 与各消费者持有数量
redis-cli XPENDING orders order-workers

再看明细：

# 分页查看前 20 条 pending，返回消息 ID、当前所有者、空闲毫秒和投递次数
redis-cli XPENDING orders order-workers - + 20

如果只怀疑某个失联实例，可以在明细命令末尾附加消费者名。需要筛选长期未处理的记录时，使用 IDLE 过滤：

# 只列出空闲时间超过 60 秒的 pending，先观察再决定是否认领
redis-cli XPENDING orders order-workers IDLE 60000 - + 20

认领阈值要覆盖正常处理时间

min-idle-time 是防止多个恢复消费者同时抢占正常工作的关键。它表示消息距离上次投递至少空闲多久才有资格被认领。阈值过小，会把仍在处理数据库事务、调用下游接口或等待重试的任务转给别人；阈值过大，又会延长故障恢复时间。

可以从“业务处理超时 + 下游最大合理抖动 + 调度余量”出发设定，并用分位耗时持续调整。例如常规处理在 10 秒内、超时上限 30 秒，就不要因为消息空闲 5 秒而认领。恢复消费者还应有唯一名称，便于从 PEL 和 XINFO CONSUMERS 判断所有权。

用 XAUTOCLAIM 批量扫描滞留消息

XAUTOCLAIM 把“扫描 PEL”和“认领符合空闲阈值的消息”合并在一个命令中。下面让恢复消费者 recovery-1 从 0-0 开始扫描，每次最多返回 20 条符合条件的消息：

# 从 PEL 起点扫描，认领空闲超过 60 秒的消息
redis-cli XAUTOCLAIM orders order-workers recovery-1 60000 0-0 COUNT 20

返回值会包含下一次扫描要使用的起始 ID，以及本次成功认领的消息。把返回的游标传给下一次调用，直到游标为 0-0，表示这一轮扫描已经到达末尾。COUNT 是返回目标上限，不保证每次一定获得这么多条，因为命令还要过滤空闲时间不足的 PEL 项。

如果恢复程序只需要消息 ID，可以使用 JUSTID；官方文档说明这种模式不会增加重试计数。一般业务恢复需要消息字段，直接取完整条目更方便。

已知消息 ID 时用 XPENDING 配合 XCLAIM

如果要人工处理少量指定记录，或者需要先按消费者、空闲时间和投递次数做更细的筛选，可以先用扩展形式的 XPENDING 找到 ID，再精确认领：

# 只认领明确选中的消息；空闲不足 60 秒时不会成功转移
redis-cli XCLAIM orders order-workers recovery-1 60000 1727510400000-0 1727510400001-0

XCLAIM 会在满足最小空闲时间时把消息所有权交给新消费者，并重置空闲时间。多个恢复实例同时竞争同一条消息时，只有满足条件并先完成转移的一方会成功。不过 Redis Streams 提供的是至少一次处理语义，网络超时、客户端重试和业务提交边界仍可能产生重复处理，不能把所有权竞争当成严格的业务去重。

认领后的处理必须幂等

恢复消费者拿到消息后，应使用稳定业务标识做幂等保护，例如订单号、事件 ID 或“消息 ID + 业务类型”。常见做法是在业务数据库中用唯一约束记录已处理事件，并让业务变更与去重记录处在同一个事务边界。只有业务结果已经持久化，才执行确认。

伪代码层面的处理边界可以概括为：

def handle_claimed(message):
    # 使用稳定事件 ID 检查业务是否已经落库，避免重复副作用
    if already_processed(message.event_id):
        ack(message.id)
        return

    try:
        # 业务变更与幂等记录应尽量放在同一事务边界
        save_business_result_and_mark_processed(message)
    except RetryableError:
        # 暂时失败时保留在 PEL，等待下一轮达到认领阈值
        return
    except PermanentError:
        # 稳定失败写入应用级隔离流，成功归档后再确认原消息
        archive_to_failure_stream(message)
        ack(message.id)
        return

    # 只有业务结果持久化成功后才确认
    ack(message.id)

这段示例强调的是边界，不绑定某个 Redis 客户端。实际实现必须处理“业务已成功但 ACK 响应丢失”的情况：下一次恢复可能再次拿到同一消息，所以幂等检查必须先于副作用。

成功后 XACK，失败消息不要无限循环

处理成功后，用 XACK 从该消费者组的 PEL 中移除记录：

# 业务持久化成功后确认；XACK 不会自动删除 Stream 中的原始条目
redis-cli XACK orders order-workers 1727510400000-0

XACK 结束的是消费者组对该消息的 pending 跟踪，不等同于从 Stream 删除消息。保留、裁剪和删除 Stream 数据应采用单独的数据生命周期策略。

投递次数超过上限、反复出现相同业务错误的消息，不应永久占用恢复循环。Redis Streams 没有替应用定义统一的死信业务规则；常见做法是把原消息、错误类别、首次/最后失败时间和投递次数写入独立的隔离 Stream，确认隔离写入成功后再 XACK 原消息。隔离操作本身也要幂等。

用反向检查确认恢复完成

恢复结束后重新执行 XPENDING，但不要只看总数。建议同时确认：

• 目标消息已经从 PEL 消失，或明确归属仍在处理的消费者。
• 恢复消费者的 pending 数没有持续增长。
• 高投递次数消息已经进入隔离策略，没有在多个消费者之间反复漂移。
• 业务去重命中数、认领数、成功确认数和隔离数能够对账。
• 新消息消费延迟没有被大规模恢复任务拖高。

恢复任务最好采用小批量、短循环和限速，而不是一次认领全部 PEL。这样更容易保护下游，也能让正常消费者继续工作。

上线前检查清单

• 每条成功消息都在业务持久化后执行 XACK。
• min-idle-time 大于正常处理上界并留有抖动余量。
• 恢复消费者使用唯一名称，小批量扫描 PEL。
• 业务处理具备稳定幂等键，能承受重复投递。
• 设置投递次数或业务错误上限，并有独立隔离 Stream。
• 监控 pending 总量、最大空闲时间、认领量、确认量和隔离量。
• 通过返回游标迭代 XAUTOCLAIM，不假设一次调用覆盖全部 PEL。

相关问题

为什么消费者重启后读不到旧 pending 消息？

使用 XREADGROUP ... > 读取的是未投递的新消息。旧 pending 仍归原消费者所有，需要读取自己的历史或由恢复消费者执行认领。

XAUTOCLAIM 会自动 XACK 吗？

不会。它只转移符合条件的 pending 消息所有权。业务成功后仍需显式 XACK。

pending 数量高就应该马上认领吗？

不一定。大量短空闲 pending 可能表示消费者仍在工作但吞吐不足。先结合空闲时间、消费者心跳和处理延迟判断。