基于Redis的Stream结构作为消息队列，实现异步秒杀下单

梦里梦外; 2024-03-22 13:57 81阅读 0赞

#### 文章目录 ####

*   *  1 认识消息队列
     *  2 基于List实现消息队列
     *  3 基于PubSub的消息队列
     *  4 基于Stream的消息队列
     *  5 基于Stream的消息队列-消费者组
     *  6 基于Redis的Stream结构作为消息队列，实现异步秒杀下单

### 1 认识消息队列 ###

什么是消息队列：字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色：

*  消息队列：存储和管理消息，也被称为消息代理（Message Broker）
 *  生产者：发送消息到消息队列
 *  消费者：从消息队列获取消息并处理消息

![1653574849336][]

消息队列是一种常用的中间件，它可以用来实现**系统解耦**、**流量削峰**和**异步调用**等功能。

所谓**解耦**，举一个生活中的例子就是：快递员(生产者)把快递放到快递柜里边(Message Queue)去，我们(消费者)从快递柜里边去拿东西，这就是一个异步，如果耦合，那么这个快递员相当于直接把快递交给你，这事固然好，但是万一你不在家，那么快递员就会一直等你，这就浪费了快递员的时间，所以这种思想在我们日常开发中，是非常有必要的。

这种场景在我们秒杀中就变成了：我们下单之后，利用redis去进行校验下单条件，再通过队列把消息发送出去，然后再启动一个线程去消费这个消息，完成解耦，同时也加快我们的响应速度。

常见的消息队列中间件有`RabbitMQ`、`Kafka`和`RocketMQ`等。这些消息队列需要独立安装部署，虽然它们具有高性能和高可靠性的优点，但是额外部署这些中间件也会增加运维成本和服务器成本。

Redis也可以用来实现消息队列。Redis提供了几种不同的方式来实现消息队列，包括使用`List`、`ZSet`、`PubSub`模式和`Stream`等。这些方式各有优劣，适用于不同的应用场景。

*  使用List实现消息队列：这种方式最为简单直接，它主要通过lpush和rpop命令来存入和读取消息。使用List实现消息队列的优点是消息可以被持久化，但缺点是不支持重复消费、没有按照主题订阅的功能、不支持消费者确认机制等。
 *  使用ZSet实现消息队列：这种方式与使用List类似，但由于ZSet多了一个分值（score）属性，我们可以使用它来实现更多的功能，比如用它来存储时间戳，以此来实现延迟消息队列等。ZSet同样具备持久化的功能，但也存在与List类似的问题。
 *  使用发布订阅模式实现消息队列：这种方式可以实现主题订阅的功能。但缺点是无法持久化保存信息，如果Redis服务器宕机或者重启，那么所有的消息都会丢失；发布订阅模式是“发后即忘”的工作模式，如果都订阅者离线重连之后就不能消费之前的历史消息了；不支持消费者确认机制，稳定性不能得到保证。
 *  使用Stream实现消息队列：这种方式是在Redis 5.0之后新增的。我们可以使用Stream的xadd和xrange命令来实现消息的存入和读取，并且Stream提供了xack命令来手动确认消息消费，用它我们就可以实现消费者确认的功能。

### 2 基于List实现消息队列 ###

**基于List结构模拟消息队列**

消息队列（Message Queue），字面意思就是存放消息的队列。而Redis的list数据结构是一个双向链表，很容易模拟出队列效果。

队列是入口和出口不在一边，因此我们可以利用：LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP来实现。  
不过要注意的是，当队列中没有消息时RPOP或LPOP操作会返回null，并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。

![1653575176451][]

基于List的消息队列有哪些优缺点？

优点：

*  利用Redis存储，不受限于JVM内存上限
 *  基于Redis的持久化机制，数据安全性有保证
 *  可以满足消息有序性

缺点：

*  无法避免消息丢失
 *  只支持单消费者

### 3 基于PubSub的消息队列 ###

PubSub（发布订阅）是Redis2.0版本引入的消息传递模型。顾名思义，消费者可以订阅一个或多个channel，生产者向对应channel发送消息后，所有订阅者都能收到相关消息。

SUBSCRIBE channel \[channel\] ：订阅一个或多个频道  
PUBLISH channel msg ：向一个频道发送消息  
PSUBSCRIBE pattern\[pattern\] ：订阅与pattern格式匹配的所有频道

![1653575506373][]

基于PubSub的消息队列有哪些优缺点？

优点：

*  采用发布订阅模型，支持多生产、多消费

缺点：

*  不支持数据持久化
 *  无法避免消息丢失
 *  消息堆积有上限，超出时数据丢失

### 4 基于Stream的消息队列 ###

Stream 是 Redis 5.0 引入的一种新数据类型，可以实现一个功能非常完善的消息队列。

发送消息的命令：

![1653577301737][]

例如：

![1653577349691][]

读取消息的方式之一：XREAD

![1653577445413][]

例如，使用XREAD读取第一个消息：

![1653577643629][]

XREAD阻塞方式，读取最新的消息：

![1653577659166][]

在业务开发中，我们可以循环的调用XREAD阻塞方式来查询最新消息，从而实现持续监听队列的效果，伪代码如下

![1653577689129][]

注意：当我们指定起始ID为`$`时，代表读取最新的消息，如果我们处理一条消息的过程中，又有超过1条以上的消息到达队列，则下次获取时也只能获取到最新的一条，会出现漏读消息的问题

STREAM类型消息队列的XREAD命令特点：

*  消息可回溯
 *  一个消息可以被多个消费者读取
 *  可以阻塞读取
 *  有消息漏读的风险

### 5 基于Stream的消息队列-消费者组 ###

消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。具备下列特点：

![1653577801668][]

创建消费者组：  
![1653577984924][]  
key：队列名称  
groupName：消费者组名称  
ID：起始ID标示，`$`代表队列中最后一个消息，0则代表队列中第一个消息  
MKSTREAM：队列不存在时自动创建队列  
其它常见命令：

**删除指定的消费者组**

XGROUP DESTORY key groupName

**给指定的消费者组添加消费者**

XGROUP CREATECONSUMER key groupname consumername

**删除消费者组中的指定消费者**

XGROUP DELCONSUMER key groupname consumername

从消费者组读取消息：

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

*  group：消费组名称
 *  consumer：消费者名称，如果消费者不存在，会自动创建一个消费者
 *  count：本次查询的最大数量
 *  BLOCK milliseconds：当没有消息时最长等待时间
 *  NOACK：无需手动ACK，获取到消息后自动确认
 *  STREAMS key：指定队列名称
 *  ID：获取消息的起始ID：

`>`：从下一个未消费的消息开始  
其它：根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从pending-list中的第一个消息开始

消费者监听消息的基本思路：

![1653578211854][]

STREAM类型消息队列的XREADGROUP命令特点：

*  消息可回溯
 *  可以多消费者争抢消息，加快消费速度
 *  可以阻塞读取
 *  没有消息漏读的风险
 *  有消息确认机制，保证消息至少被消费一次

最后我们来个小对比

![1653578560691][]

### 6 基于Redis的Stream结构作为消息队列，实现异步秒杀下单 ###

需求：

*  创建一个Stream类型的消息队列，名为stream.orders
 *  修改之前的秒杀下单Lua脚本，在认定有抢购资格后，直接向stream.orders中添加消息，内容包含voucherId、userId、orderId
 *  项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单

修改lua表达式,新增3.6

![1656082824939][]

VoucherOrderServiceImpl

private class VoucherOrderHandler implements Runnable {
        
    
        @Override
        public void run() {
        
            while (true) {
        
                try {
        
                    // 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                    );
                    // 2.判断订单信息是否为空
                    if (list == null || list.isEmpty()) {
        
                        // 如果为null，说明没有消息，继续下一次循环
                        continue;
                    }
                    // 解析数据
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    // 3.创建订单
                    createVoucherOrder(voucherOrder);
                    // 4.确认消息 XACK
                    stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
                } catch (Exception e) {
        
                    log.error("处理订单异常", e);
                    //处理异常消息
                    handlePendingList();
                }
            }
        }
    
        private void handlePendingList() {
        
            while (true) {
        
                try {
        
                    // 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create("stream.orders", ReadOffset.from("0"))
                    );
                    // 2.判断订单信息是否为空
                    if (list == null || list.isEmpty()) {
        
                        // 如果为null，说明没有异常消息，结束循环
                        break;
                    }
                    // 解析数据
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    // 3.创建订单
                    createVoucherOrder(voucherOrder);
                    // 4.确认消息 XACK
                    stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
                } catch (Exception e) {
        
                    log.error("处理pendding订单异常", e);
                    try{
        
                        Thread.sleep(20);
                    }catch(Exception e){
        
                        e.printStackTrace();
                    }
                }
            }
        }
    }

[1653574849336]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/0e949ff6e41c4157b0cd9cc1db9159d8.png
[1653575176451]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/b6c75f4684c444fc82850d65d4d023a3.png
[1653575506373]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/8657573344d646c3821a03baab9537a6.png
[1653577301737]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/3402cca4559c458790508ce2c4011c7d.png
[1653577349691]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/562e739f535a49498c46a6c16f2f4431.png
[1653577445413]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/183e589d9db04fb08dbe0a943a98d0e5.png
[1653577643629]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/d567cd84b3da410d92467a76ebe89987.png
[1653577659166]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/c06644e70722432d9b3e6ad402798cc2.png
[1653577689129]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/3c73cf833fb54d1d9b772803661001d6.png
[1653577801668]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/15beb4922feb4429bdf34ddf0e469e3b.png
[1653577984924]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/93fb310abc4b4294aa5b2ab47c821cab.png
[1653578211854]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/21eabde7f0ad4c81b0fd119f47101155.png
[1653578560691]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/693bc1d182fa4e00a47804564593058c.png
[1656082824939]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/60b4eb6360ee49a488e300fb45159484.png