Redis 消费者组怎样自动认领超时消息

我在维护 Redis Stream 消费者组时，最容易遇到的不是“消息读不到”，而是某个消费者拿到消息后进程退出，消息一直留在 Pending Entries List（PEL）里。这个场景不适合每次都先查一遍所有 pending ID，再拼一长串 XCLAIM。更直接的做法是用 XAUTOCLAIM 按最小空闲时间扫描 PEL，让恢复消费者接管符合条件的消息。

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

关键点只有三个：min-idle-time 决定消息多久没有被处理才可认领，start 是 PEL 扫描游标，返回值中的下一游标必须被保存并用于下一轮。认领成功后仍要等业务处理完成，再用 XACK 删除 pending 记录。

先分清新消息、PEL 与超时消息

XREADGROUP 以消费者组读取消息时，Redis 会把已投递但尚未确认的消息记入 PEL。正常消费者完成业务后调用 XACK，对应记录才会从 PEL 中移除；因此，“消息在 PEL 中”只说明它还没有被确认，不等于它一定已经失败。

XPENDING 适合观察数量、消费者和空闲时间，XCLAIM 适合按明确 ID 转移所有权，而 XAUTOCLAIM 把“扫描符合条件的 pending 消息”和“转移所有权”合成了带游标的操作。Redis 官方文档标注它从 Redis Open Source 6.2.0 起提供。

我会把恢复消费者看成一个新的 Consumer，而不是直接“抢回”原消费者的身份。这样做的边界更清楚：它只接管已经空闲超过阈值的消息，业务处理成功后仍由当前消费者确认。

用 XAUTOCLAIM 认领一批空闲消息

命令的核心形态是 XAUTOCLAIM key group consumer min-idle-time start [COUNT count] [JUSTID]。下面的示例从 PEL 起点开始，尝试把空闲至少 60 秒的消息交给恢复消费者：

# 从 PEL 的起点扫描，最多尝试处理 25 条空闲消息
redis-cli XAUTOCLAIM orders workers worker-recover-1 60000 0-0 COUNT 25

# 只取消息 ID，适合先做轻量盘点；随后仍需按 ID 读取并处理业务
redis-cli XAUTOCLAIM orders workers worker-recover-1 60000 0-0 COUNT 25 JUSTID

这里的 60000 是毫秒，不是重试次数。COUNT 是单次尝试扫描的上限，实际认领数量可能更少，因为 Redis 还要过滤空闲时间不足的条目。JUSTID 只返回 ID，并且不会增加该消息的 attempted deliveries 计数；需要完整字段时不要误用它。

认领会重置消息的空闲时间，并把所有权转给指定 Consumer。它不是业务幂等性的替代品：恢复消费者仍可能在执行一半时再次退出，因此处理函数要能安全重试，成功确认也要放在业务动作之后。

循环使用返回游标

XAUTOCLAIM 的返回值可以按三段理解：下一次调用要使用的 stream ID、已成功认领的消息、以及已经不在 Stream 中但仍残留在 PEL 的消息 ID。第一次通常从 0-0 开始，后续调用使用返回的下一游标。

当下一游标返回 0-0，说明本次扫描已经绕过当前 PEL 的末端；它不代表以后不会出现新的超时消息。定时恢复任务可以从 0-0 重新开始下一轮扫描，并按业务负载设置间隔。返回的 deleted IDs 则应作为清理记录或指标观察，不要把它们当成仍可处理的消息。

// XAutoClaimLoop 展示恢复消费者如何保存 Redis 返回的下一游标。
func XAutoClaimLoop(ctx context.Context, rdb *redis.Client) error {
	start := "0-0"
	for {
		next, messages, deleted, err := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{
			Stream:   "orders",             // Stream 的键名
			Group:    "workers",             // 已存在的消费者组
			Consumer: "worker-recover-1",    // 当前恢复消费者身份
			MinIdle:  time.Minute,            // 只接管空闲超过一分钟的消息
			Start:    start,                  // 使用上一轮返回的游标
			Count:    25,                     // 控制单轮扫描规模
		}).Result()
		if err != nil {
			return err
		}

		for _, message := range messages {
			// 先完成可重试的业务处理，再确认消息已经被当前消费者接管。
			if err := handleOrder(ctx, message); err != nil {
				continue // 留在 PEL 中，等待下一轮恢复
			}
			if err := rdb.XAck(ctx, "orders", "workers", message.ID).Err(); err != nil {
				return err
			}
		}

		_ = deleted // 已被裁剪或删除的 ID 只用于指标，不再执行订单处理。
		start = next
		if next == "0-0" {
			return nil // 本轮已扫描到末端，下一次定时任务重新从 0-0 开始。
		}
	}
}

示例中的 handleOrder 只是业务处理占位。真正接入时，建议把 next 记录在本轮内存状态中，并为每轮认领数量、处理失败数、deleted IDs 数量和重复尝试次数建立指标。不要把游标永久当作可靠 checkpoint：PEL 可能持续变化，定时任务通常应在一轮结束后重新从 0-0 扫描。

把认领放进恢复循环

我实际落地时会把恢复任务和正常消费任务分开。正常消费者继续用 XREADGROUP ... > 读取新消息，恢复消费者只负责按 idle 条件找回失联消息；两者共享同一个消费者组，但使用不同的 consumer 名称。这样既不会用恢复逻辑替代正常消费，也能让处理中的消息保留清晰的责任边界。

阈值不要随手写成一个很小的数字。它至少要覆盖正常业务的最长处理时间、一次部署或短暂网络抖动的余量，否则两个健康消费者可能先后处理同一条消息。另一方面，阈值过大又会让故障消息长时间无人接管。可以先用 XPENDING 观察各消费者的 idle 分布，再结合业务重试策略设置。

常见问题

返回 0-0 后还要不要继续调用？

当前扫描轮次可以结束，但后续定时恢复任务仍应重新从 0-0 开始。官方文档还提醒，如果扫描期间又经过了足够时间，原来不满足 idle 条件的消息可能已经变得可认领。

认领后是不是可以立刻 XACK？

不建议。XACK 表示业务已经完成并且消息不再需要消费者组追踪；如果刚认领就确认，进程随后退出时消息可能已经从 PEL 中消失。应先完成幂等业务，再确认。

deleted IDs 是业务失败吗？

不是。它表示扫描到的 ID 已经不在 Stream 中，Redis 同时把它从对应 PEL 清掉。把数量记录下来有助于发现裁剪策略与消费恢复之间的关系，但不能再次读取这些消息。

总结一下：XAUTOCLAIM 解决的是消费者组中“消息还在 PEL、原消费者却长期不再处理”的所有权恢复问题。把 idle 阈值、游标推进、业务幂等和最终 XACK 放在同一个设计里，恢复任务才不会从“没人处理”变成“重复处理且无法判断”。