求不更学不动之Redis5.0新特性Stream尝鲜

时间 2019-11-24

标签不动 redis5.0 redis 特性 stream 尝鲜栏目 Redis 繁體版

原文原文链接

Redis5.0最近被做者忽然放出来了，增长了不少新的特点功能。而Redis5.0最大的新特性就是多出了一个数据结构Stream，它是一个新的强大的支持多播的可持久化的消息队列，做者坦言Redis Stream狠狠地借鉴了Kafka的设计。node

Redis Stream的结构如上图所示，它有一个消息链表，将全部加入的消息都串起来，每一个消息都有一个惟一的ID和对应的内容。消息是持久化的，Redis重启后，内容还在。数组

每一个Stream都有惟一的名称，它就是Redis的key，在咱们首次使用xadd指令追加消息时自动建立。bash

每一个Stream均可以挂多个消费组，每一个消费组会有个游标last_delivered_id在Stream数组之上往前移动，表示当前消费组已经消费到哪条消息了。每一个消费组都有一个Stream内惟一的名称，消费组不会自动建立，它须要单独的指令xgroup create进行建立，须要指定从Stream的某个消息ID开始消费，这个ID用来初始化last_delivered_id变量。服务器

每一个消费组(Consumer Group)的状态都是独立的，相互不受影响。也就是说同一份Stream内部的消息会被每一个消费组都消费到。网络

同一个消费组(Consumer Group)能够挂接多个消费者(Consumer)，这些消费者之间是竞争关系，任意一个消费者读取了消息都会使游标last_delivered_id往前移动。每一个消费者者有一个组内惟一名称。数据结构

消费者(Consumer)内部会有个状态变量pending_ids，它记录了当前已经被客户端读取的消息，可是尚未ack。若是客户端没有ack，这个变量里面的消息ID会愈来愈多，一旦某个消息被ack，它就开始减小。这个pending_ids变量在Redis官方被称之为PEL，也就是Pending Entries List，这是一个很核心的数据结构，它用来确保客户端至少消费了消息一次，而不会在网络传输的中途丢失了没处理。ui

消息ID

消息ID的形式是timestampInMillis-sequence，例如1527846880572-5，它表示当前的消息在毫米时间戳1527846880572时产生，而且是该毫秒内产生的第5条消息。消息ID能够由服务器自动生成，也能够由客户端本身指定，可是形式必须是整数-整数，并且必须是后面加入的消息的ID要大于前面的消息ID。spa

消息内容

消息内容就是键值对，形如hash结构的键值对，这没什么特别之处。设计

增删改查

xadd 追加消息
xdel 删除消息，这里的删除仅仅是设置了标志位，不影响消息总长度
xrange 获取消息列表，会自动过滤已经删除的消息
xlen 消息长度
del 删除Stream

# *号表示服务器自动生成ID，后面顺序跟着一堆key/value
# 名字叫laoqian，年龄30岁
127.0.0.1:6379> xadd codehole * name laoqian age 30  
1527849609889-0  # 生成的消息ID
127.0.0.1:6379> xadd codehole * name xiaoyu age 29
1527849629172-0
127.0.0.1:6379> xadd codehole * name xiaoqian age 1
1527849637634-0
127.0.0.1:6379> xlen codehole
(integer) 3
# -表示最小值, +表示最大值
127.0.0.1:6379> xrange codehole - +
127.0.0.1:6379> xrange codehole - +
1) 1) 1527849609889-0
   2) 1) "name"
      2) "laoqian"
      3) "age"
      4) "30"
2) 1) 1527849629172-0
   2) 1) "name"
      2) "xiaoyu"
      3) "age"
      4) "29"
3) 1) 1527849637634-0
   2) 1) "name"
      2) "xiaoqian"
      3) "age"
      4) "1"
# 指定最小消息ID的列表
127.0.0.1:6379> xrange codehole 1527849629172-0 +  
1) 1) 1527849629172-0
   2) 1) "name"
      2) "xiaoyu"
      3) "age"
      4) "29"
2) 1) 1527849637634-0
   2) 1) "name"
      2) "xiaoqian"
      3) "age"
      4) "1"
# 指定最大消息ID的列表
127.0.0.1:6379> xrange codehole - 1527849629172-0
1) 1) 1527849609889-0
   2) 1) "name"
      2) "laoqian"
      3) "age"
      4) "30"
2) 1) 1527849629172-0
   2) 1) "name"
      2) "xiaoyu"
      3) "age"
      4) "29"
127.0.0.1:6379> xdel codehole 1527849609889-0
(integer) 1
# 长度不受影响
127.0.0.1:6379> xlen codehole
(integer) 3
# 被删除的消息没了
127.0.0.1:6379> xrange codehole - +
1) 1) 1527849629172-0
   2) 1) "name"
      2) "xiaoyu"
      3) "age"
      4) "29"
2) 1) 1527849637634-0
   2) 1) "name"
      2) "xiaoqian"
      3) "age"
      4) "1"
# 删除整个Stream
127.0.0.1:6379> del codehole
(integer) 1
复制代码

独立消费

咱们能够在不定义消费组的状况下进行Stream消息的独立消费，当Stream没有新消息时，甚至能够阻塞等待。Redis设计了一个单独的消费指令xread，能够将Stream当成普通的消息队列(list)来使用。使用xread时，咱们能够彻底忽略消费组(Consumer Group)的存在，就比如Stream就是一个普通的列表(list)。code

# 从Stream头部读取两条消息
127.0.0.1:6379> xread count 2 streams codehole 0-0
1) 1) "codehole"
   2) 1) 1) 1527851486781-0
         2) 1) "name"
            2) "laoqian"
            3) "age"
            4) "30"
      2) 1) 1527851493405-0
         2) 1) "name"
            2) "yurui"
            3) "age"
            4) "29"
# 从Stream尾部读取一条消息，毫无疑问，这里不会返回任何消息
127.0.0.1:6379> xread count 1 streams codehole $
(nil)
# 从尾部阻塞等待新消息到来，下面的指令会堵住，直到新消息到来
127.0.0.1:6379> xread block 0 count 1 streams codehole $
# 咱们重新打开一个窗口，在这个窗口往Stream里塞消息
127.0.0.1:6379> xadd codehole * name youming age 60
1527852774092-0
# 再切换到前面的窗口，咱们能够看到阻塞解除了，返回了新的消息内容
# 并且还显示了一个等待时间，这里咱们等待了93s
127.0.0.1:6379> xread block 0 count 1 streams codehole $
1) 1) "codehole"
   2) 1) 1) 1527852774092-0
         2) 1) "name"
            2) "youming"
            3) "age"
            4) "60"
(93.11s)
复制代码

客户端若是想要使用xread进行顺序消费，必定要记住当前消费到哪里了，也就是返回的消息ID。下次继续调用xread时，将上次返回的最后一个消息ID做为参数传递进去，就能够继续消费后续的消息。

block 0表示永远阻塞，直到消息到来，block 1000表示阻塞1s，若是1s内没有任何消息到来，就返回nil

127.0.0.1:6379> xread block 1000 count 1 streams codehole $
(nil)
(1.07s)
复制代码

建立消费组

Stream经过 xgroup create指令建立消费组(Consumer Group)，须要传递起始消息ID参数用来初始化 last_delivered_id变量。

# 表示从头开始消费
127.0.0.1:6379> xgroup create codehole cg1 0-0
OK
# $表示从尾部开始消费，只接受新消息，当前Stream消息会所有忽略
127.0.0.1:6379> xgroup create codehole cg2 $
OK
# 获取Stream信息
127.0.0.1:6379> xinfo codehole
 1) length
 2) (integer) 3  # 共3个消息
 3) radix-tree-keys
 4) (integer) 1
 5) radix-tree-nodes
 6) (integer) 2
 7) groups
 8) (integer) 2  # 两个消费组
 9) first-entry  # 第一个消息
10) 1) 1527851486781-0
    2) 1) "name"
       2) "laoqian"
       3) "age"
       4) "30"
11) last-entry  # 最后一个消息
12) 1) 1527851498956-0
    2) 1) "name"
       2) "xiaoqian"
       3) "age"
       4) "1"
# 获取Stream的消费组信息
127.0.0.1:6379> xinfo groups codehole
1) 1) name
   2) "cg1"
   3) consumers
   4) (integer) 0  # 该消费组尚未消费者
   5) pending
   6) (integer) 0  # 该消费组没有正在处理的消息
2) 1) name
   2) "cg2"
   3) consumers  # 该消费组尚未消费者
   4) (integer) 0
   5) pending
   6) (integer) 0  # 该消费组没有正在处理的消息
复制代码

消费

Stream提供了xreadgroup指令能够进行消费组的组内消费，须要提供消费组名称、消费者名称和起始消息ID。它同xread同样，也能够阻塞等待新消息。读到新消息后，对应的消息ID就会进入消费者的PEL(正在处理的消息)结构里，客户端处理完毕后使用xack指令通知服务器，本条消息已经处理完毕，该消息ID就会从PEL中移除。

# >号表示从当前消费组的last_delivered_id后面开始读
# 每当消费者读取一条消息，last_delivered_id变量就会前进
127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >
1) 1) "codehole"
   2) 1) 1) 1527851486781-0
         2) 1) "name"
            2) "laoqian"
            3) "age"
            4) "30"
127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >
1) 1) "codehole"
   2) 1) 1) 1527851493405-0
         2) 1) "name"
            2) "yurui"
            3) "age"
            4) "29"
127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 2 streams codehole >
1) 1) "codehole"
   2) 1) 1) 1527851498956-0
         2) 1) "name"
            2) "xiaoqian"
            3) "age"
            4) "1"
      2) 1) 1527852774092-0
         2) 1) "name"
            2) "youming"
            3) "age"
            4) "60"
# 再继续读取，就没有新消息了
127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >
(nil)
# 那就阻塞等待吧
127.0.0.1:6379> xreadgroup GROUP cg1 c1 block 0 count 1 streams codehole >
# 开启另外一个窗口，往里塞消息
127.0.0.1:6379> xadd codehole * name lanying age 61
1527854062442-0
# 回到前一个窗口，发现阻塞解除，收到新消息了
127.0.0.1:6379> xreadgroup GROUP cg1 c1 block 0 count 1 streams codehole >
1) 1) "codehole"
   2) 1) 1) 1527854062442-0
         2) 1) "name"
            2) "lanying"
            3) "age"
            4) "61"
(36.54s)
# 观察消费组信息
127.0.0.1:6379> xinfo groups codehole
1) 1) name
   2) "cg1"
   3) consumers
   4) (integer) 1  # 一个消费者
   5) pending
   6) (integer) 5  # 共5条正在处理的信息还有没有ack
2) 1) name
   2) "cg2"
   3) consumers
   4) (integer) 0  # 消费组cg2没有任何变化，由于前面咱们一直在操纵cg1
   5) pending
   6) (integer) 0
# 若是同一个消费组有多个消费者，咱们能够经过xinfo consumers指令观察每一个消费者的状态
127.0.0.1:6379> xinfo consumers codehole cg1  # 目前还有1个消费者
1) 1) name
   2) "c1"
   3) pending
   4) (integer) 5  # 共5条待处理消息
   5) idle
   6) (integer) 418715  # 空闲了多长时间ms没有读取消息了
# 接下来咱们ack一条消息
127.0.0.1:6379> xack codehole cg1 1527851486781-0
(integer) 1
127.0.0.1:6379> xinfo consumers codehole cg1
1) 1) name
   2) "c1"
   3) pending
   4) (integer) 4  # 变成了5条
   5) idle
   6) (integer) 668504
# 下面ack全部消息
127.0.0.1:6379> xack codehole cg1 1527851493405-0 1527851498956-0 1527852774092-0 1527854062442-0
(integer) 4
127.0.0.1:6379> xinfo consumers codehole cg1
1) 1) name
   2) "c1"
   3) pending
   4) (integer) 0  # pel空了
   5) idle
   6) (integer) 745505
复制代码

Stream消息太多怎么办

读者很容易想到，要是消息积累太多，Stream的链表岂不是很长，内容会不会爆掉就是个问题了。xdel指令又不会删除消息，它只是给消息作了个标志位。

Redis天然考虑到了这一点，因此它提供了一个定长Stream功能。在xadd的指令提供一个定长长度maxlen，就能够将老的消息干掉，确保最多不超过指定长度。

127.0.0.1:6379> xlen codehole
(integer) 5
127.0.0.1:6379> xadd codehole maxlen 3 * name xiaorui age 1
1527855160273-0
127.0.0.1:6379> xlen codehole
(integer) 3
复制代码

咱们看到Stream的长度被砍掉了。

消息若是忘记ACK会怎样

Stream在每一个消费者结构中保存了正在处理中的消息ID列表PEL，若是消费者收到了消息处理完了可是没有回复ack，就会致使PEL列表不断增加，若是有不少消费组的话，那么这个PEL占用的内存就会放大。

PEL如何避免消息丢失

在客户端消费者读取Stream消息时，Redis服务器将消息回复给客户端的过程当中，客户端忽然断开了链接，消息就丢失了。可是PEL里已经保存了发出去的消息ID。待客户端从新连上以后，能够再次收到PEL中的消息ID列表。不过此时xreadgroup的起始消息ID不能为参数>，而必须是任意有效的消息ID，通常将参数设为0-0，表示读取全部的PEL消息以及自last_delivered_id以后的新消息。

结论

Stream的消费模型借鉴了kafka的消费分组的概念，它弥补了Redis Pub/Sub不能持久化消息的缺陷。可是它又不一样于kafka，kafka的消息能够分partition，而Stream不行。若是非要分parition的话，得在客户端作，提供不一样的Stream名称，对消息进行hash取模来选择往哪一个Stream里塞。若是读者稍微研究过Redis做者的另外一个开源项目Disque的话，这很可能是做者意识到Disque项目的活跃程度不够，因此将Disque的内容移植到了Redis里面。这只是本人的猜想，未必是做者的初衷。若是读者有什么不一样的想法，能够在评论区一块儿参与讨论。

阅读更多高级文章，关注公众号「码洞」