消费者会有一个状态变量，用于记录被当前消费已读取但未ack的消息Id，如果客户端没有ack，这个变量里面的消息ID会越来越多，一旦某个消息被ack它就开始减少。这个pending_ids变量在Redis官方被称之为 PEL(Pending Entries List)，记录了当前已经被客户端读取的消息，但是还没有 ack (Acknowledge character：确认字符），它用来确保客户端至少消费了消息一次，而不会在网络传输的中途丢失了没处理

stream的常用指令

常用命令一览：

队列相关指令

消费者相关指令

四个特殊符号：

xadd命令

语法格式为：

1
2
3
4

127.0.0.1:6379> xadd message 1 key1 value1 key2 value2
"1-0"
127.0.0.1:6379> xadd message 1-2 key1 value1 key2 value2
"1-2"

ID 在一般情况下是由 redis 自动指定的，但其实 ID 也是可以自定义的，为了保证 ID 的自增状态，手动指定的 ID 必须要大于系统中存在的 ID，只不过一般不这么做。

1 2	127.0.0.1:6379> xadd message * key1 value1 key2 value2 "1604475735664-0"

以上命令添加了 key1-value 和 key2-value2 两条消息到 message 这个 key 中，返回值为当前消息的 ID，由 redis 自动生成，此时消息队列中就有一条消息可以被读取了。

1 2	127.0.0.1:6379> xadd message maxlen 10 * key3 value3 "1604476672762-0"

xread命令

语法格式为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

以上命令在非阻塞模式输出了所有的消息，因为不存在 ID 比 0 还小的消息，所以输出了所有的消息。

阻塞模式中，可以使用 $ 符号来获取最新的消息。如果在指定超时时间内没有新的消息，则返回空。

1
2
3
4
5
6
7
8
9

127.0.0.1:6379> xread count 10 block 10000 streams message $
(nil)
(10.02s)
127.0.0.1:6379> xread count 10 block 10000 streams message $
1) 1) "message"
2) 1) 1) "1604478070071-0"
2) 1) "keyblock2"
2) "value"
(5.00s)

输入命令后，可以观察到命令没有任何输出，此时新开一个 redis-cli，输入 xadd message * keyblock2 value 将一条新的消息添加到 key 中，可以看到上面的命令返回了刚才添加的值和阻塞时间。

xlen命令

语法格式：

返回 key 中消息的数量，如果 key 不存在，则会返回 0。即使 key 中消息的数量为 0，key 也不会被自动删除，因为可能还存在和 key 关联的消费者组。

1 2	127.0.0.1:6379> xlen message (integer) 5

返回 message 中所有消息的数量。

xrange命令

语法如下：

该命令返回与给定的 ID 范围相匹配的消息。ID 的范围由 start 和 end 参数来指定。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

127.0.0.1:6379> xrange message - +
1) 1) "1-0"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"
2) 1) "1-2"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"
3) 1) "1604476672762-0"
2) 1) "key3"
2) "value3"
4) 1) "1604478059261-0"
2) 1) "keyblock"
2) "value"
5) 1) "1604478070071-0"
2) 1) "keyblock2"
2) "value"

此命令由两个特殊的 ID，使用 - 表示最小的 ID 值，使用 + 表示最大的 ID 值，可以查询所有的消息。

1
2
3
4
5
6
7
8
9
10
11

127.0.0.1:6379> xrange message 1 1
1) 1) "1-0"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"
2) 1) "1-2"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"

即使 ID 不完整也可以，只输入 ID 值会输出所有拥有相同 ID 不同序列号的消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

127.0.0.1:6379> xrange message - + count 3
1) 1) "1-0"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"
2) 1) "1-2"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"
3) 1) "1604476672762-0"
2) 1) "key3"
2) "value3"

使用 count 参数可以限制输出消息的数量。

通过简单的循环可以使用少量内存迭代一个 key 中所有的值，只需要将上次迭代的结果中最大的 ID 作为下一次迭代的起始 ID 即可。

xrevrange命令

语法说明：

xrevrange 命令和 xrange 命令语法完全相同，只有一点不同，xrevrange 是反向遍历的，不再赘述。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

127.0.0.1:6379> xrevrange message + -
1) 1) "1604478070071-0"
2) 1) "keyblock2"
2) "value"
2) 1) "1604478059261-0"
2) 1) "keyblock"
2) "value"
3) 1) "1604476672762-0"
2) 1) "key3"
2) "value3"
4) 1) "1-2"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"
5) 1) "1-0"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"

xtrim命令

语法说明：

xtrim 命令会从 ID 值比较小的消息开始丢弃。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

127.0.0.1:6379> xread streams message 0
1) 1) "message"
2) 1) 1) "1-0"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"
2) 1) "1-2"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"
3) 1) "1604476672762-0"
2) 1) "key3"
2) "value3"
4) 1) "1604478059261-0"
2) 1) "keyblock"
2) "value"
5) 1) "1604478070071-0"
2) 1) "keyblock2"
2) "value"
127.0.0.1:6379> xtrim message maxlen 4
(integer) 1

xtrim 命令的返回值是修剪掉的 ID 的数量。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

127.0.0.1:6379> xrange message - +
1) 1) "1-2"
2) 1) "key1"
2) "value1"
3) "key2"
4) "value2"
2) 1) "1604476672762-0"
2) 1) "key3"
2) "value3"
3) 1) "1604478059261-0"
2) 1) "keyblock"
2) "value"
4) 1) "1604478070071-0"
2) 1) "keyblock2"
2) "value"

再次查看可以看到 key 中的消息数量被修剪掉了一个，只剩下了四个。

1 2	127.0.0.1:6379> xtrim message maxlen ~ 2 (integer) 0

如果使用了 ~ 参数，则可能不会进行修剪。此参数告诉 redis 在能够删除整个宏节点时才执行修剪，这样做效率更高，并且可以保证消息的数量不小于所需要的数量。

xdel命令

语法说明：

xdel 命令用于从 key 中删除指定 ID 的消息，当 ID 不存在时，返回的数目可能和删除的数目不一致。在执行 xdel 命令时，redis 并不会在内存中删除对应的消息，而只会把它标记为删除，在所有节点都被删除之后整个节点被销毁，内存被回收。

1
2
3
4
5
6
7
8
9
10
11
12

127.0.0.1:6379> xdel message 1-2
(integer) 1
127.0.0.1:6379> xrange message - +
1) 1) "1604476672762-0"
2) 1) "key3"
2) "value3"
2) 1) "1604478059261-0"
2) 1) "keyblock"
2) "value"
3) 1) "1604478070071-0"
2) 1) "keyblock2"
2) "value"

删除了 ID 为 1-2 的消息。

xgroup命令

语法说明：

xgroup 是一个命令组，可以通过不同的关键字执行不同的命令。

1
2
3
4
5
6
7
8
9
10
11

127.0.0.1:6379> xgroup create message read_group $
OK
127.0.0.1:6379> xreadgroup group read_group read streams message >
(nil)
127.0.0.1:6379> xadd message * readkey readvalue
"1604494179721-0"
127.0.0.1:6379> xreadgroup group read_group read streams message >
1) 1) "message"
2) 1) 1) "1604494179721-0"
2) 1) "readkey"
2) "readvalue"

使用 creat 命令创建一个 read_group 分组，指定 ID 的起点为最后一个 ID，直接读取的话是空值（xreadgroup命令下面说）。

使用 xadd 命令添加一条新的消息，再次读取发现可以正常读取到新加入的消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

127.0.0.1:6379> xgroup setid message read_group 0
OK
127.0.0.1:6379> xreadgroup group read_group read streams message >
1) 1) "message"
2) 1) 1) "1604476672762-0"
2) 1) "key3"
2) "value3"
2) 1) "1604478059261-0"
2) 1) "keyblock"
2) "value"
3) 1) "1604478070071-0"
2) 1) "keyblock2"
2) "value"
4) 1) "1604494179721-0"
2) 1) "readkey"
2) "readvalue"

使用 setid 命令重新设置读取 ID 的起点，可以读取到所有的历史消息。

1
2
3
4
5
6
7
8
9

127.0.0.1:6379> xinfo groups message
1) 1) "name"
2) "read_group"
3) "consumers"
4) (integer) 1
5) "pending"
6) (integer) 4
7) "last-delivered-id"
8) "1604494179721-0"

使用 xinfo 命令查询新创建的分组信息，可以看到分组名字，消费者数量，最后加入的消息的 ID 值（xinfo 命令下边说）。

1
2
3
4
5
6
7

127.0.0.1:6379> xinfo consumers message read_group
1) 1) "name"
2) "read"
3) "pending"
4) (integer) 4
5) "idle"
6) (integer) 476449

使用 xinfo 命令查看新加入的消费者的信息，可以看到消费者名字，处于 pending（待处理）状态的消息数量（pending 状态下边说）。

1
2
3
4

127.0.0.1:6379> xgroup delconsumer message read_group read
(integer) 4
127.0.0.1:6379> xinfo consumers message read_group
(empty array)

使用 delconsumer 命令删除分组中的消费者，使用 xinfo 命令查看分组中的消费者，返回一个空数组，说明删除成功。delconsumer 命令的返回值为当前消费者所拥有的 pending（待处理）状态的消息数量。

1
2
3
4

127.0.0.1:6379> xgroup destroy message read_group
(integer) 1
127.0.0.1:6379> xinfo groups message
(empty array)

使用 destroy 命令删除 message 中的分组，使用 xinfo 命令查看，返回一个空数组，说明删除成功。destroy 命令的返回值为删除成功的分组数量。注意：即使由活跃的消费者和 pending（待处理）状态的消息，分组仍然会被删除，需要确保在需要时才执行此命令。

xinfo命令

语法说明：

1
2
3
4
5
6
7

127.0.0.1:6379> xinfo consumers message read_group
1) 1) "name"
2) "read"
3) "pending"
4) (integer) 4
5) "idle"
6) (integer) 206796

读取 message 的 read_group 分组中所有的消费者信息。

1
2
3
4
5
6
7
8
9

127.0.0.1:6379> xinfo groups message
1) 1) "name"
2) "read_group"
3) "consumers"
4) (integer) 1
5) "pending"
6) (integer) 4
7) "last-delivered-id"
8) "1604494179721-0"

读取 message 中所有的分组信息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

127.0.0.1:6379> xinfo stream message
1) "length"
2) (integer) 4
3) "radix-tree-keys"
4) (integer) 1
5) "radix-tree-nodes"
6) (integer) 2
7) "last-generated-id"
8) "1604494179721-0"
9) "groups"
10) (integer) 1
11) "first-entry"
12) 1) "1604476672762-0"
2) 1) "key3"
2) "value3"
13) "last-entry"
14) 1) "1604494179721-0"
2) 1) "readkey"
2) "readvalue"

读取 message 所有的信息。

xpending命令

语法说明：

1
2
3
4
5
6

127.0.0.1:6379> xpending message readgroup
1) (integer) 2
2) "1604496633846-0"
3) "1604496640734-0"
4) 1) 1) "read"
2) "2"

xpending 命令可以查看对应分组中未确认的消息的数量和其所对应的消费者的名字还有起始和终止 ID。

1
2
3
4
5
6
7
8
9

127.0.0.1:6379> xpending message readgroup - + 10 read
1) 1) "1604496633846-0"
2) "read"
3) (integer) 513557
4) (integer) 4
2) 1) "1604496640734-0"
2) "read"
3) (integer) 482927
4) (integer) 1

使用 xpending 命令可以查看处于未确认状态的消息的具体信息。

xreadgroup命令

语法说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

127.0.0.1:6379> flushdb
OK
127.0.0.1:6379> xadd message * key1 value1
"1604496633846-0"
127.0.0.1:6379> xadd message * key2 value2
"1604496640734-0"
127.0.0.1:6379> xgroup create message readgroup 0
OK
127.0.0.1:6379> xadd message * key3 value3
"1604496696501-0"
127.0.0.1:6379> xadd message * key4 value4
"1604496704823-0"
127.0.0.1:6379> xreadgroup group readgroup read count 1 streams message >
1) 1) "message"
2) 1) 1) "1604496633846-0"
2) 1) "key1"
2) "value1"
127.0.0.1:6379> xreadgroup group readgroup read count 1 streams message >
1) 1) "message"
2) 1) 1) "1604496640734-0"
2) 1) "key2"
2) "value2"

从头开始，清空数据库，重新给 message 添加消息，创建分组，使用 readgroup 命令读取最新的消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

127.0.0.1:6379> xinfo consumers message readgroup
1) 1) "name"
2) "read"
3) "pending"
4) (integer) 2
5) "idle"
6) (integer) 53146
127.0.0.1:6379> xpending message readgroup - + 10 read
1) 1) "1604496633846-0"
2) "read"
3) (integer) 513557
4) (integer) 4
2) 1) "1604496640734-0"
2) "read"
3) (integer) 482927
4) (integer) 1

读取了两条消息，使用 xinfo 命令查看，可以看到有两条消息处于 pending（待处理）状态，使用 xpending 命令可以查看处于未确认状态的消息的具体信息。

xack命令

语法说明：

xack 命令从 pending 队列中删除挂起的消息，也就是确认之前未确认的消息。当使用 xreadgroup 命令读取消息时，消息同时被存储到 PEL 中，等待被确认，调用 xack 命令可以从 PEL 中删除挂起的消息并且释放内存，确保不丢失消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

127.0.0.1:6379> xack message readgroup 1604496633846-0
(integer) 1
127.0.0.1:6379> xpending message readgroup
1) (integer) 1
2) "1604496640734-0"
3) "1604496640734-0"
4) 1) 1) "read"
2) "1"
127.0.0.1:6379> xinfo consumers message readgroup
1) 1) "name"
2) "read"
3) "pending"
4) (integer) 1
5) "idle"
6) (integer) 418489

使用 xack 命令确认一条消息，再次使用 xpending 命令查看未确认消息的数量，只剩一条未确认消息，使用 xinfo 命令查看处于 pending 状态的消息数量也为 1，确认消息成功。

xclaim命令

语法说明：

xclaim 命令用于更改未确认消息的所有权，如果有消费者在读取了消息之后未处理完成就挂掉了，那么消息会一直在 pending 队列中，占用内存，这时需要使用 xclaim 命令更改此条消息的所属者，让其他的消费者去消费这条消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

127.0.0.1:6379> xreadgroup group readgroup read2 count 1 streams message >
1) 1) "message"
2) 1) 1) "1604496696501-0"
2) 1) "key3"
2) "value3"
127.0.0.1:6379> xack message readgroup 1604496696501-0
(integer) 1
127.0.0.1:6379> xpending message readgroup - + 10 read
1) 1) "1604496640734-0"
2) "read"
3) (integer) 1258517
4) (integer) 2
127.0.0.1:6379> xpending message readgroup - + 10 read2
(empty array)

新创建一个消费者，读取一条消息，然后将消息确认掉，查看两个消费者中未确认消息的数量，read 有一条，read2 没有未确认的消息。

1
2
3
4
5
6
7
8
9
10
11

127.0.0.1:6379> xclaim message readgroup read2 0 1604496640734-0
1) 1) "1604496640734-0"
2) 1) "key2"
2) "value2"
127.0.0.1:6379> xpending message readgroup - + 10 read2
1) 1) "1604496640734-0"
2) "read2"
3) (integer) 3724
4) (integer) 4
127.0.0.1:6379> xpending message readgroup - + 10 read
(empty array)

使用 xclaim 命令将消息所有权转移给 read2 这个消费者，可以看到，消费者 read2 的 pending 队列中有一条未确认消息，消费者 read 的 pending 队列中已经没有消息了。