一：Stream简介
Redis Stream是5.0版本之后新增的一种数据结构，其结构类似于‘仅追加日志’。但也实现了多种操作来克服‘仅追加日志’的一些限制，如读取策略(xread,xrange....)。.....
二：基本指令与用法
2.1 xadd 向流添加新节点（返回节点id）
127.0.0.1:6379> XADD mystream1 * name a1
"1688888761768-0"
127.0.0.1:6379> XADD mystream1 * name a2
"1688888772896-0"
127.0.0.1:6379> XADD mystream1 * name a3
"1688888782841-0"
127.0.0.1:6379> XADD mystream1 * name a4
"1688888786201-0"
指令解释
第一个参数=‘mystream’: stream的key。
第二个参数=‘*’: stream节点的id生成策略，能保证自增且格式为（毫秒值-序列号）,即使计算机时间不正确。
第三个参数=‘name a1’: 一个key-value的键值对。
2.2 XREAD 读取一个或多个节点，从给定位置开始并向前移动。
127.0.0.1:6379> XREAD COUNT 2 STREAMS mystream1 0
1) 1) "mystream1"
2) 1) 1) "1688888761768-0"
2) 1) "name"
2) "a1"
2) 1) "1688888772896-0"
2) 1) "name"
2) "a2"
指令解释
第一个参数=‘COUNT 2’: 一次读取两条数据。
第二个参数=‘STREAMS mystream1’: 读取key等于mystream1的stream。
第三个参数=‘0’: 读取策略-stream节点id大于它的数，根据id生成策略任何id都大于0，故重头读取。
若=’$‘ ：则可读取id大于当前最大值的id，即读取最新的节点。
其他参数:
如： 'BLOCK 0'(堵塞读取，0表示无限堵塞)
备注: 对于不同的客户段(client),只需简单的设置(记录自己的最大消费id)就可实现fan-out 的消费策略。
2.3 XRANGE 返回两个提供的条目 ID 之间的节点范围。
127.0.0.1:6379> XRANGE mystream1 - + COUNT 3
1) 1) "1688888761768-0"
2) 1) "name"
2) "a1"
2) 1) "1688888772896-0"
2) 1) "name"
2) "a2"
3) 1) "1688888782841-0"
2) 1) "name"
2) "a3"
指令解释
第一个参数=‘mystream1’: key等于mystream1的stream。
第二个参数=‘- +’: 特殊含义，-(表示id最小值)，+(表示id最大值)。
第三个参数=‘COUNT 3’: 一次最多读取3个
2.4 XLEN 返回流的长度(节点数)。
127.0.0.1:6379> XLEN mystream1
(integer) 4
三：消费组(Consumer groups)
与XREAD指令不同，消费组可实现再同一个消费组内的消费者只消费未被消费的消息。假设我们想象有三个消费者 C1、C2、C3 和一个包含消息 1、2、3、4、5、6、7 的流，那么我们想要的是按照下图提供消息：
1 -> C1
2 -> C2
3 -> C3
4 -> C1
5 -> C2
6 -> C3
7 -> C1
3.1 看图理解消费组模式
+----------------------------------------+
| 消费组名: mygroup
| 消费的Stream: somekey
| 最近一次消费的id: 1292309234234-92
|
| 消费者数组:
| "消费者-1" 未ack的消息
| 1292309234234-4
| 1292309234232-8
| "消费者-2" 未ack的消息
| ... (等等)
+----------------------------------------+
四： 消费组相关命令
4.1 创建消费组: XGROUP CREATE
127.0.0.1:6379> XGROUP CREATE mystream1 mygroup $
OK
指令解释
第一个参数=‘mystream1’: 消费的Stream的key。
第二个参数=‘mygroup’:命名消费组名称。
第三个参数=‘ $’: 节点的id，表示消费只消费大于该id的消息，特殊字符$表示只消费该组创建后的消息。
可选参数'MKSTREAM‘ 表示自动创建stream（当stream不存在时）。
4.2 创建组内消费者，并消费消息：XREADGROUP GROUP
127.0.0.1:6379> XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream1 >
(nil) #没有任何消息
# 因为4.1创建时用的三个参数=‘ $’
指令解释
第一个参数=‘mygroup’: 消费的Stream的key。
第二个参数=‘Alice’:消费者名称。
第三个参数=‘ COUNT 1’: 每次消费的最大数量。
第四个参数=‘ mystream1: 消费的Stream的key。
第五个参数=‘ >’: 表示未消费的消息。
示例2：创建消费组2，从头（0）消费
127.0.0.1:6379> XGROUP CREATE mystream1 mygroup1 0 # 创建消费组mygroup1
OK
127.0.0.1:6379> XREADGROUP GROUP mygroup1 Alice COUNT 1 STREAMS mystream1 > # 第1次消费Alice
1) 1) "mystream1"
2) 1) 1) "1688957166436-0"
2) 1) "name"
2) "a1"
......
127.0.0.1:6379> XREADGROUP GROUP mygroup1 Alice COUNT 1 STREAMS mystream1 > #第5次消费 Alice
(nil)
127.0.0.1:6379> XADD mystream1 * name a5 #新增消息
"1688959089805-0"
127.0.0.1:6379> XREADGROUP GROUP mygroup1 Alice COUNT 1 STREAMS mystream1 > #第6次消费 Alice
1) 1) "mystream1"
2) 1) 1) "1688959089805-0"
2) 1) "name"
2) "a5"
127.0.0.1:6379> XREADGROUP GROUP mygroup1 Alice2 COUNT 1 STREAMS mystream1 > #第7次消费 Alice2
(nil)
127.0.0.1:6379> XADD mystream1 * name a6 #新增消息
"1688959797006-0"
127.0.0.1:6379> XREADGROUP GROUP mygroup1 Alice2 COUNT 1 STREAMS mystream1 > #第8次消费 Alice2
1) 1) "mystream1"
2) 1) 1) "1688959797006-0"
2) 1) "name"
2) "a6"
4.3 查看消费组信息 XINFO CONSUMERS
127.0.0.1:6379> XINFO CONSUMERS mystream1 mygroup1
1) 1) "name" #消费者名称
2) "Alice"
3) "pending" #待确认的消息数 (ack)
4) (integer) 5
5) "idle"
6) (integer) 711930
2) 1) "name" #消费者名称
2) "Alice2"
3) "pending" #待确认的消息数(ack)
4) (integer) 1
5) "idle"
6) (integer) 4160
4.4 确认消息 XACK key group id (返回表示确认成功数)
127.0.0.1:6379> XACK mystream1 mygroup1 1688959797006-0
(integer) 1
参考资料:https://redis.io/docs/data-types/streams/
Stream全部命令:https://redis.io/commands/?group=stream