提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  数据库

当前栏目

把Redis当作队列来用，真的合适吗？中

Redis 消息 队列 队列
2023-03-14 22:50:09 时间

趋于成熟的队列：Stream

我们来看 Stream 是如何解决上面这些问题的。

我们依旧从简单到复杂，依次来看 Stream 在做消息队列时，是如何处理的？

首先，Stream 通过 XADD 和 XREAD 完成最简单的生产、消费模型：

XADD：发布消息

XREAD：读取消息

生产者发布 2 条消息：

// *表示让Redis自动生成消息ID
127.0.0.1:6379> XADD queue * name zhangsan
"1618469123380-0"
127.0.0.1:6379> XADD queue * name lisi
"1618469127777-0"

使用 XADD 命令发布消息，其中的「*」表示让 Redis 自动生成唯一的消息 ID。

这个消息 ID 的格式是「时间戳-自增序号」。

消费者拉取消息：

// 从开头读取5条消息，0-0表示从开头读取
127.0.0.1:6379> XREAD COUNT 5 STREAMS queue 0-0
1) 1) "queue"
   2) 1) 1) "1618469123380-0"
         2) 1) "name"
            2) "zhangsan"
      2) 1) "1618469127777-0"
         2) 1) "name"
            2) "lisi"

如果想继续拉取消息，需要传入上一条消息的 ID：

127.0.0.1:6379> XREAD COUNT 5 STREAMS queue 1618469127777-0
(nil)

没有消息，Redis 会返回 NULL。

image

以上就是 Stream 最简单的生产、消费。

这里不再重点介绍 Stream 命令的各种参数，我在例子中演示时，凡是大写的单词都是「固定」参数，凡是小写的单词，都是可以自己定义的，例如队列名、消息长度等等，下面的例子规则也是一样，为了方便你理解，这里有必要提醒一下。

下面我们来看，针对前面提到的消息队列要求，Stream 都是如何解决的？

1) Stream 是否支持「阻塞式」拉取消息？

可以的，在读取消息时，只需要增加 BLOCK 参数即可。

// BLOCK 0 表示阻塞等待，不设置超时时间
127.0.0.1:6379> XREAD COUNT 5 BLOCK 0 STREAMS queue 1618469127777-0

这时，消费者就会阻塞等待，直到生产者发布新的消息才会返回。

2) Stream 是否支持发布 / 订阅模式？

也没问题，Stream 通过以下命令完成发布订阅：

XGROUP：创建消费者组

XREADGROUP：在指定消费组下，开启消费者拉取消息

下面我们来看具体如何做？

首先，生产者依旧发布 2 条消息：

127.0.0.1:6379> XADD queue * name zhangsan
"1618470740565-0"
127.0.0.1:6379> XADD queue * name lisi
"1618470743793-0"

之后，我们想要开启 2 组消费者处理同一批数据，就需要创建 2 个消费者组：

// 创建消费者组1，0-0表示从头拉取消息
127.0.0.1:6379> XGROUP CREATE queue group1 0-0
OK
// 创建消费者组2，0-0表示从头拉取消息
127.0.0.1:6379> XGROUP CREATE queue group2 0-0
OK

消费者组创建好之后，我们可以给每个「消费者组」下面挂一个「消费者」，让它们分别处理同一批数据。

第一个消费组开始消费：

// group1的consumer开始消费，>表示拉取最新数据
127.0.0.1:6379> XREADGROUP GROUP group1 consumer COUNT 5 STREAMS queue >
1) 1) "queue"
   2) 1) 1) "1618470740565-0"
         2) 1) "name"
            2) "zhangsan"
      2) 1) "1618470743793-0"
         2) 1) "name"
            2) "lisi"

同样地，第二个消费组开始消费：

// group2的consumer开始消费，>表示拉取最新数据
127.0.0.1:6379> XREADGROUP GROUP group2 consumer COUNT 5 STREAMS queue >
1) 1) "queue"
   2) 1) 1) "1618470740565-0"
         2) 1) "name"
            2) "zhangsan"
      2) 1) "1618470743793-0"
         2) 1) "name"
            2) "lisi"

我们可以看到，这 2 组消费者，都可以获取同一批数据进行处理了。

这样一来，就达到了多组消费者「订阅」消费的目的。

image

3) 消息处理时异常，Stream 能否保证消息不丢失，重新消费？

除了上面拉取消息时用到了消息 ID，这里为了保证重新消费，也要用到这个消息 ID。

当一组消费者处理完消息后，需要执行 XACK 命令告知 Redis，这时 Redis 就会把这条消息标记为「处理完成」。

// group1下的 1618472043089-0 消息已处理完成
127.0.0.1:6379> XACK queue group1 1618472043089-0

image

如果消费者异常宕机，肯定不会发送 XACK，那么 Redis 就会依旧保留这条消息。

待这组消费者重新上线后，Redis 就会把之前没有处理成功的数据，重新发给这个消费者。这样一来，即使消费者异常，也不会丢失数据了。

// 消费者重新上线，0-0表示重新拉取未ACK的消息
127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 COUNT 5 STREAMS queue 0-0
// 之前没消费成功的数据，依旧可以重新消费
1) 1) "queue"
   2) 1) 1) "1618472043089-0"
         2) 1) "name"
            2) "zhangsan"
      2) 1) "1618472045158-0"
         2) 1) "name"
            2) "lisi"

4) Stream 数据会写入到 RDB 和 AOF 做持久化吗？

Stream 是新增加的数据类型，它与其它数据类型一样，每个写操作，也都会写入到 RDB 和 AOF 中。

我们只需要配置好持久化策略，这样的话，就算 Redis 宕机重启，Stream 中的数据也可以从 RDB 或 AOF 中恢复回来。

5) 消息堆积时，Stream 是怎么处理的？

其实，当消息队列发生消息堆积时，一般只有 2 个解决方案：

生产者限流：避免消费者处理不及时，导致持续积压

丢弃消息：中间件丢弃旧消息，只保留固定长度的新消息

而 Redis 在实现 Stream 时，采用了第 2 个方案。

在发布消息时，你可以指定队列的最大长度，防止队列积压导致内存爆炸。

// 队列长度最大10000
127.0.0.1:6379> XADD queue MAXLEN 10000 * name zhangsan
"1618473015018-0"

当队列长度超过上限后，旧消息会被删除，只保留固定长度的新消息。

这么来看，Stream 在消息积压时，如果指定了最大长度，还是有可能丢失消息的。

除了以上介绍到的命令，Stream 还支持查看消息长度（XLEN）、查看消费者状态（XINFO）等命令，使用也比较简单，你可以查询官方文档了解一下，这里就不过多介绍了。

好了，通过以上介绍，我们可以看到，Redis 的 Stream 几乎覆盖到了消息队列的各种场景，是不是觉得很完美？

既然它的功能这么强大，这是不是意味着，Redis 真的可以作为专业的消息队列中间件来使用呢？

但是还「差一点」，就算 Redis 能做到以上这些，也只是「趋近于」专业的消息队列。

原因在于 Redis 本身的一些问题，如果把其定位成消息队列，还是有些欠缺的。

到这里，就不得不把 Redis 与专业的队列中间件做对比了。

下面我们就来看一下，Redis 在作队列时，到底还有哪些欠缺？

相关文章