RabbitMQ 延迟队列实现订单自动关闭

约定不等于承诺〃 2022-03-16 00:42 748阅读 0赞

## 1、延迟队列 ##

延迟队列存储的对象肯定是对应的延时消息，所谓”延时消息”是指当消息被发送以后，并不想让消费者立即拿到消息，而是等待指定时间后，消费者才拿到这个消息进行消费。

在电商系统中，当用户提交订单超过 30 分钟未支付就是自动取消。一般的解决方案是使用定时任务来轮训数据库，然后找到过期的订单来修改订单状态。当系统的数据量小的时候下没有什么问题。但是如果数据量一大这种方式就会特别消耗资源。并且这种方式对于订单的取消还有延迟。  
而使用 RabbitMQ 的延迟队列就是解决这些问题的。

使用RabbitMQ来实现延迟队列必须先了解RabbitMQ的两个概念：消息的 TTL 和死信 Exchange，通过这两者的组合来实现上述需求。

## 2、RabbitMQ 实现迟队列 ##

AMQP协议和 RabbitMQ 队列本身没有直接支持延迟队列功能，但是可以通过以下 消息的TTL 与 死信 Exchange 这两个特性特性模拟出延迟队列的功能。

### 2.1 Time To Live(TTL) ###

消息的 TTL 就是消息的存活时间。RabbitMQ 可以对队列和消息分别设置TTL。RabbitMQ 可以针对 Queue 设置 x-expires 或者 针对 Message 设置 x-message-ttl，来控制消息的生存时间，如果超时(两者同时设置以最先到期的时间为准)，则消息变为dead letter(死信)。超过了这个时间，我们认为这个消息就死了，称之为死信。如果队列设置了，消息也设置了，那么会取小的。  
所以一个消息如果被路由到不同的队列中，这个消息死亡的时间有可能不一样（不同的队列设置）。这里单讲单个消息的TTL，因为它才是实现延迟任务的关键。

RabbitMQ针对队列中的消息过期时间有两种方法可以设置。

*  A: 通过队列属性设置，队列中所有消息都有相同的过期时间。
 *  B: 对消息进行单独设置，每条消息TTL可以不同。

如果同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值，就成为dead letter

### 2.2 Dead Letter Exchanges（DLX） ###

RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可选）两个参数，如果队列内出现了dead letter，则按照这两个参数重新路由转发到指定的队列。

*  x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange
 *  x-dead-letter-routing-key：出现dead letter之后将dead letter重新按照指定的routing-key发送

队列满足以下条件就会出现 dead letter 情况有：

*  消息或者队列的TTL过期
 *  队列达到最大长度
 *  消息被消费端拒绝（basic.reject or basic.nack）并且requeue = false

综合上述两个特性，设置了TTL规则之后当消息在一个队列中变成死信时，利用DLX特性它能被重新转发到另一个Exchange或者Routing Key，这时候消息就可以重新被消费了。

## 3、代码演示 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTA3MzM_size_16_color_FFFFFF_t_70]  
这里的场景是订单创建，当订单创建成功的时候会把消息发送到 RabbitMQ 的扇形交换机里面(发布/订阅模式)。因为订单创建成功有多个消息者：一个是订单创建成功需要向会员发送优惠券，别一个是订单创建成功如果多久没有支付就会关闭订单。这里就用到了 RabbitMQ 中的死信队列。

这个项目是基于 Spring Boot，如果大家不清楚 Spring Boot 的，可以自行了解一下。

### 3.1 项目依赖 ###

下面是定义在 xml 里面的项目依赖 Jar 包管理，使用的是 maven 进行项目管理。

> pom.xml

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    	<parent>
    		<groupId>org.springframework.boot</groupId>
    		<artifactId>spring-boot-starter-parent</artifactId>
    		<version>2.1.1.RELEASE</version>
    		<relativePath/>
    	</parent>
    	<groupId>cn.carlzone.netty</groupId>
    	<artifactId>netty-demo</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    	<name>netty-demo</name>
    	<description>Demo project for Spring Boot</description>
    
    	<properties>
    		<java.version>1.8</java.version>
    	</properties>
    
    	<dependencies>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-web</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-amqp</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-aop</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-test</artifactId>
    			<scope>test</scope>
    		</dependency>
    		<dependency>
    			<groupId>org.projectlombok</groupId>
    			<artifactId>lombok</artifactId>
    			<optional>true</optional>
    		</dependency>
    	</dependencies>
    
    	<build>
    		<plugins>
    			<plugin>
    				<groupId>org.springframework.boot</groupId>
    				<artifactId>spring-boot-maven-plugin</artifactId>
    			</plugin>
    		</plugins>
    	</build>
    
    </project>

### 3.2 application.properties ###

在 Spring 配置文件中配置需要依赖的 rabbitmq 服务相关的信息。

> appplication.properties

spring.rabbitmq.host=localhost
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

### 3.3 MQ 常量类 ###

这个类是定义 mq 相关的常量信息。

> MQConstant

public abstract class MQConstant {
    
        /** 订单创建 fanout(扇形) 交换机 */
        public final static String PAY_ORDER_CREATE_FANOUT_EXCHANGE = "pay.order.create.fanout.exchange";
    
        /** 订单创建 Dead Letter 交换机 */
        public final static String PAY_ORDER_CREATE_DEAD_LETTER_EXCHANGE = "pay.order.create.dead.letter.exchange";
    
        /** 订单创建优惠券订阅队列 */
        public final static String PAY_ORDER_CREATE_COUPON_QUEUE = "pay.order.create.coupon";
    
        /** 订单创建 6 秒不支付死信队列 (TTL QUEUE) */
        public final static String PAY_ORDER_CREATE_DEAD_LETTER_QUEUE = "pay.order.create.dead.letter";
    
        /** 死信转发队列(DLX QUEUE) */
        public final static String PAY_ORDER_CREATE_EXPIRE_QUEUE = "pay.order.create.expire";
    
        public final static String PAY_ORDER_CREATE_EXPIRE_ROUTING_KEY = "pay.order.create.expire.routing.key";
    
    }

这个类是抽象类，在定义常量或者工具类的时候最好把这个类定义成抽象类。因为常量类或者工具类都是不需要实例化的。

### 3.4 Rabbit MQ Java 配置类 ###

RabbitQueueConfiguration 是定义 RabbitMQ 里面的队列，交换机以及绑定信息。

@Configuration
    public class RabbitQueueConfiguration {
    
        @Bean
        public MessageConverter jackson2JsonMessageConverter(){
            return new Jackson2JsonMessageConverter();
        }
    
        /**
         * 订单创建扇形交换机
         * @return
         */
        @Bean
        public FanoutExchange orderCreateFanoutExchange(){
            return new FanoutExchange(MQConstant.PAY_ORDER_CREATE_FANOUT_EXCHANGE, true, false);
        }
    
        /**
         * 订单创建死信交换机
         * @return
         */
        @Bean
        public DirectExchange orderCreateDeadLetterExchange(){
            return new DirectExchange(MQConstant.PAY_ORDER_CREATE_DEAD_LETTER_EXCHANGE, true, false);
        }
    
        /**
         * 订单创建 -- 优惠券消费队列
         * @return
         */
        @Bean
        public Queue orderCreateCouponQueue(){
            return new Queue(MQConstant.PAY_ORDER_CREATE_COUPON_QUEUE,true);
        }
    
        /**
         * 订单创建 -- 死信队列
         * @return
         */
        @Bean
        public Queue orderCreateDeadLetterQueue(){
            Map<String, Object> arguments = new HashMap<>();
            arguments.put("x-dead-letter-exchange", MQConstant.PAY_ORDER_CREATE_DEAD_LETTER_EXCHANGE);
            arguments.put("x-dead-letter-routing-key", MQConstant.PAY_ORDER_CREATE_EXPIRE_ROUTING_KEY);
            arguments.put("x-message-ttl", 6000);
            // 是否持久化
            boolean durable = true;
            // 仅创建者可以使用的私有队列，断开后自动删除
            boolean exclusive = false;
            // 当所有消费客户端连接断开后，是否自动删除队列
            boolean autoDelete = false;
            return new Queue(MQConstant.PAY_ORDER_CREATE_DEAD_LETTER_QUEUE,durable, exclusive, autoDelete, arguments);
        }
    
        /**
         * 订单创建 -- 过期消费队列
         * @return
         */
        @Bean
        public Queue orderCreateExpireQueue() {
            Queue queue = new Queue(MQConstant.PAY_ORDER_CREATE_EXPIRE_QUEUE,true,false,false);
            return queue;
        }
    
        /**
         * {订单创建扇形交换机}绑定{优惠券消费队列}
         * @return
         */
        @Bean
        public Binding orderCreateCouponBinding() {
            return BindingBuilder.bind(orderCreateCouponQueue()).to(orderCreateFanoutExchange());
        }
    
        /**
         * {订单创建扇形交换机}绑定{死信队列}
         * @return
         */
        @Bean
        public Binding orderCreateDeadLetterBinding() {
            return BindingBuilder.bind(orderCreateDeadLetterQueue()).to(orderCreateFanoutExchange());
        }
    
        /**
         * {死信队列}绑定{订单创建死信交换机}
         * @return
         */
        @Bean
        public Binding orderCreateDeadLetterExpireBinding() {
            return BindingBuilder.bind(orderCreateDeadLetterQueue()).to(orderCreateDeadLetterExchange()).withQueueName();
        }
    
        /**
         * {订单创建死信交换机}通过{死信队列中的路由键}与{过期消费队列}绑定起来
         * @return
         */
        @Bean
        public Binding orderCreateExpireBinding() {
            return BindingBuilder.bind(orderCreateExpireQueue()).to(orderCreateDeadLetterExchange()).with(MQConstant.PAY_ORDER_CREATE_EXPIRE_ROUTING_KEY);
        }
    
    }

这里配置的是 6 秒，如果订单 6 秒没有支付就会触发订单关闭动作。

### 3.5 发送订单创建类 ###

这里通过 http 请求模拟订单创建成功，并发送信息到 RabbitMQ 服务器。

@Data
    public class Order {
    
        private String id;
        private String name;
    
    }
    
    @RestController
    public class MessageController {
    
        @Resource
        private RabbitTemplate rabbitTemplate;
    
        @RequestMapping("send")
        public String send(@RequestBody Order order){
            rabbitTemplate.convertAndSend(MQConstant.PAY_ORDER_CREATE_FANOUT_EXCHANGE, "", order);
            return "OK";
        }
    
    }

### 3. 6 订单创建监听器 ###

下面就是关注订单创建的两个监听器，一个模拟优惠券发送，一个模拟订单关闭。他们的逻辑是格式化打印当前时间。

@Component
    public class OrderCreateCouponListener {
    
        @RabbitListener(queues = MQConstant.PAY_ORDER_CREATE_COUPON_QUEUE)
        public void process(Order message) {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String time = sdf.format(new Date());
            System.out.println("订单创建发送优惠券：" + time + " 收到消息," + message);
        }
    
    }
    
    @Component
    public class OrderCreateExpireListener {
    
        @RabbitListener(queues = MQConstant.PAY_ORDER_CREATE_EXPIRE_QUEUE)
        public void process(Order message) {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String time = sdf.format(new Date());
            System.out.println("订单创建订单过期：" + time + " 收到消息," + message);
        }
    
    }

## 4、测试 ##

我们使用 postman 发送一条 http 请求。信息为订单 ID 是 `11`，订单的商品名称是`iphone 6s`

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTA3MzM_size_16_color_FFFFFF_t_70 1]  
接着我们可以在控制台打死，优惠券监听器获取到的时间与订单创建过期消费的监听器相差 6 秒。这个是符合我们预期的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTA3MzM_size_16_color_FFFFFF_t_70 2]

参考文章：

*  https://blog.csdn.net/skiof007/article/details/80914318

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTA3MzM_size_16_color_FFFFFF_t_70]: /images/20220316/986d335ca9844566a14bc5853620aa7b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTA3MzM_size_16_color_FFFFFF_t_70 1]: /images/20220316/5d720172915b4899a9a138a1d13b860a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTA3MzM_size_16_color_FFFFFF_t_70 2]: /images/20220316/38b4330221e840598181595a39651caf.png