Redis之Stream

2022-06-14 02:00:35

场景

Redis使用list可以实现队列功能，但无法做到广播模式，使用PubSub可以做到广播模式，但是PubSub并不能保证数据的持久化。

Stream就是为了满足上面的需求在Redis 5.0中发布的，它的内部结构是一个链表，将消息都串起来，每个消息拥有一个自己的ID和内容，该结构是持久化保存的。

基本概念

每个Stream都以key作为自己的名字，相当于消息队列中的topic，生产者和消费者对这个key进行发布订阅即可，Stream在第一次使用 xadd 指令时被创建。

消费组

每个Stream可以有多个消费组(Consumer Group)，每个消费组会有个游标(last_delivered_id)在Stream链表上往前移动，表示该消费组消费到了哪条消息。

消费组需要通过 xgroup create 进行创建，需要指定从Stream某个消息ID开始消费，该ID就是游标(last_delivered_id)的初始值。

每个消费者之间是独立的，A消费了数据1，不会影响到B消费数据1，即一份数据可以被多个消费组消费。

消费者

一个消费组可以有多个消费者(Consumer)，这下消费者之间是竞争关系，任意一个消费者消费数据后，都会导致游标(last_delivered_id)向前移动，每个消费者都有一个组内唯一名称。

消费者内部有一个状态变量： pending_ids，该变量记录已经被客户端读取，但是没有ACK的消息，客户端没有ACK的消息越多，这里保存的消息id就越多，ACK后将相应减少，

官方称 pending_ids 为 PEL(Pending Entires List)，用来确保客户端至少消费了一次，防止数据丢失。

消息ID与内容

消息ID格式为：timestamplnMillis-sequence，前者是消息产生的毫秒时间戳，后者是在该毫秒产生的第几条消息，例如 1527846880572-5 ，代表是 1527846880572 毫秒时的第 5 条消息，

消息ID可以由服务器自动生成，也可以是客户端自己指定，但是形式必须是整数-整数，后来的消息ID必须大于前面的消息ID。

消息内容是一个类似hash结构的键值对。

基本命令

xadd：向Stream发一条消息

# stream-1 表示stream的名字，* 则是系统该自动生成id，后面的则是我们的数据键值对

127.0.0.1:6379> xadd stream-1 * name zhangsan age 10

"1641454184593-0"

127.0.0.1:6379> xadd stream-1 * name lisi age 10

"1641454195771-0"

127.0.0.1:6379> xadd stream-1 * name wangwu age 10

"1641454208016-0"

xdel：从Stream中删除一条消息

127.0.0.1:6379> xdel stream-1 1641454195771-0

(integer) 0

xrange：获取Stream中的消息列表

127.0.0.1:6379> xrange stream-1 - +

1) 1) "1641454639427-0"

   2) 1) "name"

      2) "zhangsan"

      3) "age"

      4) "10"

2) 1) "1641454646977-0"

   2) 1) "name"

      2) "lisi"

      3) "age"

      4) "10"

3) 1) "1641454651750-0"

   2) 1) "name"

      2) "wangwu"

      3) "age"

      4) "10"

xlen：获取Stream消息长度

127.0.0.1:6379> xlen stream-1

(integer) 3

del：删除整个Stream消息列表

127.0.0.1:6379> del stream-1

(integer) 1

127.0.0.1:6379> xrange stream-1 - +

(empty array)

消费模式

消费模式有两种，独立消费与消费组，前者就是一个消息只能被全局消费一次，后者则是多个消费组可以对同一个消费者消费一次。

独立消费

Stream 提供了一个 xread 指令，该指令在消费消息的时候会忽略消费组的存在。

使用 xread 从头部读取两条消息


127.0.0.1:6379> xread count 2 streams stream-1 0-0

1) 1) "stream-1"

   2) 1) 1) "1641455116178-0"

         2) 1) "name"

            2) "zhangsan"

            3) "age"

            4) "10"

      2) 1) "1641455118418-0"

         2) 1) "name"

            2) "zhangsan1"

            3) "age"

            4) "10"

从尾部接收最新消息，之前的消息全部会被忽略

这种接收最新消息一般配合阻塞使用，如果不使用阻塞大概率读的是空

127.0.0.1:6379> xread count 1 streams stream-1 $

(nil)

阻塞读取，当有xadd往该stream-1添加消息时，将立刻返回，block以s为单位，阻塞s表后没有数据将返回，block 0 就是永远阻塞，直到有消息来临。

127.0.0.1:6379> xread block 0 count 1 streams stream-1 $

1) 1) "stream-1"

   2) 1) 1) "1641455579648-0"

         2) 1) "name"

            2) "wangliu"

            3) "age"

            4) "20"

(38.07s)

消费组

通过 xgroup create 可以创建一个消费组，并提供一个消息起始id初始化 last_delivered_id 变量。

为 stream-1 添加一个名为 cgroup-1 的消费组，从第一条消息开始消费。

xgroup create stream-1 cgroup-1 0-0

为 stream-1 添加一个名为 cgroup-2 的消费组，该消费组只监听新消息

xgroup create stream-1 cgroup-2 $

查看 stream-1 的详细信息

127.0.0.1:6379> xinfo stream stream-1

 1) "length"

 2) (integer) 4 # 共4个消息

 3) "radix-tree-keys"

 4) (integer) 1

 5) "radix-tree-nodes"

 6) (integer) 2

 7) "last-generated-id"

 8) "1641455579648-0"

 9) "groups"

10) (integer) 2 # 两个消费组

11) "first-entry" # 第一个消息

12) 1) "1641455116178-0"

    2) 1) "name"

       2) "zhangsan"

       3) "age"

       4) "10"

13) "last-entry" # 最后一个消息

14) 1) "1641455579648-0"

    2) 1) "name"

       2) "wangliu"

       3) "age"

       4) "20"

消费

xreadgroup 用于消费组进行消费消息，需要提供消费组名称、消费者名称、起始消息ID，与 xread 一样，也可以进行阻塞等待新消息，

读到新消息时，该消息ID将存入消费者的PEL中，处理完毕后客户端使用 xack 指令通知服务器，该消息ID将被从PEL中移除。

消费消息

使用消费组 cgroup-1 中的 c1 消费者读取 stream-1 中的一条消息，> 表示从当前消费组的 last_delivered_id 后面开始读，没读一条消息，该值就会往前进一个。

127.0.0.1:6379> xreadgroup GROUP cgroup-1 c1 count 1 streams stream-1 >

1) 1) "stream-1"

   2) 1) 1) "1641455579648-0"

         2) 1) "name"

            2) "wangliu"

            3) "age"

            4) "20"

127.0.0.1:6379> xreadgroup GROUP cgroup-1 c1 count 1 streams stream-1 >

(nil) 读完了

阻塞等待读取

xreadgroup GROUP cgroup-1 c1 block 0 count 1 streams stream-1 >

用另一个客户端发送消息

127.0.0.1:6379> xadd stream-1 * name mn age 30

"1641457442683-0"

读到新消息返回

127.0.0.1:6379> xreadgroup GROUP cgroup-1 c1 block 0 count 1 streams stream-1 >

1) 1) "stream-1"

   2) 1) 1) "1641457442683-0"

         2) 1) "name"

            2) "mn"

            3) "age"

            4) "30"

(45.20s)

观察消费组信息

127.0.0.1:6379> xinfo groups stream-1

1) 1) "name"

   2) "cgroup-1"

   3) "consumers"

   4) (integer) 1 # 一个消费者
 
   5) "pending"

   6) (integer) 5 # 5 条消息还没有ack

   7) "last-delivered-id"

   8) "1641457442683-0"

2) 1) "name"

   2) "cgroup-2"

   3) "consumers"

   4) (integer) 0

   5) "pending"

   6) (integer) 0

   7) "last-delivered-id"

   8) "1641455579648-0"

观察消费者信息


# cgroup-1下的消费者

127.0.0.1:6379> xinfo consumers stream-1 cgroup-1 

1) 1) "name"

   2) "c1" # c1 消费者

   3) "pending"

   4) (integer) 5 # 5条消息还没有ack

   5) "idle"

   6) (integer) 504275 # 空闲了多久 ms 没有读取消息

ack消息

# ack一条消息
xack stream-1 cgroup-1 1641455116178-0

# ack多条消息

127.0.0.1:6379> xack stream-1 cgroup-1 1641455118418-0 1641455120461-0 1641455579648-0 1641457442683-0

(integer) 4

5条消息全部ack后，待ack的数量变为0了。

127.0.0.1:6379> xinfo consumers stream-1 cgroup-1 

1) 1) "name"

   2) "c1"

   3) "pending"

   4) (integer) 0

   5) "idle"

   6) (integer) 684848

其他场景

消息定长

如果需要将Stream中的消息保持一定的长度，可以在xadd时，增加一个maxlen参数，该参数可以把老的消息剔除，保证消息长度不超过maxlen。

先增加5条数据

127.0.0.1:6379> xadd stream-1 * name n1

"1641458449368-0"

127.0.0.1:6379> xadd stream-1 * name n2

"1641458451353-0"

127.0.0.1:6379> xadd stream-1 * name n3

"1641458452737-0"

127.0.0.1:6379> xadd stream-1 * name n4

"1641458454740-0"

127.0.0.1:6379> xadd stream-1 * name n5

"1641458456357-0"

使用定长参数，设置长度为3，可以看到，当使用定长参数后，链表中的消息只有3个了。

127.0.0.1:6379> xadd stream-1 maxlen 3 * name n5

"1641458535493-0"

127.0.0.1:6379> xinfo stream stream-1

 1) "length"

 2) (integer) 3

 3) "radix-tree-keys"

 4) (integer) 1

 5) "radix-tree-nodes"

 6) (integer) 2

 7) "last-generated-id"

 8) "1641458535493-0"

 9) "groups"

10) (integer) 2

11) "first-entry"

12) 1) "1641458454740-0"

    2) 1) "name"

       2) "n4"

13) "last-entry"

14) 1) "1641458535493-0"

    2) 1) "name"

       2) "n5"

忘记ACK

消费者的PEL保存了没有ACK的消息ID，如果没有ACK的消息越来越多，那么这个结构就会占用空间越来越大，所以千万不要忘记ACK。

PEL防止消息丢失

PEL主要的作用是，如果客户端消费了一条消息，此时发生网络断开等原因，这个消息还没被客户端消费就丢失了，因此PEL可以保证，在没有ACK前，一直在PEL中保存该消息ID，等客户端重新连接上后，可以再次消费PEL中的消息ID，一般建议客户端将参数设置成 0-0，表示从laster_delivered_id后开始消费。

Stream高可用

Stream与其他数据结构并没有区别，它的高可用也是通过建立在主从复制上，在 Cluster 和 Sentinel 即可支持高可用，由于Redis指令复制是异步的，因此可能会发生丢失极小部分的数据丢失。

分区Partition

Stream没有提供分区的能力，如果需要实现的话，可以建立多个Stream结构，然后客户端通过对key进行hash的方式，路由到不同的Stream分区中。

码农公寓

场景