Redis Stream 用做消息队列完美吗

发布时间：2025-11-05 04:05:41 来源：云智核 作者：IT科技

这篇文章，列完分享笔者学习 Redis Stream 的做消心得，希望对大家有所启发。息队

图片

1 基础知识

Redis Stream 的列完结构如下图所示，它是做消一个消息链表，将所有加入的息队消息都串起来，每个消息都有一个唯一的列完 ID 和对应的内容。

图片

每个 Redis Stream 都有唯一的做消名称 ，对应唯一的息队 Redis Key 。

同一个 Stream 可以挂载多个消费者组 ConsumerGroup  ,列完 消费组不能自动创建，需要使用 XGROUP CREATE 命令创建。做消

每个消费组会有个游标 last_delivered_id，息队任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动 ，列完标识当前消费组消费到哪条消息了。

消费组 ConsumerGroup 同样可以挂载多个消费者 Consumer ,  每个 Consumer 并行的读取消息，任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动。

消费者内部有一个属性 pending_ids ， 记录了当前消费者读取但没有回复 ACK 的消息 ID 列表 。高防服务器

细品 Redis stream 的设计，我们发现它和 Kafka 非常相似，比如说消费者组，消费进度偏移量等。

但内部实现差异还是很明显，Kafka 主要是利用了文件系统的特性，每一个主题下存在多个分区 partition，而消费者处理对应分区消息成功之后，消费者会 commit 消费进度到 Broker ，Broker 修改消费进度偏移量并持久化。

2 核心命令

01 XADD 向 Stream 末尾添加消息

使用 XADD 向队列添加消息，如果指定的队列不存在，则创建一个队列。基础语法格式：

消息 ID 使用 * 表示由 redis 生成，同时也可以自定义，但是自定义时要保证递增性。

消息 ID 的格式：毫秒级时间戳 + 序号 , 例如：1712473185388-5 , 它表示当前消息在毫秒时间戳 1712473185388 产生 ，并且该毫秒内产生到了第5条消息。

在添加队列消息时，亿华云也可以指定队列的长度。

使用  XADD 命令向 mystream 的 stream 中添加了一条消息，并且指定了最大长度为 100。消息的 ID 由 Redis 自动生成，消息包含两个字段 name 和 age，分别对应的值是 value1 和 30。

02 XRANGE 获取消息列表

使用 XRANGE 获取消息列表，会自动过滤已经删除的消息。语法格式：

我们得到两条消息，第一层是消息 ID ，第二层是消息内容 ，消息内容是 Hash 数据结构 。

03 XREAD 以阻塞/非阻塞方式获取消息列表

使用 XREAD 以阻塞或非阻塞方式获取消息列表 ，语法格式：

XRED 读消息时分为阻塞和非阻塞模式，使用 BLOCK 选项可以表示阻塞模式，需要设置阻塞时长。非阻塞模式下，读取完毕（即使没有任何消息）立即返回，而在阻塞模式下，网站模板若读取不到内容，则阻塞等待。

使用 Block 模式，配合 $ 作为 ID ，表示读取最新的消息，若没有消息，命令阻塞！等待过程中，其他客户端向队列追加消息，则会立即读取到。

因此，典型的队列就是 XADD 配合 XREAD Block 完成。XADD 负责生成消息，XREAD 负责消费消息。

04 XGROUP CREATE 创建消费者组

使用 XGROUP CREATE 创建消费者组，分两种情况：

执行效果如下：

05 XREADGROUP GROUP 读取消费组中的消息

使用 XREADGROUP GROUP 读取消费组中的消息，语法格式：

示例：

消费者组 mygroup 中的消费者 consumerA ，从 名为 mystream 的 Stream 中读取消息。

06 XACK 消息消费确认

接收到消息之后，我们要手动确认一下（ack），语法格式：

示例：

消费确认增加了消息的可靠性，一般在业务处理完成之后，需要执行 ack 确认消息已经被消费完成，整个流程的执行如下图所示：

图片

我们可以使用 xpending 命令查看消费者未确认的消息ID：

07 XTRIM 限制 Stream 长度

我们使用 XTRIM 对流进行修剪，限制长度， 语法格式：

通过学习这七个核心命令，笔者发现 Redis Stream 既支持简单的生产消费模型，也支持发布订阅模型。

3 SpringBoot Redis Stream 实战

演示代码地址：

https://github.com/makemyownlife/courage-cache-demo

1、添加 SpringBoot Redis 依赖

2、yaml 文件配置

图片

3、RedisTemplate 配置

图片

4、定义stream监听器

图片

5、定义 StreamContainer 并启动

图片

6、发送消息

图片

执行完成之后，消费者就可以打印如下日志：

图片

通过实战流程，我们发现 SpringBoot 整合 Redis Stream 并不复杂，需要注意：定义注册信息时，需要确保 stream key 和消费者组已经创建好了。

4 Redis stream 用做消息队列完美吗

笔者认为 Redis stream 用于消息队列最大的进步在于：实现了发布订阅模型。

发布订阅模型具有如下特点：

我们曾经诟病 Redis List 数据结构用做队列时，因为消费时没有 Ack 机制，应用异常挂掉导致消息偶发丢失的情况，Redis Stream 从设计角度来讲已经完美的解决了。

因为消费者内部有一个属性 pending_ids ， 记录了当前消费者读取但没有回复 ACK 的消息 ID 列表 。当消费者重新上线，这些消息可以重新被消费。

但 Redis stream 用做消息队列完美吗 ？

这个还真没有，原因在于 Redis 先天不足。

1、Redis 本身定位是内存数据库，它的设计之初都是为缓存准备的，并不具备消息堆积的能力。而专业消息队列一个非常重要的功能是数据中转枢纽，Redis 的定位很难满足，使用起来要非常小心。

2、Redis 的高可用方案可能丢失消息（AOF 持久化和主从复制都是异步 ），而专业消息队列可以针对不同的场景选择不同的高可用策略。也就说，专业消息队列的高可用方案依然有可能会丢失消息，但是它有完善的方案支持不丢失消息。

所以，笔者认为 Redis 非常适合轻量级消息队列解决方案，即：数据量可控  + 业务模型简单 + 消息堆积概率低 + 报警监控 。

相关文章