消费者崩溃后的消息转移

做的事:让一个真实进程崩溃

实测 · 本机执行输出

A 读到 ['1-0', '2-0', '3-0', '4-0', '5-0']
A 已 ack 1-0
A 已 ack 2-0
A 开始处理 3-0 ……
kill 之前 CLIENT LIST 中的 worker-A: 1 个连接
kill 之后 CLIENT LIST 中的 worker-A: 0 个连接

A 是用 subprocess 启动的独立 Python 进程:读 5 条,ack 前 2 条,然后开始「处理」第 3 条。我从外面直接 kill -9。用 CLIENT LIST 数 name=worker-A 的连接:kill 前 1 个,kill 后 0 个。

连接没了,PEL 还在

实测 · 本机执行输出

> XINFO CONSUMERS s g
1) 1) "name"
   2) "A"
   3) "pending"
   4) (integer) 3
   5) "idle"
   6) (integer) 306
   7) "inactive"
   8) (integer) 306
> XPENDING s g - + 10
1) 1) "3-0"
   2) "A"
   3) (integer) 1309
   4) (integer) 1
2) 1) "4-0"
   2) "A"
   3) (integer) 1309
   4) (integer) 1
3) 1) "5-0"
   2) "A"
   3) (integer) 1309
   4) (integer) 1

消费者 A 依然存在(XINFO CONSUMERS),pending 是 3(3-0、4-0、5-0,前两条已 ack),idle 306 ms。约 1 秒后 XPENDING 明细里这三条消息的 idle 是 1309 ms。注意两个 idle 不是同一个计数:XINFO CONSUMERS 的是「消费者距最后一次活动」,XPENDING 的是「这条消息距上次投递」。这里的关键事实:连接断开不会触发任何重新分配。

新消费者拿不到这些消息

实测 · 本机执行输出

B 用 > 读到的 ID: ['6-0', '7-0', '8-0']

B 用 XREADGROUP … > 读,只拿到游标之后的 6-0、7-0、8-0。这是 > 的定义:只要「从没投递给任何人」的新消息。

XCLAIM:主动认领

实测 · 本机执行输出

> XCLAIM s g B 1000 3-0 4-0 5-0
1) 1) "3-0"
   2) 1) "job"
      2) "j3"
2) 1) "4-0"
   2) 1) "job"
      2) "j4"
3) 1) "5-0"
   2) 1) "job"
      2) "j5"
> XPENDING s g - + 10 B
1) 1) "3-0"
   2) "B"
   3) (integer) 0
   4) (integer) 2
2) 1) "4-0"
   2) "B"
   3) (integer) 0
   4) (integer) 2
3) 1) "5-0"
   2) "B"
   3) (integer) 0
   4) (integer) 2
4) 1) "6-0"
   2) "B"
   3) (integer) 0
   4) (integer) 1
5) 1) "7-0"
   2) "B"
   3) (integer) 0
   4) (integer) 1
6) 1) "8-0"
   2) "B"
   3) (integer) 0
   4) (integer) 1
(投递次数变成 2:每次被认领 +1)

B 认领 A 空闲超过 1000 ms 的三条:成功,消息内容一并返回。认领后 XPENDING 显示它们的消费者变成 B、idle 归零、投递次数变成 2(6、7、8 号是 B 自己刚读的,所以次数是 1)。

实测 · 本机执行输出

> XCLAIM s g C 1000 3-0
(empty array)

C 立刻再来认领 3-0:返回空——3-0 刚被 B 认领,idle 只有几毫秒,不满足 1000 ms 的 min-idle-time。这个参数是防止两个恢复进程同时抢同一条消息的保护。

危险:XGROUP DELCONSUMER

实测 · 本机执行输出

> XGROUP DELCONSUMER s g B
(integer) 6
> XPENDING s g
1) (integer) 0
2) (nil)
3) (nil)
4) (nil)
> XRANGE s 3-0 3-0
1) 1) "3-0"
   2) 1) "job"
      2) "j3"
(返回值是被丢弃的 PEL 个数;消息还在流里,但已经没有任何组的 PEL 记着它没处理完)

删除消费者 B 时返回 6,表示丢弃了 B 名下 6 条 PEL 记录;之后 XPENDING s g 是 0。消息本身还在流里(XRANGE 读得到 3-0),但组里已经没有任何记录说它「还没处理完」。下线消费者之前,应先把它的 PEL 认领走再删。

分步演示:崩溃后消息的去向

消费者崩溃

本步变化的消息

本文用到的工具与边界

本机 macOS 27 / Redis 8.10.2 独立实例(本机回环端口,save 关闭),redis-cli 风格输出由自写的 RESP 客户端打印;计时用 time.perf_counter,数字每次运行会有小幅抖动。A 是独立 Python 进程,被 SIGKILL;其余操作在同一个测试脚本里串行完成
没有验证:客户端库自带的自动重试与重连行为、连接断开时 Redis 是否有 CLIENT 相关的超时清理(只观察了 5 秒内)、消费者数量很多时 XINFO CONSUMERS 的开销

下一篇 · REDIS 45

XAUTOCLAIM