Redis Stream 怎样处理 pending 消息转移

Redis Stream 消费组里的 pending 消息，不能靠再次从 > 读取来找回。它们已经投递给某个消费者，只是还没有被 XACK 确认。处理这类消息的通用做法是：先用 XPENDING 了解积压和空闲时间，再用 XAUTOCLAIM 把超过业务阈值的消息转给恢复消费者，业务成功后仍由恢复消费者执行 XACK。

官方命令地址：https://redis.io/docs/latest/commands/xautoclaim/

一、先区分 pending、空闲时间和新的消费位置

消费组把已经交给消费者但还没有确认的消息放进 Pending Entries List，也就是 PEL。消息可能因为消费者崩溃、网络中断、处理超时而长期留在这里。它和“Stream 里还没有被任何消费者读过的消息”不是一回事。

恢复任务的第一个判断是 idle 时间，而不是消息 ID 新旧。min-idle-time 表示消息至少空闲了多少毫秒；阈值太小，正常的慢任务可能被第二个消费者抢走；阈值太大，故障消息又会在 PEL 里积压。这个门槛应该参考业务处理的 p99 时长、重试间隔和消费者重启时间。

可以把它理解为两段责任：XAUTOCLAIM 负责把“符合空闲条件的待处理消息”交给新的消费者，业务代码负责真正处理，XACK 负责在成功后从消费组的待确认集合中移除。接管动作本身不代表业务已经完成。

二、用 XPENDING 先判断是否值得接管

不要把恢复任务写成定时无条件抢消息。先观察目标消费组的 PEL，再决定本轮扫描的阈值和批量大小。最小查询可以看到 pending 总数以及最早、最晚的消息 ID：

# 查看消费组的 pending 总数、最小 ID 和最大 ID；只读观察，不会接管消息
redis-cli XPENDING orders-stream order-workers

# 按消费者查看 pending 数量和 idle 范围，帮助判断是否是单个消费者失联
redis-cli XPENDING orders-stream order-workers - + 20 worker-a

不同客户端对 XPENDING 的详细返回会包装成不同结构，但判断思路相同：总量说明是否存在积压，消费者维度帮助定位故障归属，idle 范围帮助选择阈值。假设正常处理最多需要 30 秒，恢复阈值可以先从 120 秒开始，再根据重试和重复处理指标调整；不要把这个示例数字直接当成所有业务的标准。

如果 PEL 为空，应该结束本次恢复，而不是直接把 XREADGROUP 的起点改成旧 ID。尚未投递的消息仍然应该由正常消费者使用 > 读取；恢复路径只处理已进入 PEL 的消息。

三、用 XAUTOCLAIM 按游标转移长时间空闲消息

XAUTOCLAIM 的命令形式是 XAUTOCLAIM key group consumer min-idle-time start [COUNT count] [JUSTID]。consumer 是接管后的新所有者，start 是扫描 PEL 的起点，第一次通常从 0-0 开始。COUNT 是本次尝试处理的上限，不保证一定能接管同样数量的消息，因为其中一些消息可能还没有达到 idle 阈值。

# 从 PEL 的开头扫描；只把空闲至少 120 秒的消息交给 recovery-1
# COUNT 25 限制本次尝试规模，避免恢复任务一次占满连接和业务线程
redis-cli XAUTOCLAIM orders-stream order-workers recovery-1 120000 0-0 COUNT 25

返回结果包含三部分：下一次调用应使用的游标、成功接管的消息数组，以及已经不在 Stream 中但仍残留在 PEL 的消息 ID。Redis 7.0 起第三部分用于报告这类被清理的幽灵 ID。不要只取第二部分就丢弃其余信息，否则无法知道扫描是否完成，也无法统计数据被裁剪或删除的情况。

下面用 Python 的伪生产逻辑展示游标循环。示例使用 redis-py 的响应结构表达意图，实际项目应按客户端是否返回 bytes、是否启用 RESP3 适配字段类型。代码不会把处理成功假设为接管成功：

from redis import Redis

def reclaim_pending(redis: Redis, stream: str, group: str, consumer: str) -> dict:
    # 阈值要高于正常处理时长；这里的120000表示120秒
    min_idle_ms = 120_000
    cursor = "0-0"
    claimed = 0
    deleted = 0

    while True:
        # 返回值依次是下一游标、已接管消息、已从PEL清理的不存在消息ID
        next_cursor, messages, deleted_ids = redis.xautoclaim(
            stream, group, consumer, min_idle_ms, cursor, count=25
        )
        deleted += len(deleted_ids)

        # 只有业务处理成功才确认；失败分支保留消息供后续重试或隔离
        for message_id, fields in messages:
            if process_one(message_id, fields):
                redis.xack(stream, group, message_id)
                claimed += 1

        # 0-0表示本轮已经扫完当前PEL；下一轮定时任务可以从头再扫
        if next_cursor == "0-0":
            break
        cursor = next_cursor

    return {"acked": claimed, "deleted_from_pel": deleted}

这里的 process_one 代表业务处理函数，不应在失败时无条件 XACK。如果处理函数抛出异常，应该记录消息 ID、重试次数和异常分类，并交给下一轮恢复或隔离策略；否则消息虽然看似“恢复完成”，实际上已经从可靠处理链路中消失。

四、把转移、业务处理和 XACK 分成两个责任

消息转移解决的是“谁来继续处理”，确认解决的是“业务是否已经完成”。这两个动作不能合并成一个成功标记。一个恢复消费者接管消息后，如果在调用外部服务前就执行 XACK，进程随后崩溃，Redis 就不会再把这条消息视为待确认。

更稳妥的顺序是：接管消息、做幂等检查、执行业务、持久化业务结果、最后 XACK。外部副作用无法和 Redis 的确认放进同一个事务时，必须用业务幂等键、结果表或去重记录承受“业务已成功但确认尚未发出”的重复投递窗口。

def process_one(message_id: str, fields: dict) -> bool:
    # 用Stream消息ID作为幂等键，避免接管后的重复投递重复产生外部副作用
    if already_completed(message_id):
        return True

    try:
        # 先完成业务写入；这里应使用调用方自己的超时、重试和幂等策略
        save_order_result(fields)
        mark_completed(message_id)
        return True
    except TemporaryError:
        # 临时错误不确认，让消息保留在PEL中等待下一次恢复
        return False
    except PermanentError:
        # 永久错误进入隔离记录后再确认，避免同一坏消息无限阻塞恢复循环
        save_dead_letter(message_id, fields)
        return True

永久错误是否可以在隔离后确认，取决于你的补偿流程；关键是隔离记录必须先写成功。对于反复崩溃的消息，还要观察 attempted deliveries，必要时给重试次数设置告警或转人工处理，不能只依赖“下一次再抢”。

五、处理 Redis 版本和异常边界

XAUTOCLAIM 从 Redis Open Source 6.2.0 开始提供。服务端若返回 NOGROUP，说明目标消费组不存在或名称不匹配，应该修正组初始化流程，而不是不断重试相同命令。JUSTID 只返回成功接管的消息 ID，不返回字段，并且不会增加消息的重试计数，适合先做轻量扫描，但后续还需要用 ID 取回消息内容。

当返回游标为 0-0 时，表示当前扫描已经走到 PEL 的末尾，不表示未来不会出现新的可接管消息。下一次定时恢复可以重新从 0-0 开始；如果本轮运行时间很长，也要结合业务窗口决定是否再次扫一遍，因为之前还没达到阈值的消息可能已经变得符合条件。

还要记录第三段返回中的已删除 ID：消息可能因为 XDEL 或裁剪已经不存在，但它的 ID 仍在 PEL 中。Redis 会在扫描时把这类 ID 从对应 PEL 清理掉；这不是业务成功处理，也不应计入成功确认数量。建议把接管、确认、隔离、PEL 清理和命令错误分成独立指标。

相关问题

XAUTOCLAIM 会把所有 pending 消息都转走吗？

不会。只有 idle 时间超过 min-idle-time、并且位于本次游标扫描范围内的消息才会被尝试接管；COUNT 还会限制每次扫描规模。

接管后还需要 XACK 吗？

需要。接管只是改变消费组中的消费者归属，业务成功后仍要对原 Stream 和消费组执行 XACK。

为什么 COUNT 25 最后只接管了几条？

COUNT 是尝试上限，不是保证数量。扫描到的消息可能仍未达到 idle 门槛，也可能在 Stream 中已经不存在。

能不能直接把 XREADGROUP 的起点改回旧 ID？

不能把它当成 PEL 恢复方案。已投递消息的所有权和确认状态由消费组维护，应使用 XPENDING 与 XAUTOCLAIM 处理；新的未投递消息仍走 >。

落地时可以用一句话检查方案是否完整：XPENDING 负责发现，XAUTOCLAIM 负责转移，业务代码负责成功与否，XACK 负责最后确认。只要把这四个责任拆开，Redis Stream 的 pending 消息就不会因为消费者故障而只能人工翻 ID 恢复。