【延时任务处理、订单失效】RabbitMQ死信队列实现

【订单失效】RabbitMQ死信队列实现

之前做商城遇到一个关于订单未支付超时失效的问题，总结一下

1.订单失效问题

订单失效问题比较麻烦的地方就是如何能够实时获取失效的订单。

对于这种问题一般有两种解决方案：定时任务处理，延时任务处理

2.定时任务处理

用户下订单后先生成订单信息，然后将该订单加入到定时任务中（30分钟后执行），当到达指定时间后检查订单状态，如果未支付则标识该订单失效。
定时去轮询数据库/缓存，看订单的状态。这种方式的问题很明显，当集群部署服务器的时候需要做分布式锁进行协调，而且实时性不高，对数据库会产生压力

3.延时任务处理

当用户下订单后，将用户的订单的标识全部发送到延时队列中，30分钟后进去消费队列中被消费，消费时先检查该订单的状态，如果未支付则标识该订单失效。

有以下几种延时任务处理方式

~~Java自带的DelayedQuene队列~~

这是java本身提供的一种延时队列，如果项目业务复杂性不高可以考虑这种方式。它是使用jvm内存来实现的，停机会丢失数据，扩展性不强

~~使用redis监听key的过期来实现~~

当用户下订单后把订单信息设置为redis的key，30分钟失效，程序编写监听redis的key失效，然后处理订单（我也尝试过这种方式）。这种方式最大的弊端就是只能监听一台redis的key失效，集群下将无法实现，也有人监听集群下的每个redis节点的key，但我认为这样做很不合适。如果项目业务复杂性不高，redis单机部署，就可以考虑这种方式

RabbitMQ死信队列实现

重点介绍这种方式

4.RabbitMQ死信队列实现监听订单失效

AMQP协议和RabbitMQ队列本身没有直接支持延迟队列功能，但是可以通过以下特性模拟出延迟队列的功能。

【延时任务处理、订单失效】RabbitMQ死信队列实现

Time To Live(TTL)

RabbitMQ可以针对Queue设置x-expires 或者针对Message设置 x-message-ttl，来控制消息的生存时间，如果超时(两者同时设置以最先到期的时间为准)，则消息变为dead letter(死信)

Dead Letter Exchanges（DLX）

下面来做一个例子来实现订单失效，为了效果明显，我们把订单的失效时间设置为10秒（java实现）

5.具体实现

5.1.环境/版本一览：

开发工具：Intellij IDEA 2020.2.3
springboot：2.4.1
jdk：1.8.0_211
maven: 3.6.3
RabbitMQ：3.8.9

docker安装rabbitmq可以查看我这一篇博客 docker 安装rabbitMQ ：https://laoniu.blog.csdn.net/article/details/

项目结构

【延时任务处理、订单失效】RabbitMQ死信队列实现

5.2.项目搭建

最下方有完整代码

新建项目，配置pom.xml加入依赖

 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency>

配置RabbitMQ的队列和路由信息创建类 RabbitMQConfiuration

package com.niu.springbootrabbitmqdelayqueue.config; import org.springframework.amqp.core.*; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.HashMap; import java.util.Map; / * @description: rabbitMQ配置信息 * @author: nxq email:  * @createDate: 2020/12/18 8:09 上午 * @updateUser: nxq email:  * @updateDate: 2020/12/18 8:09 上午 * @updateRemark: * @version: 1.0 / @Configuration public class RabbitMQConfiguration { //队列名称 public final static String orderQueue = "order_queue"; //交换机名称 public final static String orderExchange = "order_exchange"; // routingKey public final static String routingKeyOrder = "routing_key_order"; //死信消息队列名称 public final static String dealQueueOrder = "deal_queue_order"; //死信交换机名称 public final static String dealExchangeOrder = "deal_exchange_order"; //死信 routingKey public final static String deadRoutingKeyOrder = "dead_routing_key_order"; //死信队列 交换机标识符 public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange"; //死信队列交换机绑定键标识符 public static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key"; @Autowired private CachingConnectionFactory connectionFactory; @Bean public Queue orderQueue() { // 将普通队列绑定到死信队列交换机上 Map<String, Object> args = new HashMap<>(2); //args.put("x-message-ttl", 5 * 1000);//直接设置 Queue 延迟时间 但如果直接给队列设置过期时间,这种做法不是很灵活 //这里采用发送消息动态设置延迟时间,这样我们可以灵活控制 args.put(DEAD_LETTER_QUEUE_KEY, dealExchangeOrder); args.put(DEAD_LETTER_ROUTING_KEY, deadRoutingKeyOrder); return new Queue(RabbitMQConfiguration.orderQueue, true, false, false, args); } //声明一个direct类型的交换机 @Bean DirectExchange orderExchange() { return new DirectExchange(RabbitMQConfiguration.orderExchange); } //绑定Queue队列到交换机,并且指定routingKey @Bean Binding bindingDirectExchangeDemo5( ) { return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(routingKeyOrder); } //创建配置死信队列 @Bean public Queue deadQueueOrder() { Queue queue = new Queue(dealQueueOrder, true); return queue; } //创建死信交换机 @Bean public DirectExchange deadExchangeOrder() { return new DirectExchange(dealExchangeOrder); } //死信队列与死信交换机绑定 @Bean public Binding bindingDeadExchange() { return BindingBuilder.bind(deadQueueOrder()).to(deadExchangeOrder()).with(deadRoutingKeyOrder); } }

配置消息生产者

package com.niu.springbootrabbitmqdelayqueue.controller; import com.niu.springbootrabbitmqdelayqueue.config.RabbitMQConfiguration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import java.util.UUID; / * @description: 订单的控制器 * @author: nxq email:  * @createDate: 2020/12/18 8:20 上午 * @updateUser: nxq email:  * @updateDate: 2020/12/18 8:20 上午 * @updateRemark: * @version: 1.0 / @RestController @RequestMapping("/order") public class OrderController { private static final Logger logger = LoggerFactory.getLogger(OrderController.class); @Autowired private AmqpTemplate rabbitTemplate; / * 模拟提交订单 * @author nxq * @return java.lang.Object */ @GetMapping("") public Object submit(){ String orderId = UUID.randomUUID().toString(); logger.info("submit order {}", orderId); this.rabbitTemplate.convertAndSend( RabbitMQConfiguration.orderExchange, //发送至订单交换机 RabbitMQConfiguration.routingKeyOrder, //订单定routingKey orderId //订单号 这里可以传对象 比如直接传订单对象 , message -> { // 如果配置了 params.put("x-message-ttl", 5 * 1000); // 那么这一句也可以省略,具体根据业务需要是声明 Queue 的时候就指定好延迟时间还是在发送自己控制时间 message.getMessageProperties().setExpiration(1000 * 10 + ""); return message; }); return "{'orderId':'"+orderId+"'}"; } }

配置消费者

package com.niu.springbootrabbitmqdelayqueue.listener; import com.niu.springbootrabbitmqdelayqueue.config.RabbitMQConfiguration; import com.niu.springbootrabbitmqdelayqueue.controller.OrderController; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.messaging.handler.annotation.Headers; import org.springframework.stereotype.Component; import java.io.IOException; import java.util.Date; import java.util.Map; import com.rabbitmq.client.Channel; / * @description: 订单失效监听器 * @author: nxq email:  * @createDate: 2020/12/18 8:30 上午 * @updateUser: nxq email:  * @updateDate: 2020/12/18 8:30 上午 * @updateRemark: * @version: 1.0 / @Component public class OrderFailureListener { private static final Logger logger = LoggerFactory.getLogger(OrderFailureListener.class); @RabbitListener( queues = RabbitMQConfiguration.dealQueueOrder //设置订单失效的队列 ) public void process(String order, Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException { logger.info("【订单号】 - [{}]", order); // 判断订单是否已经支付，如果支付则；否则，取消订单（逻辑代码省略) // 手动ack // Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG); // 手动签收 // channel.basicAck(deliveryTag, false); System.out.println("执行结束...."); } }

配置文件 application.yml 加入rabbitmq的相关配置

spring: rabbitmq: host: localhost port: 5672 username: admin password: admin

配置启动类

package com.niu.springbootrabbitmqdelayqueue; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class SpringbootRabbitmqDelayQueueApplication { public static void main(String[] args) { SpringApplication.run(SpringbootRabbitmqDelayQueueApplication.class, args); } }