一：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 01) 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 31) 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 -> C23 -> C34 -> C15 -> C26 -> C37 -> 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   # 创建消费组mygroup1OK127.0.0.1:6379> XREADGROUP GROUP mygroup1 Alice COUNT 1 STREAMS mystream1 >  # 第1次消费Alice1) 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次消费  Alice1) 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次消费 Alice21) 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) 7119302) 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