Redis - 消息队列 Stream

一、概述

消息队列

1. 定义
   1. 消息队列模型：一种分布式系统中的消息传递方案，由消息队列、生产者和消费者组成
   2. 消息队列：负责存储和管理消息的中间件，也称为消息代理（Message Broker）
   3. 生产者：负责 产生并发送 消息到队列的应用程序
   4. 消费者：负责从队列 获取并处理 消息的应用程序
2. 功能：实现消息发送和处理的解耦，支持异步通信，提高系统的可扩展性和可靠性
3. 主流消息队列解决方案
   1. RabbitMQ：轻量级，支持多种协议，适合中小规模应用
   2. RocketMQ：阿里开源，高性能，适合大规模分布式应用

Stream

1. 定义：Stream：Redis 5.0 引入的一种数据类型，用于处理高吞吐量的消息流、事件流等场景
2. 功能：按时间顺序 ”添加、读取、消费“ 消息，支持消费者组、消息确认等功能

二、Stream 工作流程

1. 写入消息：
   1. 生产者通过 XADD 向 Stream 中添加消息。每条消息自动获得唯一的 ID，按时间顺序存入 Stream。
2. 创建消费者组
   1. 如果使用消费者组，首先需要通过 XGROUP CREATE 创建消费者组。
   2. 消费者组会根据时间顺序将消息分配给组内的消费者。
3. 读取消息：
   1. 消费者使用 XREADGROUP 命令读取 Stream 中的消息。
   2. 消息按规则分配给不同消费者处理，每个消费者读取到不同的消息。
4. 确认消息：
   1. 消费者在处理完消息后，使用 XACK 命令确认消息，表示该消息已成功处理。
   2. 如果消息未确认（例如消费者崩溃或超时），它将保持在 Pending 状态，等待重新分配给其他消费者。
5. 重新分配未确认消息：
   1. 如果消息在一定时间内没有被确认，其他消费者可以读取未确认的消息并进行处理。
   2. 可通过 XPENDING 命令查看未确认消息，或在消费者组中设置时间阈值自动重新分配。
6. 删除消费者组：
   1. 不再需要消费者组时，使用 XGROUP DESTROY 命令删除消费者组

三、Stream 实现

消费者组模式

1. 定义：Redis Streams 的一部分，用于处理消息的分布式消费
2. 优点
   1. 消息分流：多消费者争抢消息，加快消费速度，避免消息堆积
   2. 消息标示：避免消息漏读，消费者读取消息后不马上销毁，加入 consumerGroup 维护的 pending list 队列等待 ACK
   3. 消息确认：通过消息 ACK 机制，保证消息至少被消费一次
   4. 可以阻塞读取，避免盲等
3. 实现方法 ：通过 Stream 数据类型实现消息队列，命令以 “X” 开头

常用命令

XGROUP CREATE key groupName ID [MKSTREAM]

1. 功能：创建消费者组
2. 参数
   1. key：队列名称
   2. groupName：组名称
   3. ID：起始 ID 标识，$ 表示队列中最后一个消息，0 表示队列中第一个消息
   4. MKSTREAM：队列不存在则创建队列

XGROUP DESTORY key groupName

1. 功能：删除指定消费者组

XGROUP CREATECONSUMER key groupName consumerName

1. 功能：添加组中消费者

XGROUP DELCONSUMER key groupName consumerName

1. 功能：删除组中消费者

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key …] ID [ID …]

1. 功能：读取组中的消息
2. gourp：消费者组名称
3. consumer：消费者名称(不存在则自动创建)
4. count：本次查询的最大数量
5. BLOCK milliseconds：当没有消息时最长等待时间
6. NOACK：无需手动 ACK，获取到消息后自动确认
7. STREAMS KEY：指定队列名称
8. ID：获取消息的起始 ID，> 表示从下一个未消费消息开始 (常用)

XPENDING key group [ [ IDLE min-idle-time ] start end count [consumer] ]

1. 功能：获取 pending-list 中的消息
2. IDLE：获取消息后、确认消息前的这段时间，空闲时间超过 min-idle-time 则取出
3. start：获取的最小目标 ID
4. end：获取的最大目标 ID
5. count：获取的数量
6. consumer：获取 consumer 的 pending-list

XACK key group ID [ ID … ]

1. 功能：确认从组中读取的消息已被处理
2. key：队列名称
3. group：组名称
4. ID：消息的 ID

表格版命令

1. 命令

   命令	功能
   XGROUP CREATE key groupName ID [MKSTREAM]	创建消费者组
   XGROUP DESTORY key groupName	删除指定消费者组
   XGROUP CREATECONSUMER key groupName consumerName	添加组中消费者
   XGROUP DELCONSUMER key groupName consumerName	删除组中消费者
   XREADGROUP GROUP groupName consumerName [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key …] ID [ID …]	读取组中的消息，ID 填写 “>” 则读取第一条未读消息
   XACK key group ID [ ID … ]	确认从组中读取的消息已被处理

2. 属性

   属性名	定义
   key	队列名称
   groupName	消费者组名称
   ID	起始 ID 标示，$ 代表队列中最后一个消息，0 代表第一个消息
   MKSTREAM	队列不存在时自动创建队列
   BLOCK milliseconds	没有消息时的最大等待时长
   NOACK	无需手动 ACK，获取到消息后自动确认
   STREAMS key	指定队列名称

运行逻辑

while(true) {// 尝试监听队列，使用阻塞模式，最长等待 2000 msObject msg = redis.call("XREADGROUP GROUP group1 consumer1 COUNT 1 BLOCK 2000 STREAMS s1 >");if(msg == null) {continue;}try {// 处理消息，完成后一定要 ACKhandleMessage(msg);} catch (Exception e) {while(true) {// 重新读取阻塞队列消息Object msg = redis.call("XREADGROUP GROUP group1 consumer1 COUNT 1 STREAM S1 0");if(msg == null)               // 如果阻塞队中的消息已经全部处理则退出pending-listbreak;try {handleMessage(msg);    			// 重新处理 pending-list 中的消息} catch (Exception e){continue;                   // 如果还出错, 则继续重新读取}}}
}

四、示例

1. 目标：消息队列实现数据库异步修改数据库，将下单 message 缓存在 redis 中，减小下单操作对数据库的冲击

2. 项目结构

   1. RedisConfig 配置类：创建消费者组是一次性的操作，适合放在配置类中
   2. VoucherOrderHandler 内部类：消费者的逻辑和订单业务相关，因此适合放在 VoucherOrderServiceImpl 中
   3. 多线程启动逻辑：消费者线程的启动与订单业务密切相关，直接放在 VoucherOrderServiceImpl 类中更符合职责分离原则

   src/main/java
   ├── com/example
   │   ├── config
   │   │   └── RedisConfig.java                  // Redis 配置类，包含消费者组初始化
   │   ├── service
   │   │   ├── VoucherOrderService.java
   │   │   └── impl
   │   │       └── VoucherOrderServiceImpl.java  // 包含 VoucherOrderHandler 内部类
   │   ├── entity
   │   │   └── VoucherOrder.java                 // 优惠券订单实体
   │   ├── utils
   │   │   └── BeanUtil.java                     // 用于 Map 转 Bean 的工具类
   │   └── controller
   │       └── VoucherOrderController.java       // 如果有 Controller
   

3. 创建消费者组（config.RedisConfig）

   @Bean
   public void initStreamGroup() {// 检查是否存在消费者组 g1try {stringRedisTemplate.opsForStream().createGroup("stream.orders", "g1");} catch (RedisSystemException e) {// 如果 group 已存在，抛出异常，可忽略log.warn("消费者组 g1 已存在");}
   }
   

4. 创建消费者线程

   1. 位置：作为 VoucherOrderServiceImpl 内的预构造部分

   @PostConstruct
   public void startConsumers() {for (int i = 0; i < 5; i++) { // 5 个线程，模拟多个消费者new Thread(new VoucherOrderHandler()).start();}
   }
   

5. 添加消息到消息队列 （src/main/resources/lua/SECKILL_SCRIPT.lua）

   --1. 参数列表
   --1.1. 优惠券id
   local voucherId = ARGV[1]
   --1.2. 用户id
   local userId = ARGV[2]
   --1.3. 订单id
   local orderId = ARGV[3]--2. 数据key
   local stockKey = 'seckill:stock:' .. voucherId          --2.1. 库存key
   local orderKey = 'seckill:order' .. voucherId           --2.2. 订单key--3. 脚本业务
   --3.1. 判断库存是否充足 get stockKey
   if( tonumber( redis.call('GET', stockKey) ) <= 0 ) thenreturn 1
   end
   --3.2. 判断用户是否重复下单 SISMEMBER orderKey userId
   if( redis.call( 'SISMEMBER', orderKey, userId ) == 1 ) thenreturn 2
   end
   --3.4 扣库存 incrby stockKey -1
   redis.call( 'INCRBY', stockKey, -1 )
   --3.5 下单(保存用户) sadd orderKey userId
   redis.call( 'SADD', orderKey, userId )
   -- 3.6. 发送消息到队列中
   redis.call( 'XADD', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId )
   

6. 创建消费者类（ServiceImpl）

   1. 位置：作为 VoucherOrderServiceImpl 内的私有类

   // 在ServiceImpl中创建一个VoucherOrderHandler消费者类,专门用于处理消息队列中的消息
   private class VoucherOrderHandler implements Runnable {@Overridepublic void run() {while (true) {try {// 1. 获取消息队列中的订单信息List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create( "stream.order", ReadOffset.lastConsumed()));// 2. 没有消息则重新监听if (list == null || list.isEmpty() ) continue;// 3. 获取消息中的 voucherOrderMapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> value = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 4. 创建订单createVoucherOrder(voucherOrder);// 5. 确认当前消息已消费 XACKstringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());} catch ( Exception e) {log.error("处理订单异常", e);// 6. 处理订单失败则消息会加入pending-list,继续处理pending-listhandlePendingList();}}}// 处理pending-list中的消息private void handlePendingList() {while(true) {try {// 1. 消费pending-list中的消息List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),                                   // 消费者此消息的消费者StreamReadOptions.empty().count(1),                          // StreamOffset.create("stream.order", ReadOffset.from("0"))     // 从pending-list的第一条消息开始读);// 2. 退出条件, list 为空 -> pending-list 已全部处理if(list == null || list.isEmpty()) break;// 3. 获取消息中的 voucherOrderMapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> value = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 4. 创建订单createVoucherOrder(voucherOrder);// 5. 确认消息已消费(XACK)stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());} catch (Exception e) {log.error("处理pendding订单异常", e);try{Thread.sleep(20);     // 如果发生异常则休眠一会再重新消费pending-list中的消息} catch (Exception e2) {e.printStackTrace(); }}}}
   }
   

7. 创建消息方法

   1. 目标：用户通过这个方法发送一条创建订单的 Message 给 Redis Stream

   // 创建Lua脚本对象
   private static final DefaultRedisScript<Long> SECKILL_SCRIPT;// Lua脚本初始化 (通过静态代码块)
   static {SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("lua/SECKILL_SCRIPT.lua"));SECKILL_SCRIPT.setResultType(Long.class);
   }@Override
   public void createVoucherOrder(Long voucherId, Long userId) {// 生成订单 ID（模拟）long orderId = System.currentTimeMillis();// 执行 Lua 脚本Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),                    // 使用空的 key 列表voucherId.toString(), userId.toString(), String.valueOf(orderId));// 根据 Lua 脚本返回结果处理if (result == 1) {throw new RuntimeException("库存不足！");} else if (result == 2) {throw new RuntimeException("不能重复下单！");}// 如果脚本执行成功，则订单消息会进入 Redis Stream，消费者组会自动处理System.out.println("订单创建成功！");
   }
   

(缺陷) 单消费者模式

1. 常用命令
   1. XADD key [NOMKSTREAM] [MAXLEN | MINID [=|~] threshold [LIMIT count] * | ID field value [field value …]
   2. XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key …] ID [ ID … ]
2. 缺陷：有消息漏读风险

---

五、其他消息队列方案

(缺陷) List 实现

1. 优点
   1. 不受 JVM 内存上限限制：因为利用 Redis 存储
   2. 数据安全 ：因为基于 List 结构本身是数据存储，基于 Redis 持久化机制
   3. 消息有序性：通过 List 结构的 LPUSH & BRPOP 命令实现顺序
2. 缺点
   1. 消息丢失：BRPOP 的时候如果宕机则消息会丢失
   2. 只支持单消费者

(缺陷) PubSub 实现

1. 定义
   1. Publish & Subscribe 模型，一种消息队列模型
   2. 生产者向指定的 channel 来 public 消息
   3. 消费者从 subscribe 的 channel 中接收消息
2. 功能：支持多消费者模式，多个消费者可以同时 subscribe 一个 channel
3. 优点：采用发布订阅模型，支持多生产者、消费者
4. 缺点
   1. 不支持数据持久化
   2. 无法避免消息丢失
   3. 消息堆积有上限，超出时数据丢失

三种消息队列对比

---