redis专题-----11-----redis订阅发布以及stream

参考文章：
Redis发布订阅模式（publish/subscribe）。

一订阅发布

1 概念

1）为了支持消息的多播机制，redis 引入了发布订阅模块。Redis发布/订阅（Pub/Sub）是一种通信机制，将数据推到某个信息管道中，其他客户端可通过订阅这些管道来获取推送信息，以此用于消息的传输。
2）发布者发布的消息分到不同的频道，不需要知道什么样的订阅者者订阅。订阅者对一个或多个频道感兴趣，只需要接收感兴趣的消息，不需要知道什么样的发布者发布。主要目的是解除消息的发布者与订阅者之间的耦合关系。
3）发布者和订阅者都是Redis客户端，频道则是服务器端。
4）由三部分组成：发布者（Publisher）、频道（Channel）、订阅者（Subscriber）。

2 原理

2.1 底层结构原理
Redis通过PUBLISH、PUBSUB、SUBSCRIBE、PSUBSCRIBE、UNSUBSCRIBE和PUNSUBSCRIBE等命令实现发布和订阅功能。

在Redis底层结构中，客户端和频道的订阅关系是通过一个字典加链表的结构保存的，如下图：

在 Redis 的底层结构中，redis服务器中定义了一个pubsub_channels字典，用于保存所有频道的订阅关系，在这个字典中，key为所有频道名称，value结构是一个链表，其中存放的是所有订阅这个频道的订阅者客户端。subscribe命令的实质即为在key中添加value的订阅链。

若频道首次被订阅说明在字典中并不存在该渠道的信息，那么程序首先要新建一个对应的 key，并且要赋值一个空链表，然后将对应的客户端加入到链表中。此时链表只有一个元素。若该渠道已经被其他客户端订阅过：这个时候就直接找到key值对应的value客户端信息添加到链表的末尾即可。

2.2 消费者消费消息原理
生产者生产一次消息，由redis负责将消息复制到多个消息队列中，每个消息队列由相应的消费者进行消费。
它是分布式系统中常用的一种解耦方式，用于将多个消费者的逻辑进行拆分。多个消费者的逻辑就可以放到不同的子系统中完成。

看下图，此时m1队列可以认为是上面结构原理的频道，多个subscriber及其内部的队列相当于上面结构原理的客户端链表。

3 订阅时服务器推送的消息格式

所有订阅接收的消息均为由三个元素组成的多块响应，其中第一个元素是消息类型，有四种类型：

1）subscribe：该类型表示成功订阅到频道响应，第一个元素是消息类型，第二个元素为订阅的频道名称，第三个元素为已订阅的频道数量，例：
2）unsubscribe：该类型表示成功取消订阅到的频道响应，第一个元素是消息类型，第二个元素为订阅的频道名称，第三个元素为已订阅的频道数量，例：
3）message：该类型表示订阅者客户端接收到其他客户端发出的发布命令结果，第一个元素是消息类型，第二个元素表示来源频道的名称，第三个元素是实际的消息内容，例：
首先在一个redis客户端连接随便订阅一到多个通道，我这里订阅两个。

然后开启一个新的客户端连接。用于往上面的某个通道发布消息。此时原来的客户端即可看到发布的消息。
4）pmessage：这个类型是使用订阅模式频道时，redis返回推送消息的类型。以下面的例子的截图为例，第一行元素是消息类型，第二个是订阅模式，第三个是具体的订阅频道，第四个是时基的消息内容。和message相比，只多了第二行的订阅模式的信息返回。

4 发布订阅命令

4.1 发布订阅命令语法

1. 订阅频道
SUBSCRIBE命令为订阅频道，返回值如上面的消息格式所述。

SUBSCRIBE channel [channel ...]

2. 订阅模式频道
PSUBSCRIBE是订阅给定的模式(patterns)。

PSUBSCRIBE pattern [pattern ...]

例如：

h?llo subscribes to hello, hallo and hxllo
h*llo subscribes to hllo and heeeello
h[ae]llo subscribes to hello and hallo, but not hillo
如果想输入普通的字符，可以在前面添加\

3. 取消订阅频道
指示客户端退订给定的频道，若没有指定频道，则退订所有频道。

如果没有频道被指定，即，一个无参数的 UNSUBSCRIBE 调用被执行，那么客户端使用 SUBSCRIBE 命令订阅的所有频道都会被退订。在这种情况下，命令会返回一个信息，告知客户端所有被退订的频道。

UNSUBSCRIBE [channel [channel ...]]

4. 取消订阅模式频道

指示客户端退订指定模式，若果没有提供模式则退出所有模式。

如果没有模式被指定，即一个无参数的 PUNSUBSCRIBE 调用被执行，那么客户端使用 PSUBSCRIBE 命令订阅的所有模式都会被退订。在这种情况下，命令会返回一个信息，告知客户端所有被退订的模式。

PUNSUBSCRIBE [pattern [pattern ...]]

5. 发布具体频道或模式频道的内容
将信息 message 发送到指定的频道channel。

返回值integer-reply: 收到消息的客户端数量。

PUBLISH channel message

6. pubsub
PUBSUB 是自省命令，能够检测PUB/SUB子系统的状态。它由分别详细描述的子命令组成。
返回值array-reply: 活跃的信道列表，或者符合指定模式的信道。

PUBSUB subcommand [argument [argument ...]]

4.2 发布订阅命令例子

简单的发布订阅命令就不列举了，主要使用订阅模式频道以及发布模式频道来举例。
例子1：

# 1. 首先一个客户端先订阅模式频道PSUBSCRIBE new.*

# 2. 然后开启另一个客户端，发布模式频道消息。
publish new.showbiz 'c++ c go'

结果如下：

例子2：多个客户端订阅，其中一个客户端订阅方式为订阅频道，另一个客户端订阅方式为订阅模式频道。

这个例子想说明的是：当不同的订阅客户端使用订阅某种模式或者符合该模式的具体某个频道时，那么这些客户端都会接收到发布者推送的信息，但是两次接收的信息格式不同，一个为message类型，另一个为pmessage类型，消息内容一致。
例子和上面例子1差不多。

# 1. 首先一个客户端先订阅模式频道PSUBSCRIBE news.*

# 2. 然后再开启另一个客户端，但是使用订阅频道subscribe news.a news.b news.c

# 3. 然后开启另一个客户端，发布模式频道消息。总共使用了3个客户端。
publish news.a 'c++ c go'

下面看到，两个客户端都能收到发布的消息。

例子3：
这个想说明的是pubsub命令的简单使用。
下面用到两个客户端，直接看图操作即可。

关于更详细的发布订阅命令说明，请参考中文手册：http://redis.cn/commands.html。想要看什么命令，直接在 “直接搜索” 中输入命令即可。

5 注意

发布订阅功能一般要区别命令连接重新开启一个连接，因为命令连接严格遵循请求回应模式。
而pub/sub能收到redis主动推送的内容，所以实际项目中如果支持pub/sub的话，需要另开一条连接用于处理发布订阅。

6 缺点

发布订阅的生产者传递过来一个消息，redis会直接找到相应的消费者并传递过去。这个传递可能发生以下情况：

1）假如没有消费者，消息直接丢弃。
2）假如开始有2个消费者，一个消费者突然挂掉了，另外一个消费者依然能收到消息，但是如果刚挂掉的消费者重新连上后，在断开连接期间的消息对于该消费者来说彻底丢失了。
3）若订阅者订阅了频道，但自己读取消息的速度很慢的话，那么不断积压的消息会使redis输出缓冲区的体积变得越来越大，这可能使得redis本身的速度变慢，甚至直接崩溃。
4）另外，redis停机重启，pubsub的消息是不会持久化的，所有的消息被直接丢弃。

总结上面例子的缺点：

1）redis无法对消息持久化存储，消息一旦被发送，如果没有订阅者接收，数据会丢失。(客户端原因)
2）若订阅者订阅了频道，但自己读取消息的速度很慢的话，那么不断积压的消息会使redis输出缓冲区的体积变得越来越大，这可能使得redis本身的速度变慢，甚至直接崩溃。(客户端原因)
3）另外，redis停机重启，pubsub的消息是不会持久化的，所有的消息被直接丢弃。(服务器原因)

二 stream

1 stream的介绍

1）stream支持多播的可持久化消息队列。
2）一个消息链表将加入的消息都串起来，每个消息都有一个唯一的消息ID和对应的内容。消息都是持久化的，redis 重启后，内容还在。
3）每个 stream 对象通过一个 key 来唯一索引。每个 stream 都可以挂多个消费组（consumer group），每个消费组会有个游标 last_delivered_id 在 stream 数组之上往前移动，表示当前消费组已经消费到哪条消息了。
4）stream 在第一次使用 xadd 命令后自动创建，而消费组不会自动创建，需要通过命令 xgroup create 进行创建，并且需要指定从 stream 的某个消息 ID 开始消费。
5）每个消费组都是相互独立的，互相不受影响；也就是同一份 stream 内部的消息会被每个消费组都消费到。
6）同一个消费组可以挂接多个消费者，这些消费者之间是竞争关系，任意一个消费者读取了消息都会使游标往前移动。
7）消费者内部会有一个状态变量 pending_ids，它记录了当前已经被客户端读取，但是还没有 ack 的消息。当客户端 ack 一条消息后，pending_ids 将会删除该消息 ID。它用来确保客户端至少消费了消息一次，而不会在网络传输的中途丢失了而没有被处理。

上面几点关键是看懂stream、消费组、消费者这三者的作用。stream是用于持久化消息队列即存储消息的。消费组是独立的，不同消费组互不影响。每个消费组包含多个消费者，这些消费者之间是竞争关系。

例如stream、消费组、消费者三者作用的示例图：

2 基本命令

2.1 XADD、XDEL、XRANGE、XLEN、DEL

# 1. 向 stream 追加消息。
# 1）将指定的流条目追加到指定key的流中。 如果key不存在，作为运行这个命令的副作用，将使用流的条目自动创建key。
# 2）一个条目是由一组键值对组成的，它基本上是一个小的字典。 键值对以用户给定的顺序存储，并且读取流的命令（如XRANGE 或者 XREAD） 可以保证按照通过XADD添加的顺序返回。
# 3）XADD是唯一可以向流添加数据的Redis命令，但是还有其他命令， 例如XDEL和XTRIM，他们能够从流中删除数据。
# 4）将Stream ID指定为参数。流条目ID标识流内的给定条目。 如果指定的ID参数是字符*（星号ASCII字符），XADD命令会自动为您生成一个唯一的ID。
# 但是，也可以指定一个良好格式的ID，以便新的条目以指定的ID准确存储， 虽然仅在极少数情况下有用。
# ID是由-隔开的两个数字组成的，例如：1526919030474-55。关于手动生成具体看中文文档解释：http://redis.cn/commands/xadd.html。
# 5）返回值bulk-string-reply：
# 该命令返回添加的条目的ID。如果ID参数传的是*，那么ID是自动生成的， 否则，命令仅返回用户在插入期间指定的相同的ID。
# 强调：ID不要自己设置，填*就好。除非用到MYSQL+REDIS、或者有分布式系统集群了REDIS，他们需要一个全局的唯一id，否则统一使用*即可。
XADD key ID field string [field string ...]# 2. 从 stream 中删除消息。
# 1）从指定流中移除指定的条目，并返回成功删除的条目的数量，在传递的ID不存在的情况下， 返回的数量可能与传递的ID数量不同。
# 2）返回值integer-reply：返回成功删除的条目的数量。
# 3）理解删除条目的底层细节：
# Redis流以一种使其内存高效的方式表示：使用基数树来索引包含线性数十个Stream条目的宏节点。 通常，当你从Stream中删除一个条目的时候，条目并没有真正被驱逐，只是被标记为删除。
# 最终，如果宏节点中的所有条目都被标记为删除，则会销毁整个节点，并回收内存。 这意味着如果你从Stream里删除大量的条目，比如超过50%的条目，则每一个条目的内存占用可能会增加， 因为Stream将会开始变得碎片化。然而，流的表现将保持不变。
# 在Redis未来的版本中，当一个宏节点内删除条目达到一定数量的时候，我们有可能会触发节点垃圾回收机制。 目前，根据我们对这种数据结构的预期用途，还不太适合增加这样的复杂度。
XDEL key ID [ID ...]# 3. 获取 stream 中消息列表，会自动过滤已经删除的消息。
# 1）特殊ID：- 和 +。特殊ID-和+分别表示流中可能的最小ID和最大ID。
# 2）不完全ID：具体看中文文档，很简单，但是注意，-18446744073709551615的-只范围而不是减法符号，例如0-10的-代表范围。
# 3）使用COUNT返回最大条目数：具体看中文文档，很简单。
# 4）迭代流：具体看中文文档，很简单。
# 5）获取单个项目：具体看中文文档，很简单。
# 6）返回值array-reply：
# 该命令返回ID与指定范围匹配的条目。返回的条目是完整的，这意味着ID和所有组成条目的字段都将返回。此外，返回的条目及其字段和值的顺序与使用XADD添加它们的顺序完全一致。具体看中文文档解释。
# XRANGE的上面几点都理解了，看文档即可，主要知道如何遍历即可。
XRANGE key start end [COUNT count]# 4. 获取 stream 消息长度。
# 1）返回流中的条目数。如果指定的key不存在，则此命令返回0，就好像该流为空。 但是请注意，与其他的Redis类型不同，零长度流是可能的，所以你应该调用TYPE 或者 EXISTS 来检查一个key是否存在。
# 2）一旦内部没有任何的条目（例如调用XDEL后），流不会被自动删除，因为可能还存在与其相关联的消费者组。
XLEN key # 5. 删除 stream 中所有消息。
# 1）返回值integer-reply：返回被删除的keys的数量。删除指定的一批keys，如果删除中的某些key不存在，则直接忽略。
DEL key [key ...]

2.2 XREAD、XGROUP、XREADGROUP、XACK

# 1. 独立消费。注：XREAD、COUNT、BLOCK、STREAMS这四个是关键字。
# 1）返回值array-reply：该命令返回一个结果数组：返回数组的每个元素都是一个由两个元素组成的数组（键名和为该键报告的条目）。
#   报告的条目是完整的流条目，具有ID以及所有字段和值的列表。返回的条目及其字段和值的顺序与使用XADD添加它们的顺序完全一致。
# 关于更详细的看中文文档(不过这个命令的中文文档我还不是看得很懂，还是要项目开发时才能理解更深入)。
XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]# 2. 创建消费者。
# 该命令实际可以参创建消费者、删除消费者组、删除消费者。具体看中文文档，比较简单，已看懂。
XGROUP [CREATE key groupname id-or-$] [SETID key id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]# 3. 消费消息。
# 关于更详细的看中文文档(不过这个命令的中文文档我还不是看得很懂(看下面例子，现在比较熟了)，还是要项目开发时才能理解更深入，与XREAD一样)。
XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]# 4. 确认消费消息。
# 比较简单，就是用于处理服务器已经发送给客户端，但客户端未处理的消息。
# 返回值integer-reply：该命令返回成功确认的消息数。 某些消息ID可能不再是PEL的一部分（例如因为它们已经被确认）， 而且XACK不会把他们算到成功确认的数量中。
XACK key group ID [ID ...]

2.3 演示上面的基本命令

例如添加消息后，再里面读取消息：

# 1. 往stream添加消息，redis服务会返回这个消息的ID。
192.168.1.9:6379> XADD stream * message "hello world"
"1646747759983-0"# 2. 创建消费者组。因为消费者组必须自己创建，而上面的stream不存在时，XADD会自动创建。
# 此时的ID即0-0代表的是这个消费者组从0-0开始消费消息，即从头开始读取，意思更多的是代表一个范围。
192.168.1.9:6379> XGROUP CREATE stream g1 0-0
OK# 3. 查看stream消息队列的长度。
192.168.1.9:6379> XLEN stream
(integer) 1# 4. 查看具体的消息内容。
192.168.1.9:6379> XRANGE stream - +
1) 1) "1646747759983-0"2) 1) "message"2) "hello world"# 5. 读取消息。
# 1）GROUP代表关键字。g1是上面创建的消费者组的名字。tyy是消费者名字，不存在会自动创建。
# 2）COUNT 1不写时，默认也是1.
# 3）STREAMS代表关键字。后面stream是我们XADD时创建的流名字，可在XADD自行取名。
# 4）0代表我们要读取已经发送给客户端，但客户端未ACK的消息。因为这里还没有将任何消息发送过给任何客户端(消费者)，所以返回空数组。
# 这里强调，这个命令的ID实际一般只有特殊ID">"与其它ID的区别。
#   1. 特殊ID>，意味着消费者希望只接收从未发送给任何其他消费者的消息。
#   2. 任意其他的ID，即0或任意其他有效ID或不完整的ID（只有毫秒时间部分），将具有返回发送命令的消费者的待处理条目的效果。所以，基本上如果ID不是>，命令将让客户端访问它的待处理条目（已发送给它，但尚未确认的条目）。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream 0
1) 1) "stream"2) (empty array)
# 5)同4一样，ID=1、1111111111111、11111111111111不是特殊ID，所以返回空数组。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream 1
1) 1) "stream"2) (empty array)
192.168.1.9:6379>
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream 1111111111111
1) 1) "stream"2) (empty array)
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream 11111111111111
1) 1) "stream"2) (empty array)
192.168.1.9:6379>
# 6）传特殊字符>，能够读取从未发送给任何其他消费者的消息。例如上面XADD的message "hello world"消息还没有其它消费者读取过，那么使用特殊字符必定会返回。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream >
1) 1) "stream"2) 1) 1) "1646747759983-0"2) 1) "message"2) "hello world"# 7）ID传负数会报错。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream -2
(error) ERR Invalid stream ID specified as stream command argument# 8）由于上面传完特殊ID后，消息被消费者读取过即redis已经发送了，但是目前消费者还没有ACK，所以我们可以使用其它字符可以读取到message "hello world"这条消息。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream 2
1) 1) "stream"2) 1) 1) "1646747759983-0"2) 1) "message"2) "hello world"
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream 1111111111111
1) 1) "stream"2) 1) 1) "1646747759983-0"2) 1) "message"2) "hello world"
# 不过注意，当我上面传13个1时，可以读取到消息，传14个1返回空数组了。估计与ID的位数有关(具体可自行查看ID的范围)。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream 11111111111111
1) 1) "stream"2) (empty array)
# 传完14个1返回空数组后，消息正常仍然能传其它ID正常读取的。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream 111111111111
1) 1) "stream"2) 1) 1) "1646747759983-0"2) 1) "message"2) "hello world"# 9）若此时再用特殊字符读取，则会返回nil，因为此时stream中没有这样的消息，即不存在未发送过给客户端的消息。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream >
(nil)# 10）若此时使用XACK确认处理完这条消息，redis会从待处理的消息队列去掉，那么使用其它任意ID读取消息时，那么不会再像上面第8点一样返回消息，而是返回空数组。
192.168.1.9:6379> XACK stream g1 1646747759983-0
(integer) 1
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream 0
1) 1) "stream"2) (empty array)
192.168.1.9:6379> # 我在进行到下面的第6点的XADD时，发现redis报错："MISCONF Redis is configured to save RDB snapshotsxxx"。
# 这个错误和一些配置修改看这里：https://blog.csdn.net/xc_zhou/article/details/80893326?spm=1001.2014.3001.5502。    大概意思就是内存和硬盘不足。
# 扩展硬盘大小的文章(内存很简单的，改一下配置即可)：https://blog.csdn.net/hanpengyu/article/details/7475645
# 然后下面的命令就可以继续正常执行了，不过上面的命令内容最好重新输入一遍，防止影响正确的结果。# 6. 然后我们多添加几条消息。
# 注意：message可以相同，因为每个消息可以认为是队列的一个元素，类似上图的id1，id2。
#   这个元素可认为是一个map，它可以存放多个key-value结构，这里我只存放了一个key-value结构，即message "hello tyy"。
192.168.1.9:6379> XADD stream * message "hello tyy"
"1646747906781-0"
192.168.1.9:6379> XADD stream * message "hello lqq"
"1646747915368-0"
192.168.1.9:6379> XADD stream * message "hello hc"
"1646747918299-0"
# 加上原来那条，总共有4条数据。
192.168.1.9:6379> XRANGE stream - +
1) 1) "1646747759983-0"2) 1) "message"2) "hello world"
2) 1) "1646747906781-0"2) 1) "message"2) "hello tyy"
3) 1) "1646747915368-0"2) 1) "message"2) "hello lqq"
4) 1) "1646747918299-0"2) 1) "message"2) "hello hc"
192.168.1.9:6379> # 7. 然后使用特殊字符读取一条消息。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream >
1) 1) "stream"2) 1) 1) "1646747906781-0"2) 1) "message"2) "hello tyy"# 8. 由于第7点读取了一条消息，且现在还未XACK，所以使用0可以从待处理中读取到这条消息。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 1 STREAMS stream 0
1) 1) "stream"2) 1) 1) "1646747906781-0"2) 1) "message"2) "hello tyy"
192.168.1.9:6379>
# 注意，COUNT参数不填，默认是返回1条。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy STREAMS stream 0
1) 1) "stream"2) 1) 1) "1646747906781-0"2) 1) "message"2) "hello tyy"
# 如果COUNT填3条，由于此时待处理的消息只有一条1646747906781-0(原先的1646747759983被XACK了，所以不会存在待处理的队列中)，所以也会返回一条。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 3 STREAMS stream 0
1) 1) "stream"2) 1) 1) "1646747906781-0"2) 1) "message"2) "hello tyy"# 9. 使用XACK确认消息。
192.168.1.9:6379> XACK stream g1 1646747906781-0
(integer) 1
# 所以此时再次从待处理的队列中读取消息，便为空了。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 3 STREAMS stream 0
1) 1) "stream"2) (empty array)
192.168.1.9:6379># 10. 再次读取未向客户端发送过的消息。因为还剩两条，所以我使COUNT=3。实际COUNT大于等于2都行。
192.168.1.9:6379>
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 3 STREAMS stream >
1) 1) "stream"2) 1) 1) "1646747915368-0"2) 1) "message"2) "hello lqq"2) 1) "1646747918299-0"2) 1) "message"2) "hello hc"
192.168.1.9:6379> # 11. 最后传指定ID。它会从该ID的下一个ID开始返回消息，返回的个数由COUNT决定，如果待处理的消息不足COUNT的个数，那么全部返回。如果该ID的下一个ID没有消息了，那么返回空数组。
# 不过只有这里有个疑问：为什么中文文档对这个命令的解释不是很准确，它描述说只有特殊字符和其它任意ID的区别吗，不是很理解中文文档想表达的一样。
192.168.1.9:6379>
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 3 STREAMS stream 1646747915368-0
1) 1) "stream"2) 1) 1) "1646747918299-0"2) 1) "message"2) "hello hc"
192.168.1.9:6379>

# 再次验证第11点，即它会从该ID的下一个ID开始返回消息，返回的个数由COUNT决定，如果待处理的消息不足COUNT的个数，那么全部返回。如果该ID的下一个ID没有消息了，那么返回空数组。
# 结果是没有问题的。
192.168.1.9:6379> del s1
(integer) 1
192.168.1.9:6379>
192.168.1.9:6379> EXISTS s1
(integer) 0
192.168.1.9:6379> # 1. 往stream中添加多个消息。
192.168.1.9:6379> XADD s1 * message "hello tyy"
"1646834322070-0"
192.168.1.9:6379> XADD s1 * message "hello lqq"
"1646834325931-0"
192.168.1.9:6379> XADD s1 * message "hello hc"
"1646834331433-0"
192.168.1.9:6379> XADD s1 * message "hello hrc"
"1646834335236-0"
192.168.1.9:6379> XADD s1 * message "hello ljm"
"1646834338844-0"
192.168.1.9:6379> # 2. 创建消费者组，它只能通过手动调XGROUP命令去创建。
192.168.1.9:6379> XGROUP CREATE s1 g1 0-0
OK
192.168.1.9:6379> # 3. 查看流消息队列的长度和内容。
192.168.1.9:6379> XLEN s1
(integer) 5
192.168.1.9:6379> XRANGE s1 - +
1) 1) "1646834322070-0"2) 1) "message"2) "hello tyy"
2) 1) "1646834325931-0"2) 1) "message"2) "hello lqq"
3) 1) "1646834331433-0"2) 1) "message"2) "hello hc"
4) 1) "1646834335236-0"2) 1) "message"2) "hello hrc"
5) 1) "1646834338844-0"2) 1) "message"2) "hello ljm"
192.168.1.9:6379> # 4. 先把消息发送给客户端(这里我COUNT=10可以把5条消息全部发送完)，使消息进入待处理队列。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 >
1) 1) "s1"2) 1) 1) "1646834322070-0"2) 1) "message"2) "hello tyy"2) 1) "1646834325931-0"2) 1) "message"2) "hello lqq"3) 1) "1646834331433-0"2) 1) "message"2) "hello hc"4) 1) "1646834335236-0"2) 1) "message"2) "hello hrc"5) 1) "1646834338844-0"2) 1) "message"2) "hello ljm"
192.168.1.9:6379> # 5. 传0或者其它ID(不包含消息的唯一ID，即下面的tyy的xxx070、lqq的xxx931等它们自己的ID)，代表读取已经发送给客户端但客户端未ACK的消息。这里COUNT=10可以读取待处理队列的全部5条消息。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 0
1) 1) "s1"2) 1) 1) "1646834322070-0"2) 1) "message"2) "hello tyy"2) 1) "1646834325931-0"2) 1) "message"2) "hello lqq"3) 1) "1646834331433-0"2) 1) "message"2) "hello hc"4) 1) "1646834335236-0"2) 1) "message"2) "hello hrc"5) 1) "1646834338844-0"2) 1) "message"2) "hello ljm"
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 1
1) 1) "s1"2) 1) 1) "1646834322070-0"2) 1) "message"2) "hello tyy"2) 1) "1646834325931-0"2) 1) "message"2) "hello lqq"3) 1) "1646834331433-0"2) 1) "message"2) "hello hc"4) 1) "1646834335236-0"2) 1) "message"2) "hello hrc"5) 1) "1646834338844-0"2) 1) "message"2) "hello ljm"
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 11111111111
1) 1) "s1"2) 1) 1) "1646834322070-0"2) 1) "message"2) "hello tyy"2) 1) "1646834325931-0"2) 1) "message"2) "hello lqq"3) 1) "1646834331433-0"2) 1) "message"2) "hello hc"4) 1) "1646834335236-0"2) 1) "message"2) "hello hrc"5) 1) "1646834338844-0"2) 1) "message"2) "hello ljm"
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 1111111111111
1) 1) "s1"2) 1) 1) "1646834322070-0"2) 1) "message"2) "hello tyy"2) 1) "1646834325931-0"2) 1) "message"2) "hello lqq"3) 1) "1646834331433-0"2) 1) "message"2) "hello hc"4) 1) "1646834335236-0"2) 1) "message"2) "hello hrc"5) 1) "1646834338844-0"2) 1) "message"2) "hello ljm"
# 但是这里注意，传14个1时返回了空数组。可能与位数有关。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 11111111111111
1) 1) "s1"2) (empty array)
192.168.1.9:6379> # 6. 传具体的消息ID。
# 它会从该ID的下一个ID开始返回消息，返回的个数由COUNT决定，如果待处理的消息不足COUNT的个数，那么全部返回。如果该ID的下一个ID没有消息了，那么返回空数组。
# 1）从070开始读，那么从下一个ID即931开始返回，共4条，由于COUNT=10，所以这4条会全部被返回。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 1646834322070-0
1) 1) "s1"2) 1) 1) "1646834325931-0"2) 1) "message"2) "hello lqq"2) 1) "1646834331433-0"2) 1) "message"2) "hello hc"3) 1) "1646834335236-0"2) 1) "message"2) "hello hrc"4) 1) "1646834338844-0"2) 1) "message"2) "hello ljm"
192.168.1.9:6379>
# 2）从931开始读，那么会从433开始返回，共返回3条。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 1646834325931-0
1) 1) "s1"2) 1) 1) "1646834331433-0"2) 1) "message"2) "hello hc"2) 1) "1646834335236-0"2) 1) "message"2) "hello hrc"3) 1) "1646834338844-0"2) 1) "message"2) "hello ljm"
# 3）从433开始读，那么会从236开始返回，共返回2条。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 1646834331433-0
1) 1) "s1"2) 1) 1) "1646834335236-0"2) 1) "message"2) "hello hrc"2) 1) "1646834338844-0"2) 1) "message"2) "hello ljm"
# 4）从236开始读，那么会从844开始返回，共返回1条。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 1646834335236-0
1) 1) "s1"2) 1) 1) "1646834338844-0"2) 1) "message"2) "hello ljm"
# 5）从844开始读，由于844后面没有待处理的消息了，所以会返回空数组，代表这个ID是这个待处理消息队列的最后一个ID。
#   但不能说待处理消息队列已经处理完毕，因为这个具体ID的前面可能有消息。判断是否待处理消息队列是否处理完毕，应该使用ID=0和XACK去判断，
#   因为ID=0时代表从头开始读取消息，如果返回空数组，那么这个待处理消息队列肯定是空，说明待处理消息队列已经处理完毕。
192.168.1.9:6379> XREADGROUP GROUP g1 tyy COUNT 10 STREAMS s1 1646834338844-0
1) 1) "s1"2) (empty array)
192.168.1.9:6379>