#yyds干货盘点#Redis之Stream

别裁伪体亲风雅,转益多师是汝师。这篇文章主要讲述#yyds干货盘点#Redis之Stream相关的知识,希望能为你提供帮助。
1、简介Stream弥补了Redis作为MQ(message queue)技术选型上的不足之处;Redis 5.0发布的Stream相比Pub/Sub模块,Stream支持消息持久化,结合sentinel或cluster使其成为了一个比较可靠的消息队列。尽管我认为它很难成为公司MQ的技术选型产品,但是关于Stream的使用和特性(消费组),仍值得一探究竟。


Stream对标消息队列,因此几乎具备了MQ所有的特性,以下列出Stream所具有的部分特性:

  • 消息顺序存储
  • 消息ID序列化规则生成
  • 消息的遍历
  • 消息阻塞/非阻塞式获取
  • 客户端分组消费消息
  • 消息确认机制
  • 消息异常机制
  • 消息队列监控
在文中也会说到Stream的这些特性。


2、Stream内部探索2.1 Stream 结构
在探索Stream的内部结构之前,先看一张清晰的Stream结构图:

如下是关于上图的名词解析:
  • Message Content:消息内容
  • Consumer group:消费组,通过XGROUP CREATE 命令创建,一个消费组可以有多个消费者
  • Last_delivered_id:游标,每个消费组有一个游标,任意消费者读取消息后,游标都会向前移动
  • Consumer:消费者,消费组中的消费者
  • Pending_ids:状态变量,每个消费者会有一个状态变量,用于记录被当前消费者读取,但是并未ack的消息id


2.2 四个唯一
Stream内部维护了一个消息链表,以此使得消息能够具有队列的特性。在Stream中有四个唯一需要了解:
  1. 每个Stream都具有唯一的名称
  2. 每个消息(Message)都具有一个由系统分配或者客户端指定唯一ID
  3. 每个Stream中的消费组(Consumer_Group)具有唯一名称
  4. 每个消费组(Consumer_Group)中的消费者(Consumer)具有唯一名称


2.3 消息ID
Stream的消息ID可以由服务端自动生成,也可以由客户端传入,如下图是自动生成的结构:

系统自动生成的规则
< millisecondsTime> -< sequenceNumber>
millisecondsTime指的是Redis节点服务器的本地时间,如果存在当前的毫秒时间戳比以前已经存在的数据的时间戳小的话(本地时间钟后跳),那么系统将会采用以前相同的毫秒创建新的ID。
sequenceNumber指的是序列号,在相同的millisecondsTime毫秒下,序列号从0开始递增,序列号是64位长度,理论上在统一毫秒内生成的数据量无法到达这个级别,因此不用担心sequenceNumber会不够用。


客户端显示传入规则
Redis对于ID有强制要求,格式必须是< millisecondsTime> -< sequenceNumber> ,最小ID为0-1,并且后续ID不能小于前一个ID


2.4 消息内容
Stream的消息内容,也就是图中的Message Content它的结构类似Hash结构,以key-value的形式存在。




3、Stream指令3.1 指令汇总
Stream的指令根据可以分为两类,分别是消息队列相关指令,消费组相关指令。
消息队列相关指令:
指令名称
指令作用
XADD
添加消息到队列末尾
XTRIM
限制Stream的长度,如果已经超长会进行截取
XDEL
删除消息
XLEN
获取Stream中的消息长度
XRANGE
获取消息列表(可以指定范围),忽略删除的消息
XREVRANGE
和XRANGE相比区别在于反向获取,ID从大到小
XREAD
获取消息(阻塞/非阻塞),返回大于指定ID的消息
消费组相关指令:
指令名称
指令作用
XGROUP CREATE
创建消费者组
XREADGROUP GROUP
读取消费者组中的消息
XACK
ack消息,消息被标记为“已处理”
XGROUP SETID
设置消费者组最后递送消息的ID
XGROUP DELCONSUMER
删除消费者组
XPENDING
打印待处理消息的详细信息
XCLAIM
转移消息的归属权(长期未被处理/无法处理的消息,转交给其他消费者组进行处理)
XINFO
打印Stream\\Consumer\\Group的详细信息
XINFO GROUPS
打印消费者组的详细信息
XINFO STREAM
打印Stream的详细信息


3.2 XADD
XADD 用于向Stream 队列中添加消息,如果指定的Stream 队列不存在,则该命令执行时会新建一个Stream 队列。


XADD的指令语法:
XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|ID field value [field value ...]


如下通过XADD展示了定义ID的两种方式,具体可以看2.3。



3.2 XTRIM
XTRIM 用于对Stream的长度进行限定。


XTRIM 的指令语法:
XTRIM key MAXLEN|MINID [=|~] threshold [LIMIT count]
  • MAXLEN 允许的最大长度,如果长度超出则会抛弃队列前面的消息
  • MINID 允许的最小id,从某个id值开始保留,其余的将会被抛弃

【#yyds干货盘点#Redis之Stream】

3.3 XDEL
XDEL 用于删除消息。


XDEL 的指令语法:
XDEL key ID [ID ...]



3.4 XLEN
XLEN 用于获取Stream 队列的消息的长度。


XLEN 的指令语法:
XLEN key



3.5 XRANGE
XRANGE 用于获取消息列表(可以指定范围),忽略删除的消息。


XRANGE 的指令语法:
XRANGE key start end [COUNT count]
  • start 表示开始值,-代表最小值
  • end 表示结束值,+代表最大值
  • count 表示最多获取多少个值



3.6 XREVRANGE
XREVRANGE 用于获取消息列表(可以指定范围),忽略删除的消息。与XRANGE 的区别在于,获取消息列表元素的方向是相反的,end在前,start在后。


XREVRANGE 的指令语法:
XREVRANGE key end start [COUNT count]



3.7 XREAD
XREAD 用于获取消息(阻塞/非阻塞),只会返回大于指定ID的消息。


XREAD 的指令语法:
XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
  • COUNT 最多读取多少条消息
  • BLOCK 是否已阻塞的方式读取消息,默认不阻塞,如果milliseconds设置为0,表示永远阻塞


$代表特殊ID,表示以当前Stream已经存储的最大的ID作为最后一个ID,当前Stream中不存在大于当前最大ID的消息,因此此时返回nil。
0-0代表从最小的ID开始获取Stream中的消息,当不指定count,将会返回Stream中的所有消息,注意也可以使用0(00/000也都是可以的……)。



阻塞方式获取Stream中的指令,这里演示阻塞获取一条消息




3.8 XGROUP CREATE
XGROUP CREATE 用于创建消费者组。


XGROUP CREATE 的指令语法:
XGROUP [CREATE key groupname ID|$ [MKSTREAM]] [SETID key groupname ID|$] [DESTROY key groupname] [CREATECONSUMER key groupname consumername] [DELCONSUMER key groupname consumername]
XGROUP CREATE中的指令没什么复杂的,第一个中括号中的几个参数最为重要,如下图两种方式:
  • $表示从Stream尾部开始消费,会忽略Stream中目前已有的数据
  • 0表示从Stream头部开始消费



如果Stream不存在,XGROUP CREATE 语法将会报错,因此可以得出不允许在不存在的Stream上创建消费者组



3.9 XREADGROUP GROUP
XREADGROUP GROUP 用于读取消费者组中的消息。


XREADGROUP GROUP 的指令语法:
XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]


注意,这里有一个比较重要的知识点,刚开始的时候可能容易搞错:
> 这个特殊符号表示消息到目前为止,从未传递给其他消费者的消息
0表示指定消息ID,因为ID均大于0-0(0代指0-0),因此代表从Stream 的队列头部开始获取消息


在如下截图中,为何第一次 mystream 0 获取消息返回empty,在执行完 mystream > 之后,第二 mystream 0 却成功的获取到了消息,但是很明显mystream中刚添加了两条消息,第一次不应该失败才对呀?
这是因为,当指定ID进行消息获取时,命令将会让我们访问我们的历史待处理消息(曾被获取,但是未ack)。即传递给这个指定消费者(由提供的名称标识)的消息集,并且到目前为止从未使用XACK进行确认。



XREADGROUP GROUP 也可以像XREAD 一样使用阻塞的方式获取消息

当向mystream中添加消息后,阻塞读返回



3.10 XACK
XACK 用于标记为“已处理”。


XACK 的指令语法:
XACK key group ID [ID ...]


结合XREADGROUP GROUP 中指定ID的方式只能获取未ack的未处理消息的特性,测试XACK指令。从如下的测试示例中可以得出两个结论:
  • 消息首次ack成功,返回1,ack失败返回0
  • 3.9中的结论是正确的



3.11 XPENDING
XPENDING 用于打印待处理消息的详细信息。


XPENDING 指令是非常有用的,因为它可以打印待处理消息的信息。如果在一个消费者组中存在多个消费者,如果存在部分消费者永久的故障,无法再处理消息了,我们就可以通过XPENDING 指令来查看指定消费者组中的消费者未ack的消息,然后转移给其他消费者进行处理。


XPENDING 的指令语法:
XPENDING key group [[IDLE min-idle-time] start end count [consumer]]


XPENDING 返回值解析:
  1. 第一个参数表示当前消费者中待处理消息的总数
  2. 第二个参数表示待处理消息的最小ID
  3. 第三个参数表示待处理消息的最大ID
  4. 第四个参数表示消费者列表和未处理的消息数量



3.11 XCLAIM
XCLAIM 用于转移消息的归属权。


XCLAIM 的指令语法:
XCLAIM key group consumer min-idle-time ID [ID ...] [IDLE ms] [TIME ms-unix-time] [RET


指令参数解析:
  • key 表示Stream的名称
  • group 表示需要转移消息的归属权的消费者组名称
  • consumer 表示接收消息的消费者名称
  • min-idle-time 表示最小空闲时间,只有后续指定ID的消息空闲时间大于指定的空闲时间,消息归属权转移指令才会生效
  • ID [] 需要转移归属权的消息ID,数组,可以是多个


示例中,将consumer-1中ID为1631719560149-0的未处理的消息的归属权转移到consumer-2下:





3.13 XINFO
XINFO 用于打印Stream\\Consumer\\Group的详细信息。


XINFO的指令语法:
XINFO [CONSUMERS key groupname] [GROUPS key] [STREAM key] [HELP]


示例打印指定STREAM的详细消息



4、关于Stream优化内存的事情使用Stream有两个点需要注意,如果使用不当都会导致内存消耗增大。
  1. 待处理消息过多,消息未及时ack
  2. Stream消息持续持久化,使用XDEL删除消息


关于第一点,待处理消息过多,消息未及时ack,其导致内存增加的原因是,Stream会为每个消费者维护一个PEL列表,PEL列表用于存储处理完但未及时ack的消息ID。我们在实际使用过程中,处理完的消息一定要及时ack,也有定时检查是否有消费者不可用导致消息堆积的情况。
XPENDING能查询出消费者中待处理的消息,就是因为有PEL的存在。



关于第二点,使用XDEL删除Stream中不在需要的消息,其导致内存增加的原因是,Stream的XDEL删除消息的指令,并不会从内存上删除消息,它只是给消息打上标记位,下次通过XRANGE指令忽略这些消息而已。因此我们可以设置Stream的最大长度,来解决这个问题,在XADD中使用MAXLEN指定Stream队列的长度,当消息超出长度就会将队列头消息清除掉。(不过这种处理方式一定要做到及时处理消息,避免消息的丢失。)
XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|ID field value [field value ...]







    推荐阅读