Redis高级功能之 - 发布/订阅与Stream

2021-04-13  本文已影响0人  kyo1992

发布订阅

发布/ 订阅系统 是 Web 系统中比较常用的一个功能。简单点说就是 发布者发布消息，订阅者接受消息。
虽然可以使用一个 list 列表结构结合 lpush 和 rpop 来实现消息队列的功能，但是很难实现消息多播的功能，
即消息只能同时被一个消费者处理，为此Redis新增Pub/Sub模块实现消息多播功能。

简介

基于 list 结构的消息队列，是一种 Publisher 与 Consumer 点对点的强关联关系，Redis 为了消除这样的强关联，引入了另一种概念：频道 (channel)，将发布者和多个订阅者解耦。

当 Publisher 往 channel 中发布消息时，关注了指定 channel 的 Consumer 就能够同时受到消息。但这里的 问题是，消费者订阅一个频道是必须明确指定频道名称的，这意味着，如果我们想要 订阅多个 频道，那么就必须显式地关注多个 名称。

为了简化订阅的繁琐操作，Redis 提供了模式订阅的功能 Pattern Subscribe，这样就可以一次性关注多个频道 了，即使生产者新增了同模式的频道，消费者也可以立即受到消息。

常用命令
# 订阅频道：
SUBSCRIBE channel [channel ....]   # 订阅给定的一个或多个频道的信息
PSUBSCRIBE pattern [pattern ....]  # 订阅一个或多个符合给定模式的频道
# 发布频道：
PUBLISH channel message  # 将消息发送到指定的频道
# 退订频道：
UNSUBSCRIBE [channel [channel ....]]   # 退订指定的频道
PUNSUBSCRIBE [pattern [pattern ....]]  # 退订所有给定模式的频道
实验

consumer1 订阅 abc.com 频道

127.0.0.1:6379> SUBSCRIBE abc.com
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "abc.com"
3) (integer) 1

consumer2 模式订阅 abc.* 频道

127.0.0.1:6379> PSUBSCRIBE abc.*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "abc.*"
3) (integer) 1

publisher 发布消息到 abc.com 频道

127.0.0.1:6379> PUBLISH abc.com "hello"
(integer) 2

此时consumer1 和 consumer2均收到消息

新增consumer3，订阅abc.net 频道

127.0.0.1:6379> SUBSCRIBE abc.net
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "abc.net"
3) (integer) 1

publisher发布消息到 abc.net 频道

127.0.0.1:6379> PUBLISH abc.net "hello2"
(integer) 2

此时consumer2 和 consumer3均收到消息，consumer1不会收到消息。

Pub/Sub缺点

Stream | 持久化的发布/订阅系统

Redis Stream 从概念上来说，就像是一个 仅追加内容 的 消息链表，把所有加入的消息都一个一个串起来，每个消息都有一个唯一的 ID 和内容，这很简单，让它复杂的是从 Kafka 借鉴的另一种概念：消费者组(Consumer Group) (思路一致，实现不同)

Stream结构

Stream由一个类似list的消息链表，每个 Stream 都有唯一的名称，它就是 Redis 的 key，Stream上按顺序存放着带id的消息。

typedef ConsumerGroup{
    int last_delivered_id
    Consumer arr[]
}

typedef Consumer{
    int pending_ids[]
}
消息id

消息 ID 如果是由 XADD 命令返回自动创建的话，那么它的格式会像这样：timestampInMillis-sequence (毫秒时间戳-序列号)，例如 1527846880585-5，它表示当前的消息是在毫秒时间戳 1527846880585 时产生的，并且是该毫秒内产生的第 5 条消息。

这些 ID 的格式看起来有一些奇怪，为什么要使用时间来当做 ID 的一部分呢？ 一方面，我们要 满足 ID 自增 的属性，另一方面，也是为了 支持范围查找 的功能。由于 ID 和生成消息的时间有关，这样就使得在根据时间范围内查找时基本上是没有额外损耗的。

当然消息 ID 也可以由客户端自定义，但是形式必须是 "整数-整数"，而且后面加入的消息的 ID 必须要大于前面的消息 ID。

消息内容

消息内容就是普通的键值对，形如 hash 结构的键值对。

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

# *号表示服务器自动生成ID，后面顺序跟着一堆key value
127.0.0.1:6379> xadd mq * name kyo age 29 # #  名字叫kyo，年龄29岁
"1617866075083-0"
127.0.0.1:6379> xadd mq * name joe age 28
"1617866082114-0"
127.0.0.1:6379> xadd mq * name smith age 27
"1617866088370-0"
127.0.0.1:6379> xlen mq
(integer) 3
127.0.0.1:6379> xrange mq - +   # -表示最小值, +表示最大值
1) 1) "1617866075083-0"
   2) 1) "name"
      2) "kyo"
      3) "age"
      4) "29"
2) 1) "1617866082114-0"
   2) 1) "name"
      2) "joe"
      3) "age"
      4) "28"
3) 1) "1617866088370-0"
   2) 1) "name"
      2) "smith"
      3) "age"
      4) "27"
127.0.0.1:6379> xrange mq 1617866075083-0 +  # 指定最小消息ID的列表
1) 1) "1617866075083-0"
   2) 1) "name"
      2) "kyo"
      3) "age"
      4) "29"
2) 1) "1617866082114-0"
   2) 1) "name"
      2) "joe"
      3) "age"
      4) "28"
3) 1) "1617866088370-0"
   2) 1) "name"
      2) "smith"
      3) "age"
      4) "27"
127.0.0.1:6379> xrange mq - 1617866082114-0   # 指定最大消息ID的列表
1) 1) "1617866075083-0"
   2) 1) "name"
      2) "kyo"
      3) "age"
      4) "29"
2) 1) "1617866082114-0"
   2) 1) "name"
      2) "joe"
      3) "age"
      4) "28"
127.0.0.1:6379> xdel mq 1617866082114-0
(integer) 1
127.0.0.1:6379> xlen mq  
(integer) 2
127.0.0.1:6379> xrange mq - +  # 被删除的消息没了
1) 1) "1617866075083-0"
   2) 1) "name"
      2) "kyo"
      3) "age"
      4) "29"
2) 1) "1617866088370-0"
   2) 1) "name"
      2) "smith"
      3) "age"
      4) "27"
127.0.0.1:6379> del mq   # 删除Stream
(integer) 1
实验 - 独立消费

我们可以在不定义消费组的情况下进行 Stream 消息的独立消费，当 Stream 没有新消息时，甚至可以阻塞等待。Redis 设计了一个单独的消费指令 xread，可以将 Stream 当成普通的消息队列(list)来使用。使用 xread 时，我们可以完全忽略 消费组(Consumer Group) 的存在，就好比 Stream 就是一个普通的列表(list)：

# 从Stream头部读取两条消息
127.0.0.1:6379> xread count 2 streams mq 0-0
1) 1) "mq"
   2) 1) 1) "1617866523741-0"
         2) 1) "name"
            2) "kyo"
            3) "age"
            4) "29"
      2) 1) "1617866529567-0"
         2) 1) "name"
            2) "joe"
            3) "age"
            4) "28"
# 从Stream尾部非阻塞读取一条消息，毫无疑问，这里不会返回任何消息
127.0.0.1:6379> xread count 1 streams mq $
(nil)
# 从尾部阻塞等待新消息到来，下面的指令会堵住，直到新消息到来
127.0.0.1:6379> xread block 0 count 1 streams mq $
# 我们从新打开一个窗口，在这个窗口往Stream里塞消息
127.0.0.1:6379> xadd mq * name smith age 27
"1617866635743-0"
# 再切换到前面的窗口，可以看到阻塞解除了，返回了新的消息内容
# 而且还显示了一个等待时间，这里我们等待了8.41s
1) 1) "mq"
   2) 1) 1) "1617866635743-0"
         2) 1) "name"
            2) "smith"
            3) "age"
            4) "27"
(8.41s)

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

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

127.0.0.1:6379> xread block 1000 count 1 streams mq $
(nil)
(1.06s)
实验 - 创建消费者组

Stream 通过 xgroup create 指令创建消费组(Consumer Group)，需要传递起始消息 ID 参数用来初始化 last_delivered_id 变量：

#  表示从头开始消费
127.0.0.1:6379> xgroup create mq cg1 0-0 
OK
# $表示从尾部开始消费，只接受新消息，当前Stream消息会全部忽略
127.0.0.1:6379> xgroup create mq cg2 $
OK
# 获取Stream信息
127.0.0.1:6379> xinfo stream mq
 1) "length"
 2) (integer) 3  # 共3个消息
 3) "radix-tree-keys"
 4) (integer) 1
 5) "radix-tree-nodes"
 6) (integer) 2
 7) "last-generated-id"
 8) "1617866635743-0"
 9) "groups"
10) (integer) 2  # 两个消费组
11) "first-entry"  # 第一个消息
12) 1) "1617866523741-0"
    2) 1) "name"
       2) "kyo"
       3) "age"
       4) "29"
13) "last-entry"  # 最后一个消息
14) 1) "1617866635743-0"
    2) 1) "name"
       2) "smith"
       3) "age"
       4) "27"
# 获取Stream的消费组信息
127.0.0.1:6379> xinfo groups mq
1) 1) "name"
   2) "cg1"
   3) "consumers"
   4) (integer) 0   # 该消费组还没有消费者
   5) "pending"
   6) (integer) 0   # 该消费组没有正在处理的消息
   7) "last-delivered-id"
   8) "0-0"
2) 1) "name"
   2) "cg2"
   3) "consumers"
   4) (integer) 0   # 该消费组还没有消费者
   5) "pending"
   6) (integer) 0   # 该消费组没有正在处理的消息
   7) "last-delivered-id"
   8) "1617866635743-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 mq >
1) 1) "mq"
   2) 1) 1) "1617866523741-0"
         2) 1) "name"
            2) "kyo"
            3) "age"
            4) "29"
127.0.0.1:6379> xreadgroup group cg1 c1 count 1 streams mq >
1) 1) "mq"
   2) 1) 1) "1617866529567-0"
         2) 1) "name"
            2) "joe"
            3) "age"
            4) "28"
127.0.0.1:6379> xreadgroup group cg1 c1 count 2 streams mq >
1) 1) "mq"
   2) 1) 1) "1617866635743-0"
         2) 1) "name"
            2) "smith"
            3) "age"
            4) "27"
      2) 1) "1617867664100-0"
         2) 1) "name"
            2) "haha"
            3) "age"
            4) "100"
# 再继续读取，就没有新消息了
127.0.0.1:6379> xreadgroup group cg1 c1 count 1 streams mq >
(nil)
# 那就阻塞等待吧
127.0.0.1:6379> xreadgroup group cg1 c1 block 0 count 1 streams mq >
# 开启另一个窗口，往里塞消息
127.0.0.1:6379> xadd mq * name aruh age 67
"1617867784813-0"
# 回到前一个窗口，发现阻塞解除，收到新消息了
1) 1) "mq"
   2) 1) 1) "1617867784813-0"
         2) 1) "name"
            2) "aruh"
            3) "age"
            4) "67"
(43.70s)
# 观察消费组信息
127.0.0.1:6379> xinfo groups mq
1) 1) "name"
   2) "cg1"
   3) "consumers"
   4) (integer) 1  # 一个消费者
   5) "pending"
   6) (integer) 5  # 共5条正在处理的信息还有没有ack
   7) "last-delivered-id"
   8) "1617867784813-0"
2) 1) "name"
   2) "cg2"
   3) "consumers"
   4) (integer) 0   # 消费组cg2没有任何变化，因为前面我们一直在操纵cg1
   5) "pending"
   6) (integer) 0
   7) "last-delivered-id"
   8) "1617866635743-0"
# 如果同一个消费组有多个消费者，我们可以通过xinfo consumers指令观察每个消费者的状态
127.0.0.1:6379> xinfo consumers mq cg1
1) 1) "name"
   2) "c1"
   3) "pending"
   4) (integer) 5   共5条待处理消息
   5) "idle"
   6) (integer) 29945  # 空闲了多长时间ms没有读取消息了
127.0.0.1:6379> xack mq cg1 617867784813-0
(integer) 0
# 接下来我们ack一条消息
127.0.0.1:6379> xack mq cg1 1617867784813-0
(integer) 1
127.0.0.1:6379> xinfo consumers mq cg1
1) 1) "name"
   2) "c1"
   3) "pending"
   4) (integer) 4  # 变成了4条
   5) "idle"
   6) (integer) 101553
# 下面ack所有消息
127.0.0.1:6379> xack mq cg1 1617866529567-0 1617866523741-0 1617866635743-0 1617867664100-0
(integer) 4
127.0.0.1:6379> xinfo consumers mq cg1
1) 1) "name"
   2) "c1"
   3) "pending"
   4) (integer) 0   # pel空了
   5) "idle"
   6) (integer) 176571
127.0.0.1:6379>
问题1：Stream 消息太多怎么办？ | Stream 的上限

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

Redis 自然考虑到了这一点，所以它提供了一个定长 Stream 功能。在 xadd 的指令提供一个定长长度 maxlen，就可以将老的消息干掉，确保最多不超过指定长度，使用起来也很简单：

> XADD mystream MAXLEN 2 * value 1
1526654998691-0
> XADD mystream MAXLEN 2 * value 2
1526654999635-0
> XADD mystream MAXLEN 2 * value 3
1526655000369-0
> XLEN mystream
(integer) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

如果使用 MAXLEN 选项，当 Stream 的达到指定长度后，老的消息会自动被淘汰掉，因此 Stream 的大小是恒定的。目前还没有选项让 Stream 只保留给定数量的条目，因为为了一致地运行，这样的命令必须在很长一段时间内阻塞以淘汰消息。(例如在添加数据的高峰期间，你不得不长暂停来淘汰旧消息和添加新的消息)

另外使用 MAXLEN 选项的花销是很大的，Stream 为了节省内存空间，采用了一种特殊的结构表示，而这种结构的调整是需要额外的花销的。所以我们可以使用一种带有 ~ 的特殊命令：

XADD mystream MAXLEN ~ 1000 * ... entry fields here ...

它会基于当前的结构合理地对节点执行裁剪，来保证至少会有 1000 条数据，可能是 1010 也可能是 1030。

问题2：PEL 是如何避免消息丢失的

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

Redis Stream Vs Kafka

Redis Stream是基于内存存储，这意味着它会比基于磁盘的 Kafka 快上一些，也意味着使用 Redis 我们 不能长时间存储大量数据。不过如果您想以 最小延迟 实时处理消息的话，您可以考虑 Redis，但是如果 消息很大并且应该重用数据 的话，则应该首先考虑使用 Kafka。

另外从某些角度来说，Redis Stream 也更适用于小型、廉价的应用程序，因为 Kafka 相对来说更难配置一些。

上一篇下一篇

猜你喜欢

热点阅读