4.延迟消息

在电商的支付业务中，对于一些库存有限的商品，为了更好的用户体验，通常都会在用户下单时立刻扣减商品库存。例如电影院购票、高铁购票，下单后就会锁定座位资源，其他人无法重复购买。

但是这样就存在一个问题，假如用户下单后一直不付款，就会一直占有库存资源，导致其他客户无法正常交易，最终导致商户利益受损！

因此，电商中通常的做法就是：对于超过一定时间未支付的订单，应该立刻取消订单并释放占用的库存。

例如，订单支付超时时间为30分钟，则我们应该在用户下单后的第30分钟检查订单支付状态，如果发现未支付，应该立刻取消订单，释放库存。

但问题来了：如何才能准确的实现在下单后第30分钟去检查支付状态呢？

像这种在一段时间以后才执行的任务，我们称之为延迟任务，而要实现延迟任务，最简单的方案就是利用MQ的延迟消息了。

在RabbitMQ中实现延迟消息也有两种方案：

• 死信交换机+TTL

• 延迟消息插件

这一章我们就一起研究下这两种方案的实现方式，以及优缺点。

4.1.死信交换机和延迟消息

首先我们来学习一下基于死信交换机的延迟消息方案。

4.1.1.死信交换机

什么是死信？

当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

• 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false

• 消息是一个过期消息，超时无人消费

• 要投递的队列消息满了，无法投递

如果一个队列中的消息已经成为死信，并且这个队列通过**dead-letter-exchange**属性指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机就称为死信交换机（Dead Letter Exchange）。而此时加入有队列与死信交换机绑定，则最终死信就会被投递到这个队列中。

死信交换机有什么作用呢？

1. 收集那些因处理失败而被拒绝的消息

2. 收集那些因队列满了而被拒绝的消息

3. 收集因TTL（有效期）到期的消息

4.1.2.延迟消息

前面两种作用场景可以看做是把死信交换机当做一种消息处理的最终兜底方案，与消费者重试时讲的RepublishMessageRecoverer作用类似。

而最后一种场景，大家设想一下这样的场景： 如图，有一组绑定的交换机（ttl.fanout）和队列（ttl.queue）。但是ttl.queue没有消费者监听，而是设定了死信交换机hmall.direct，而队列direct.queue1则与死信交换机绑定，RoutingKey是blue：

假如我们现在发送一条消息到ttl.fanout，RoutingKey为blue，并设置消息的有效期为5000毫秒：

warning

注意：尽管这里的ttl.fanout不需要RoutingKey，但是当消息变为死信并投递到死信交换机时，会沿用之前的RoutingKey，这样hmall.direct才能正确路由消息。

消息肯定会被投递到ttl.queue之后，由于没有消费者，因此消息无人消费。5秒之后，消息的有效期到期，成为死信：

死信被再次投递到死信交换机hmall.direct，并沿用之前的RoutingKey，也就是blue：

由于direct.queue1与hmall.direct绑定的key是blue，因此最终消息被成功路由到direct.queue1，如果此时有消费者与direct.queue1绑定， 也就能成功消费消息了。但此时已经是5秒钟以后了：

也就是说，publisher发送了一条消息，但最终consumer在5秒后才收到消息。我们成功实现了延迟消息。

4.1.3.总结

注意：

RabbitMQ的消息过期是基于追溯方式来实现的，也就是说当一个消息的TTL到期以后不一定会被移除或投递到死信交换机，而是在消息恰好处于队首时才会被处理。 当队列中消息堆积很多的时候，过期消息可能不会被按时处理，因此你设置的TTL时间不一定准确。

4.2.DelayExchange插件

基于死信队列虽然可以实现延迟消息，但是太麻烦了。因此RabbitMQ社区提供了一个延迟消息插件来实现相同的效果。 官方文档说明：

Scheduling Messages with RabbitMQ | RabbitMQ - Blog

4.2.1.下载

插件下载地址：

GitHub - rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ

4.2.2.安装

因为我们是基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。

 docker volume inspect mq-plugins

结果如下：

 [
     {
         "CreatedAt": "2024-06-19T09:22:59+08:00",
         "Driver": "local",
         "Labels": null,
         "Mountpoint": "/var/lib/docker/volumes/mq-plugins/_data",
         "Name": "mq-plugins",
         "Options": null,
         "Scope": "local"
     }
 ]
 ​

插件目录被挂载到了/var/lib/docker/volumes/mq-plugins/_data这个目录，我们上传插件到该目录下。

接下来执行命令，安装插件：

 docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

运行结果如下：

4.2.3.声明延迟交换机

基于注解方式：

 @RabbitListener(bindings = @QueueBinding(
         value = @Queue(name = "delay.queue", durable = "true"),
         exchange = @Exchange(name = "delay.direct", delayed = "true"),
         key = "delay"
 ))
 public void listenDelayMessage(String msg){
     log.info("接收到delay.queue的延迟消息：{}", msg);
 }

基于@Bean的方式：

 package com.itheima.consumer.config;
 ​
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.amqp.core.*;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 ​
 @Slf4j
 @Configuration
 public class DelayExchangeConfig {
 ​
     @Bean
     public DirectExchange delayExchange(){
         return ExchangeBuilder
                 .directExchange("delay.direct") // 指定交换机类型和名称
                 .delayed() // 设置delay的属性为true
                 .durable(true) // 持久化
                 .build();
     }
 ​
     @Bean
     public Queue delayedQueue(){
         return new Queue("delay.queue");
     }
     
     @Bean
     public Binding delayQueueBinding(){
         return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");
     }
 }
 ​

4.2.4.发送延迟消息

发送消息时，必须通过x-delay属性设定延迟时间：

 @Test
 void testPublisherDelayMessage() {
     // 1.创建消息
     String message = "hello, delayed message";
     // 2.发送消息，利用消息后置处理器添加消息头
     rabbitTemplate.convertAndSend("delay.direct", "delay", message, new MessagePostProcessor() {
         @Override
         public Message postProcessMessage(Message message) throws AmqpException {
             // 添加延迟消息属性
             message.getMessageProperties().setDelay(5000);
             return message;
         }
     });
 }

:::warning 注意： 延迟消息插件内部会维护一个本地数据库表，同时使用Elang Timers功能实现计时。如果消息的延迟时间设置较长，可能会导致堆积的延迟消息非常多，会带来较大的CPU开销，同时延迟消息的时间会存在误差。 因此，不建议设置延迟时间过长的延迟消息。 :::

4.5.订单状态同步问题

接下来，我们就在交易服务中利用延迟消息实现订单支付状态的同步。其大概思路如下：

假如订单超时支付时间为30分钟，理论上说我们应该在下单时发送一条延迟消息，延迟时间为30分钟。这样就可以在接收到消息时检验订单支付状态，关闭未支付订单。 但是大多数情况下用户支付都会在1分钟内完成，我们发送的消息却要在MQ中停留30分钟，额外消耗了MQ的资源。因此，我们最好多检测几次订单支付状态，而不是在最后第30分钟才检测。 例如：我们在用户下单后的第10秒、20秒、30秒、45秒、60秒、1分30秒、2分、...30分分别设置延迟消息，如果提前发现订单已经支付，则后续的检测取消即可。 这样就可以有效避免对MQ资源的浪费了。

优化后的实现思路如下：

由于我们要多次发送延迟消息，因此需要先定义一个记录消息延迟时间的消息体，处于通用性考虑，我们将其定义到hm-common模块下，代码如下：

 package com.hmall.common.domain;
 ​
 import com.hmall.common.utils.CollUtils;
 import lombok.Data;
 ​
 import java.util.List;
 ​
 @Data
 public class MultiDelayMessage<T> {
     /**
      * 消息体
      */
     private T data;
     /**
      * 记录延迟时间的集合
      */
     private List<Long> delayMillis;
 ​
     public MultiDelayMessage(T data, List<Long> delayMillis) {
         this.data = data;
         this.delayMillis = delayMillis;
     }
     public static <T> MultiDelayMessage<T> of(T data, Long ... delayMillis){
         return new MultiDelayMessage<>(data, CollUtils.newArrayList(delayMillis));
     }
 ​
     /**
      * 获取并移除下一个延迟时间
      * @return 队列中的第一个延迟时间
      */
     public Long removeNextDelay(){
         return delayMillis.remove(0);
     }
 ​
     /**
      * 是否还有下一个延迟时间
      */
     public boolean hasNextDelay(){
         return !delayMillis.isEmpty();
     }
 }
 ​

4.5.1.定义常量

无论是消息发送还是接收都是在交易服务完成，因此我们在trade-service中定义一个常量类，用于记录交换机、队列、RoutingKey等常量，内容如下：

 package com.hmall.trade.constants;
 ​
 public interface MqConstants {
     String DELAY_EXCHANGE = "trade.delay.topic";
     String DELAY_ORDER_QUEUE = "trade.order.delay.queue";
     String DELAY_ORDER_ROUTING_KEY = "order.query";
 }

4.5.2.抽取共享mq配置

我们将mq的配置抽取到nacos中，方便各个微服务共享配置。 在nacos中定义一个名为shared-mq.xml的配置文件，内容如下：

 spring:
   rabbitmq:
     host: ${hm.mq.host:192.168.150.101} # 主机名
     port: ${hm.mq.port:5672} # 端口
     virtual-host: ${hm.mq.vhost:/hmall} # 虚拟主机
     username: ${hm.mq.un:hmall} # 用户名
     password: ${hm.mq.pw:123} # 密码
     listener:
       simple:
         prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

这里只添加一些基础配置，至于生产者确认，消费者确认配置则由微服务根据业务自己决定。

在trade-service模块添加共享配置：

4.5.3.改造下单业务

接下来，我们改造下单业务，在下单完成后，发送延迟消息，查询支付状态。

1）引入依赖 在trade-service模块的pom.xml中引入amqp的依赖：

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

2）改造下单业务 修改trade-service模块的com.hmall.trade.service.impl.OrderServiceImpl类的createOrder方法，添加消息发送的代码：

4.5.4.编写查询支付状态接口

由于MQ消息处理时需要查询支付状态，因此我们要在pay-service模块定义一个这样的接口，并提供对应的FeignClient. 首先，在hm-api模块定义三个类：

说明：

• PayOrderDTO：支付单的数据传输实体

• PayClient：支付系统的Feign客户端

• PayClientFallback：支付系统的fallback逻辑

PayClient代码如下：

 package com.hmall.api.client;
 ​
 import com.hmall.api.client.fallback.PayClientFallback;
 import com.hmall.api.dto.PayOrderDTO;
 import org.springframework.cloud.openfeign.FeignClient;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
 ​
 @FeignClient(value = "pay-service", fallbackFactory = PayClientFallback.class)
 public interface PayClient {
     /**
      * 根据交易订单id查询支付单
      * @param id 业务订单id
      * @return 支付单信息
      */
     @GetMapping("/pay-orders/biz/{id}")
     PayOrderDTO queryPayOrderByBizOrderNo(@PathVariable("id") Long id);
 }

PayClientFallback代码如下：

 package com.hmall.api.client.fallback;
 ​
 import com.hmall.api.client.PayClient;
 import com.hmall.api.dto.PayOrderDTO;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.cloud.openfeign.FallbackFactory;
 ​
 @Slf4j
 public class PayClientFallback implements FallbackFactory<PayClient> {
     @Override
     public PayClient create(Throwable cause) {
         return new PayClient() {
             @Override
             public PayOrderDTO queryPayOrderByBizOrderNo(Long id) {
                 return null;
             }
         };
     }
 }
 ​

最后，在pay-service模块的PayController中实现该接口：

 @ApiOperation("根据id查询支付单")
 @GetMapping("/biz/{id}")
 public PayOrderDTO queryPayOrderByBizOrderNo(@PathVariable("id") Long id){
     PayOrder payOrder = payOrderService.lambdaQuery().eq(PayOrder::getBizOrderNo, id).one();
     return BeanUtils.copyBean(payOrder, PayOrderDTO.class);
 }

4.5.5.消息监听

接下来，我们在trader-service编写一个监听器，监听延迟消息，查询订单支付状态： 代码如下：

​ @Slf4j
 @Component
 @RequiredArgsConstructor
 public class OrderStatusListener {
 ​
     private final IOrderService orderService;
 ​
     private final PayClient payClient;
 ​
     private final RabbitTemplate rabbitTemplate;
 ​
     @RabbitListener(bindings = @QueueBinding(
             value = @Queue(name = MqConstants.DELAY_ORDER_QUEUE, durable = "true"),
             exchange = @Exchange(name = MqConstants.DELAY_EXCHANGE, type = ExchangeTypes.TOPIC),
             key = MqConstants.DELAY_ORDER_ROUTING_KEY
     ))
     public void listenOrderCheckDelayMessage(MultiDelayMessage<Long> msg) {
         // 1.获取消息中的订单id
         Long orderId = msg.getData();
         // 2.查询订单，判断状态：1是未支付，大于1则是已支付或已关闭
         Order order = orderService.getById(orderId);
         if (order == null || order.getStatus() > 1) {
             // 订单不存在或交易已经结束，放弃处理
             return;
         }
         // 3.可能是未支付，查询支付服务
         PayOrderDTO payOrder = payClient.queryPayOrderByBizOrderNo(orderId);
         if (payOrder != null && payOrder.getStatus() == 3) {
             // 支付成功，更新订单状态
             orderService.markOrderPaySuccess(orderId);
             return;
         }
         // 4.确定未支付，判断是否还有剩余延迟时间
         if (msg.hasNextDelay()) {
             // 4.1.有延迟时间，需要重发延迟消息，先获取延迟时间的int值
             int delayVal = msg.removeNextDelay().intValue();
             // 4.2.发送延迟消息
             rabbitTemplate.convertAndSend(MqConstants.DELAY_EXCHANGE, MqConstants.DELAY_ORDER_ROUTING_KEY, msg,
                     message -> {
                         message.getMessageProperties().setDelay(delayVal);
                         return message;
                     });
             return;
         }
         // 5.没有剩余延迟时间了，说明订单超时未支付，需要取消订单
         orderService.cancelOrder(orderId);
     }
 }
 ​

注意，这里要在OrderServiceImpl中实现cancelOrder方法，留作作业大家自行实现。

5.2.抽取MQ工具

MQ在企业开发中的常见应用我们就学习完毕了，除了收发消息以外，消息可靠性的处理、生产者确认、消费者确认、延迟消息等等编码还是相对比较复杂的。 因此，我们需要将这些常用的操作封装为工具，方便在项目中使用。要求如下：

• 在hm-commom模块下编写发送消息的工具类RabbitMqHelper

• 定义一个自动配置类MqConsumeErrorAutoConfiguration，内容包括：

  • 声明一个交换机，名为error.direct，类型为direct

  • 声明一个队列，名为：微服务名 + error.queue，也就是说要动态获取

  • 将队列与交换机绑定，绑定时的RoutingKey就是微服务名

  • 声明RepublishMessageRecoverer，消费失败消息投递到上述交换机

  • 给配置类添加条件，当spring.rabbitmq.listener.simple.retry.enabled为true时触发

通用配置类MqConfig如下：

 @Configuration
 @ConditionalOnClass(value = {MessageConverter.class, AmqpTemplate.class})
 public class MqConfig implements EnvironmentAware{
 ​
     private String defaultErrorRoutingKey;
     private String defaultErrorQueue;
 ​
     @Bean(name = "rabbitListenerContainerFactory")
     @ConditionalOnProperty(prefix = "spring.rabbitmq.listener", name = "type", havingValue = "simple",
             matchIfMissing = true)
     SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(
             SimpleRabbitListenerContainerFactoryConfigurer configurer,
             ConnectionFactory connectionFactory,
             ObjectProvider<ContainerCustomizer<SimpleMessageListenerContainer>> simpleContainerCustomizer) {
         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
         configurer.configure(factory, connectionFactory);
         simpleContainerCustomizer.ifUnique(factory::setContainerCustomizer);
         factory.setAfterReceivePostProcessors(message -> {
             Object header = message.getMessageProperties().getHeader(REQUEST_ID_HEADER);
             if(header != null) {
                 MDC.put(REQUEST_ID_HEADER, header.toString());
             }
             return message;
         });
         return factory;
     }
 ​
     @Bean
     public MessageConverter messageConverter(ObjectMapper mapper){
         // 1.定义消息转换器
         Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter(mapper);
         // 2.配置自动创建消息id，用于识别不同消息
         jackson2JsonMessageConverter.setCreateMessageIds(true);
         return jackson2JsonMessageConverter;
     }
 ​
     /**
      * <h1>消息处理失败的重试策略</h1>
      * 本地重试失败后，消息投递到专门的失败交换机和失败消息队列：error.queue
      */
     @Bean
     @ConditionalOnClass(MessageRecoverer.class)
     @ConditionalOnMissingBean
     public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
         // 消息处理失败后，发送到错误交换机：error.direct，RoutingKey默认是error.微服务名称
         return new RepublishMessageRecoverer(
                 rabbitTemplate, ERROR_EXCHANGE, defaultErrorRoutingKey);
     }
 ​
     /**
      * rabbitmq发送工具
      *
      */
     @Bean
     @ConditionalOnMissingBean
     @ConditionalOnClass(RabbitTemplate.class)
     public RabbitMqHelper rabbitMqHelper(RabbitTemplate rabbitTemplate){
         return new RabbitMqHelper(rabbitTemplate);
     }
 ​
     /**
      * 专门接收处理失败的消息
      */
     @Bean
     public DirectExchange errorMessageExchange(){
         return new DirectExchange(ERROR_EXCHANGE);
     }
 ​
     @Bean
     public Queue errorQueue(){
         return new Queue(defaultErrorQueue, true);
     }
 ​
     @Bean
     public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
         return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with(defaultErrorRoutingKey);
     }
 ​
     @Override
     public void setEnvironment(Environment environment) {
         String appName = environment.getProperty("spring.application.name");//从配置文件中获取当前微服务的名称
         this.defaultErrorRoutingKey = ERROR_KEY_PREFIX + appName;
         this.defaultErrorQueue = StringUtils.format(ERROR_QUEUE_TEMPLATE, appName);
     }
 }

RabbitMqHelper如下：

 @Slf4j
 public class RabbitMqHelper {
 ​
     private final RabbitTemplate rabbitTemplate;
     private final MessagePostProcessor processor = new BasicIdMessageProcessor();
     private final ThreadPoolTaskExecutor executor;
 ​
     public RabbitMqHelper(RabbitTemplate rabbitTemplate) {
         this.rabbitTemplate = rabbitTemplate;
         executor = new ThreadPoolTaskExecutor();
         //配置核心线程数
         executor.setCorePoolSize(10);
         //配置最大线程数
         executor.setMaxPoolSize(15);
         //配置队列大小
         executor.setQueueCapacity(99999);
         //配置线程池中的线程的名称前缀
         executor.setThreadNamePrefix("mq-async-send-handler");
 ​
         // 设置拒绝策略：当pool已经达到max size的时候，如何处理新任务
         // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
         //执行初始化
         executor.initialize();
     }
 ​
     /**
      * 根据exchange和routingKey发送消息
      */
     public <T> void send(String exchange, String routingKey, T t) {
         log.debug("准备发送消息，exchange：{}， RoutingKey：{}， message：{}", exchange, routingKey, t);
         // 1.设置消息标示，用于消息确认，消息发送失败直接抛出异常，交给调用者处理
         String id = UUID.randomUUID().toString(true);
         CorrelationData correlationData = new CorrelationData(id);
         // 2.设置发送超时时间为500毫秒
         rabbitTemplate.setReplyTimeout(500);
         // 3.发送消息，同时设置消息id
         rabbitTemplate.convertAndSend(exchange, routingKey, t, processor, correlationData);
     }
 ​
     /**
      * 根据exchange和routingKey发送消息，并且可以设置延迟时间
      */
     public <T> void sendDelayMessage(String exchange, String routingKey, T t, Duration delay) {
         // 1.设置消息标示，用于消息确认，消息发送失败直接抛出异常，交给调用者处理
         String id = UUID.randomUUID().toString(true);
         CorrelationData correlationData = new CorrelationData(id);
         // 2.设置发送超时时间为500毫秒
         rabbitTemplate.setReplyTimeout(500);
         // 3.发送消息，同时设置消息id
         rabbitTemplate.convertAndSend(exchange, routingKey, t, new DelayedMessageProcessor(delay), correlationData);
     }
 ​
 ​
     /**
      * 根据exchange和routingKey 异步发送消息，并指定一个延迟时间
      *
      * @param exchange   交换机
      * @param routingKey 路由KEY
      * @param t          数据
      * @param <T>        数据类型
      */
     public <T> void sendAsync(String exchange, String routingKey, T t, Long time) {
         String requestId = MDC.get(REQUEST_ID_HEADER);
         CompletableFuture.runAsync(() -> {
             try {
                 MDC.put(REQUEST_ID_HEADER, requestId);
                 // 发送延迟消息
                 if (time != null && time > 0) {
                     sendDelayMessage(exchange, routingKey, t, Duration.ofMillis(time));
                 } else {
                     send(exchange, routingKey, t);
                 }
             } catch (Exception e) {
                 log.error("推送消息异常，t:{},", t, e);
             }
         }, executor);
     }
 ​
 ​
     /**
      * 根据exchange和routingKey 异步发送消息
      *
      * @param exchange   交换机
      * @param routingKey 路由KEY
      * @param t          数据
      * @param <T>        数据类型
      */
     public <T> void sendAsync(String exchange, String routingKey, T t) {
         sendAsync(exchange, routingKey, t, null);
     }
 ​
 }
 ​