京公网安备11010502063487号Stream 的消费组解决的是「多个消费者分担同一条流」的问题:组内每条消息只交给一个消费者,不同的组则各读一份。这篇在真实实例上把它拆开:组的游标、消费者怎么出现、消息怎么分配、几个 XINFO 字段的含义,以及创建时最容易踩的坑。
> XGROUP CREATE s gnew $ "OK" > XGROUP CREATE s gall 0 "OK" > XGROUP CREATE nokey g $ (error) ERR The XGROUP subcommand requires the key to exist. Note that for CREATE you may want to use the MKSTREAM option to create an empty stream automatically. > XGROUP CREATE nokey g $ MKSTREAM "OK" > XGROUP CREATE s gall 0 (error) BUSYGROUP Consumer Group name already exists
$ 表示「当前最后一条之后」,创建后才写入的消息才会被这个组读到;0 表示从头开始。key 不存在时 XGROUP CREATE 会报错,要加 MKSTREAM 才会连流一起创建;同名组再创建报 BUSYGROUP。
c1 ['1-0', '4-0', '5-0', '10-0'] c2 ['2-0', '6-0', '9-0', '11-0'] c3 ['3-0', '7-0', '8-0', '12-0'] > XREADGROUP GROUP gall c1 COUNT 1 STREAMS s > (nil) (读完了,再读一次返回 nil)
三个消费者 c1、c2、c3 按我指定的顺序(c1 c2 c3 c1 c1 c2 c3 c3 c2 c1 c2 c3)每次读 1 条(XREADGROUP … COUNT 1 … >)。12 条消息被完整分配、没有重复,每人拿到 4 条。这个「平均」是我安排的读取顺序造成的:组本身没有轮询或均衡,谁先来读、谁就拿走游标后面的下一条。
gnew(创建于 $): None XADD 一条之后 gnew 读到: [['s', [['13-0', ['job', 'j13']]]]]
组 gnew 创建于 $:那时流里已有 12 条,它一条也读不到(nil);之后写入 13-0,它才读到。而 gall 从头读过了全部消息——同一条消息被两个组各消费一次,这是 Stream 相对 List 队列的关键差别(见第 48 篇)。
> XINFO GROUPS s
1) 1) "name"
2) "gall"
3) "consumers"
4) (integer) 3
5) "pending"
6) (integer) 12
7) "last-delivered-id"
8) "12-0"
9) "entries-read"
10) (integer) 12
11) "lag"
12) (integer) 1
2) 1) "name"
2) "gnew"
3) "consumers"
4) (integer) 1
5) "pending"
6) (integer) 1
7) "last-delivered-id"
8) "13-0"
9) "entries-read"
10) (integer) 13
11) "lag"
12) (integer) 0
> XINFO CONSUMERS s gall
1) 1) "name"
2) "c1"
3) "pending"
4) (integer) 4
5) "idle"
6) (integer) 0
7) "inactive"
8) (integer) 1
2) 1) "name"
2) "c2"
3) "pending"
4) (integer) 4
5) "idle"
6) (integer) 0
7) "inactive"
8) (integer) 0
3) 1) "name"
2) "c3"
3) "pending"
4) (integer) 4
5) "idle"
6) (integer) 0
7) "inactive"
8) (integer) 0组的字段:last-delivered-id 是游标;pending 是 PEL 里未确认的条数(gall 读了 12 条都没 ack,所以是 12);entries-read 是组读过的条目总数;lag 是流里还没被这个组读到的条数(gall 是 1,因为 13-0 是它之后才写入的;gnew 是 0)。消费者的字段:pending、idle(距上次读的毫秒数)、inactive(距上次真正取到消息的毫秒数)。
> XGROUP CREATECONSUMER s gall c9 (integer) 1 > XGROUP CREATECONSUMER s gall c9 (integer) 0 > XGROUP DESTROY s gnew (integer) 1 > XGROUP DESTROY s gnew (integer) 0
消费者第一次 XREADGROUP 时自动出现;也可以显式 XGROUP CREATECONSUMER(新建返回 1,已存在返回 0)。XGROUP DESTROY 删掉整个组,第二次返回 0。