RabbitMQ从0到1完整学习笔记二：《高级篇》-CSDN博客

目录

1. 发送者的可靠性

1.1.生产者连接重试机制

1.2.生产者确认机制发布确认

1.3.实现生产者确认

1.3.1.开启生产者确认

1.3.2.定义ReturnCallback

1.3.3.定义ConfirmCallback

拓展 confirm 模式细节处理

2.MQ的可靠性

2.1.数据持久化

2.1.1.交换机持久化

2.1.2.队列持久化

2.1.3.消息持久化

2.2.LazyQueue

2.2.1.控制台配置Lazy模式

2.2.2.代码配置Lazy模式

2.2.3.更新已有队列为lazy模式

拓展惰性队列vs持久化队列 

拓展队列的销毁注意细节 

3.消费者的可靠性

3.1.消费者确认机制

拓展消息应答策略的选择讨论

3.2.失败重试机制

3.3.失败处理策略

3.4.业务幂等性

3.4.1.唯一消息ID

3.4.2.业务判断

3.5.兜底方案

拓展应答和发布区别

拓展备份交换机 

什么是备份交换机呢

原理步骤

4. 延迟消息

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

4.1.1死信交换机

4.1.2.延迟消息

4.1.3.总结

4.2.DelayExchange插件

4.2.1.下载

4.2.3.声明延迟交换机

4.2.4.发送延迟消息

拓展优先级队列 

使用场景

实现步骤

4.5.订单状态同步问题

4.5.1.定义常量

4.5.2.抽取共享mq配置

4.5.3.改造下单业务

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

4.5.5.消息监听

---

在昨天的练习作业中我们改造了余额支付功能在支付成功后利用RabbitMQ通知交易服务更新业务订单状态为已支付。 但是大家思考一下如果这里MQ通知失败支付服务中支付流水显示支付成功而交易服务中的订单状态却显示未支付数据出现了不一致。 此时前端发送请求查询支付状态时肯定是查询交易服务状态会发现业务订单未支付而用户自己知道已经支付成功这就导致用户体验不一致。

因此这里我们必须尽可能确保MQ消息的可靠性即消息应该至少被消费者处理1次 那么问题来了

• 我们该如何确保MQ消息的可靠性

• 如果真的发送失败有没有其它的兜底方案

这些问题在今天的学习中都会找到答案。

1. 发送者的可靠性

首先我们一起分析一下消息丢失的可能性有哪些。 消息从发送者发送消息到消费者处理消息需要经过的流程是这样的

消息从生产者到消费者的每一步都可能导致消息丢失

• 发送消息时丢失

  • 生产者发送消息时连接MQ失败

  • 生产者发送消息到达MQ后未找到Exchange

  • 生产者发送消息到达MQ的Exchange后未找到合适的Queue

  • 消息到达MQ后处理消息的进程发生异常

• MQ导致消息丢失

  • 消息到达MQ保存到队列后尚未消费就突然宕机

• 消费者处理消息时

  • 消息接收后尚未处理突然宕机

  • 消息接收后处理过程中抛出异常

综上我们要解决消息丢失问题保证MQ的可靠性就必须从3个方面入手

• 确保生产者一定把消息发送到MQ

• 确保MQ不会将消息弄丢

• 确保消费者一定要处理消息

---

 这一章我们先来看如何确保生产者一定能把消息发送到MQ。

1.1.生产者连接重试机制

首先第一种情况就是生产者发送消息时出现了网络故障导致与MQ的连接中断。

为了解决这个问题SpringAMQP提供的消息发送时的重试机制。即

当RabbitTemplate与MQ连接超时后多次重试

修改publisher模块的application.yaml文件添加下面的内容

spring:
  rabbitmq:
    connection-timeout: 1s # 设置MQ的连接超时时间
    template:
      retry:
        enabled: true # 开启超时重试机制
        initial-interval: 1000ms # 失败后的初始等待时间
        multiplier: 1 # 失败后下次的等待时长倍数下次等待时长 = initial-interval * multiplier
        max-attempts: 3 # 最大重试次数

我们利用命令停掉RabbitMQ服务

docker stop mq

然后测试发送一条消息会发现会每隔1秒重试1次总共重试了3次。消息发送的超时重试机制配置成功了

注意当网络不稳定的时候利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试也就是说多次重试等待的过程中当前线程是被阻塞的。 如果对于业务性能有要求建议禁用重试机制。如果一定要使用请合理配置等待时长和重试次数当然也可以考虑使用异步线程来执行发送消息的代码。 

---

1.2.生产者确认机制发布确认

一般情况下只要生产者与MQ之间的网路连接顺畅基本不会出现发送消息丢失的情况因此大多数情况下我们无需考虑这种问题。 不过在少数情况下也会出现消息发送到MQ之后丢失的现象比如

• MQ内部处理消息的进程发生了异常

• 生产者发送消息到达MQ后未找到Exchange

• 生产者发送消息到达MQ的Exchange后未找到合适的Queue因此无法路由

针对上述情况RabbitMQ提供了生产者消息确认机制包括Publisher Confirm和Publisher Return两种。在开启确认机制的情况下当生产者发送消息给MQ后MQ会根据消息处理的情况返回不同的回执。 具体如图所示

总结如下

• 当消息投递到MQ但是路由失败时通过Publisher Return返回异常信息同时返回ack的确认信息代表投递成功失败了还会返回ack是因为确认机制只能保证该消息是否到达MQ不管他是否被消费

• 临时消息投递到了MQ并且入队成功返回ACK告知投递成功

• 持久消息投递到了MQ并且入队完成持久化返回ACK 告知投递成功

• 其它情况都会返回NACK告知投递失败

其中ack和nack属于Publisher Confirm机制ack是投递成功nack是投递失败。而return则属于Publisher Return机制。 默认两种机制都是关闭状态需要通过配置文件来开启。

        RabbitMQ的发布确认机制只能确保消息已经成功发布到消息队列但无法直接知道消息是否被消费。发布确认机制只告知发布者消息是否成功到达消息队列而不涉及消息被消费的情况。

        如果需要知道消息是否被消费可结合RabbitMQ的消费者确认机制Consumer Acknowledgements。在消费者从消息队列中接收到消息后可以发送个确认消息给RabbitMQ来表示该消息已被成功消费。这种方式可以通过设置自动确认或手动确认来实现。

---

1.3.实现生产者确认

1.3.1.开启生产者确认

在publisher模块的application.yaml中添加配置

spring:
  rabbitmq:
    publisher-confirm-type: correlated # 开启publisher confirm机制并设置confirm类型
    publisher-returns: true # 开启publisher return机制

这里publisher-confirm-type有三种模式可选

• none关闭confirm机制

• simple同步阻塞等待MQ的回执

• correlatedMQ异步回调返回回执

一般我们推荐使用correlated异步回调机制。

1.3.2.定义ReturnCallback

每个RabbitTemplate只能配置一个ReturnCallback因此我们可以在配置类中统一设置。我们在publisher模块定义一个配置类

内容如下

@Slf4j
@AllArgsConstructor
@Configuration
public class MqConfig {
    private final RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                log.error("触发return callback,");
                log.debug("exchange: {}", returned.getExchange());
                log.debug("routingKey: {}", returned.getRoutingKey());
                log.debug("message: {}", returned.getMessage());
                log.debug("replyCode: {}", returned.getReplyCode());
                log.debug("replyText: {}", returned.getReplyText());
            }
        });
    }
}

public void sendMessage(String exchange, String routingKey, Object message) {
        rabbitTemplate.send(exchange, routingKey, (Message) message);
        // 在发送消息的时候如果消息没有被正确路由到指定的队列将执行ReturnCallback的回调函数
    }

@PostConstruct是Java EE或Spring框架中的一个注解用于指定一个方法在对象实例化后立即执行。通常在使用IoC控制反转容器时我们可以通过@PostConstruct注解来标记一个方法在对象创建完成后执行一些初始化逻辑。

---

1.3.3.定义ConfirmCallback

由于每个消息发送时的处理逻辑不一定相同因此ConfirmCallback需要在每次发消息时定义。具体来说是在调用RabbitTemplate中的convertAndSend方法时多传递一个参数

这里的CorrelationData中包含两个核心的东西

• id消息的唯一标示MQ对不同的消息的回执以此做判断避免混淆

• SettableListenableFuture回执结果的Future对象

将来MQ的回执就会通过这个Future来返回我们可以提前给CorrelationData中的Future添加回调函数来处理消息回执

我们新建一个测试向系统自带的交换机发送消息并且添加ConfirmCallback

@Test
void testPublisherConfirm() {
    // 1.创建CorrelationData
    CorrelationData cd = new CorrelationData();
    // 2.给Future添加ConfirmCallback
    cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
        @Override
        public void onFailure(Throwable ex) {
            // 2.1.Future发生异常时的处理逻辑基本不会触发
            log.error("send message fail", ex);
        }
        @Override
        public void onSuccess(CorrelationData.Confirm result) {
            // 2.2.Future接收到回执的处理逻辑参数中的result就是回执内容
            if(result.isAck()){ // result.isAck()boolean类型true代表ack回执false 代表 nack回执
                log.debug("发送消息成功收到 ack!");
            }else{ // result.getReason()String类型返回nack时的异常描述
                log.error("发送消息失败收到 nack, reason : {}", result.getReason());
            }
        }
    });
    // 3.发送消息
    rabbitTemplate.convertAndSend("hmall.direct", "q", "hello", cd);
}

执行结果如下

可以看到由于传递的RoutingKey是错误的路由失败后触发了return callback同时也收到了ack。 当我们修改为正确的RoutingKey以后就不会触发return callback了只收到ack。 而如果连交换机都是错误的则只会收到nack。

注意 开启生产者确认比较消耗MQ性能一般不建议开启。而且大家思考一下触发确认的几种情况

• 路由失败一般是因为RoutingKey错误导致往往是编程导致

• 交换机名称错误同样是编程错误导致

• MQ内部故障这种需要处理但概率往往较低。因此只有对消息可靠性要求非常高的业务才需要开启而且仅仅需要开启ConfirmCallback处理nack就可以了。

---

拓展 confirm 模式细节处理

        confirm 模式最大的好处在于是异步的一旦发布一条消息生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息当消息最终得到确认之后生产者应用便可以通过回调方法来处理该确认消息如果RabbitMQ 因为自身内部错误导致消息丢失就会发送一条 nack 消息 生产者应用程序同样可以在回调方法中处理该 nack 消息。

细节

监听器要定义在发布消息的步骤前面不然在发消息的过程中就无法捕捉发布确认结果所以说要将其提前打开监听器再发布消息

deliveryTag一条消息的标记号的初始值也是0L,但是标记发送的消息是从1开始的1-65535这样getNextPublishSeqNo()下一条消息的标记号的值和deliveryTag的值就是对应的

---

2.MQ的可靠性

消息到达MQ以后如果MQ不能及时保存也会导致消息丢失所以MQ的可靠性也非常重要。

2.1.数据持久化

为了提升性能默认情况下MQ的数据都是在内存存储的临时数据重启后就会消失。为了保证数据的可靠性必须配置数据持久化包括

• 交换机持久化

• 队列持久化

• 消息持久化

我们以控制台界面为例来说明。

2.1.1.交换机持久化

在控制台的Exchanges页面添加交换机时可以配置交换机的Durability参数

设置为Durable就是持久化模式Transient就是临时模式。

2.1.2.队列持久化

在控制台的Queues页面添加队列时同样可以配置队列的Durability参数

除了持久化以外你可以看到队列还有很多其它参数有一些我们会在后期学习。

RabbitMQ的非持久队列确实是应该在生产者和消费者断开连接后就被销毁的但是如果队列里还存在未被消费的消息那么队列就不会被自动删除直到所有的消息都被消费或备份。另外如果设置了auto_delete参数为false也会导致队列不会自动删除。建议使用命令行或者管理界面查看队列状态以确认队列是否已被删除。

2.1.3.消息持久化

在控制台发送消息的时候可以添加很多参数而消息的持久化是要配置一个properties

说明在开启持久化机制以后如果同时还开启了生产者确认那么MQ会在消息持久化以后才发送ACK回执进一步确保消息的可靠性。 不过出于性能考虑为了减少IO次数发送到MQ的消息并不是逐条持久化到数据库的而是每隔一段时间批量持久化。一般间隔在100毫秒左右这就会导致ACK有一定的延迟因此建议生产者确认全部采用异步方式。

持久化消息会同时写入磁盘和内存加快读取速度非持久化消息会暂存在内存中只有当内存不够用时才将消息写入磁盘但一般重启之后在磁盘中的非持久化消息就没有了。

消息持久化这个方法并不能保证消息100%不丢失如果在当消息刚准备存储在磁盘的时候 但是还没有存储完消息还在缓存的一个间隔点。此时并没有真正写入磁盘。如果一旦服务崩溃宕机消息还是会丢失。

---

2.2.LazyQueue

在默认情况下RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。但在某些特殊情况下这会导致消息积压比如

• 消费者宕机或出现网络故障

• 消息发送量激增超过了消费者处理速度

• 消费者处理业务发生阻塞

一旦出现消息堆积问题RabbitMQ的内存占用就会越来越高直到触发内存预警上限。此时RabbitMQ会将内存消息刷到磁盘上这个行为成为PageOut. PageOut会耗费一段时间并且会阻塞队列进程。因此在这个过程中RabbitMQ不会再处理新的消息生产者的所有请求都会被阻塞。

为了解决这个问题从RabbitMQ的3.6.0版本开始就增加了Lazy Queues的模式也就是惰性队列。惰性队列的特征如下

• 接收到消息后直接存入磁盘而非内存

• 消费者要消费消息时才会从磁盘中读取并加载到内存也就是懒加载

• 支持数百万条的消息存储

而在3.12版本之后LazyQueue已经成为所有队列的默认格式。因此官方推荐升级MQ为3.12版本或者所有队列都设置为LazyQueue模式。

2.2.1.控制台配置Lazy模式

在添加队列的时候添加x-queue-mod=lazy参数即可设置队列为Lazy模式

2.2.2.代码配置Lazy模式

在利用SpringAMQP声明队列的时候添加x-queue-mod=lazy参数也可设置队列为Lazy模式

@Bean
public Queue lazyQueue(){
    return QueueBuilder
            .durable("lazy.queue")
            .lazy() // 开启Lazy模式
            .build();
}

这里是通过QueueBuilder的lazy()函数配置Lazy模式底层源码如下

当然我们也可以基于注解来声明队列并设置为Lazy模式

@RabbitListener(queuesToDeclare = @Queue(
        name = "lazy.queue",
        durable = "true",
        arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg){
    log.info("接收到 lazy.queue的消息{}", msg);
}

---

2.2.3.更新已有队列为lazy模式

对于已经存在的队列也可以配置为lazy模式但是要通过设置policy实现。 可以基于命令行设置policy

rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues  

命令解读

• rabbitmqctl RabbitMQ的命令行工具

• set_policy 添加一个策略

• Lazy 策略名称可以自定义

• "^lazy-queue$" 用正则表达式匹配队列的名字

• '{"queue-mode":"lazy"}' 设置队列模式为lazy模式

• --apply-to queues策略的作用对象是所有的队列

当然也可以在控制台配置policy进入在控制台的Admin页面点击Policies即可添加配置

---

拓展惰性队列vs持久化队列 

        RabbitMQ的惰性队列是指在队列中添加消息时会将其暂存到磁盘上而不是保存在内存中。相反持久化队列是指消息可以被保存在队列中并在RabbitMQ服务重启时保留。

        惰性队列的主要优点在于它可以避免内存溢出和性能问题。当许多消息被添加到队列中时若使用持久化队列会占用大量的内存。但使用惰性队列时消息在必要时被加载到 RAM 中因而保证了高效的内存利用率。惰性队列适用于需要处理大量消息的极端情况并且长时间运行数据量增加的情况下更适合使用。

        持久化队列主要适用于需要确保即使在RabbitMQ宕机或重启的情况下仍能找回的消息。持久化队列将队列中的消息保存到磁盘上因此可以确保消息不会丢失。持久化队列适用于数据量小特别是小型应用场景下使用。

总之对于性能要求高的大型应用程序可以考虑使用惰性队列而对于小型应用程序特别是需要保证消息的完整性和不可丢失性的应用程序持久化队列更加适用。

---

拓展队列的销毁注意细节 

1. 当队列被显式地删除时队列将被销毁。

2. 当最后一个绑定队列的消费者关闭其通道/连接时队列将被删除。 如果显式地将queue.unused-queue-ttl参数设置为非零值则队列可能会在未使用一段时间后被自动删除。

3. 当RabbitMQ服务器重启时非持久队列会被自动删除。

4. 如果将x-expires参数设置为非零值则队列可能在未使用一段时间后被自动删除。

5. 如果队列被指定为具有自动删除属性则队列将在完全消耗或关闭时自动删除。

请注意如果队列中未设置消息/元素计数则即使队列未在消费者中使用过也不会被自动删除。

RabbitMQ的非持久化队列在生产者和消费者断开连接后会被销毁。这是因为非持久化队列的生存期仅限于存在连接的时候。一旦所有消费者和生产者都断开连接队列就会被删除。如果您需要在RabbitMQ断开连接后仍要使用队列请创建一个持久化队列。持久化队列会将队列的元数据和消息保存在磁盘上即使RabbitMQ重启或断电也能保证数据不会丢失

---

3.消费者的可靠性

当RabbitMQ向消费者投递消息以后需要知道消费者的处理状态如何。因为消息投递给消费者并不代表就一定被正确消费了可能出现的故障有很多比如

• 消息投递的过程中出现了网络故障

• 消费者接收到消息后突然宕机

• 消费者接收到消息后因处理不当导致异常

• ...

一旦发生上述情况消息也会丢失。因此RabbitMQ必须知道消费者的处理状态一旦消息处理失败才能重新投递消息。 但问题来了RabbitMQ如何得知消费者的处理状态呢

---

3.1.消费者确认机制

为了确认消费者是否成功处理消息RabbitMQ提供了消费者确认机制Consumer Acknowledgement。即当消费者处理消息结束后应该向RabbitMQ发送一个回执告知RabbitMQ自己消息处理状态。回执有三种可选值

• ack成功处理消息RabbitMQ从队列中删除该消息

• nack消息处理失败RabbitMQ需要再次投递消息

• reject消息处理失败并拒绝该消息RabbitMQ从队列中删除该消息

一般reject方式用的较少除非是消息格式有问题那就是开发问题了。因此大多数情况下我们需要将消息处理的代码通过try catch机制捕获消息处理成功时返回ack处理失败时返回nack.

由于消息回执的处理代码比较统一因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式有三种模式

• none不处理。即消息投递给消费者后立刻ack消息会立刻从MQ删除。非常不安全不建议使用

• manual手动模式。需要自己在业务代码中调用api发送ack或reject存在业务入侵但更灵活

• auto自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强当业务正常执行时则自动返回ack. 当业务出现异常时根据异常判断返回不同结果

  • 如果是业务异常会自动返回nack

  • 如果是消息处理或校验异常自动返回reject;

通过下面的配置可以修改SpringAMQP的ACK处理方式

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: none # 不做处理

修改consumer服务的SpringRabbitListener类中的方法模拟一个消息处理的异常

@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {
    log.info("spring 消费者接收到消息【" + msg + "】");
    if (true) {
        throw new MessageConversionException("故意的");
    }
    log.info("消息处理完成");
}

测试可以发现当消息处理发生异常时消息依然被RabbitMQ删除了。

我们再次把确认机制修改为auto

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto # 自动ack

在异常位置打断点再次发送消息程序卡在断点时可以发现此时消息状态为unacked未确定状态

放行以后由于抛出的是消息转换异常因此Spring会自动返回reject所以消息依然会被删除

我们将异常改为RuntimeException类型

@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {
    log.info("spring 消费者接收到消息【" + msg + "】");
    if (true) {
        throw new RuntimeException("故意的");
    }
    log.info("消息处理完成");
}

在异常位置打断点然后再次发送消息测试程序卡在断点时可以发现此时消息状态为unacked未确定状态

放行以后由于抛出的是业务异常所以Spring返回ack最终消息恢复至Ready状态并且没有被RabbitMQ删除

当我们把配置改为auto时消息处理失败后会回到RabbitMQ并重新投递到消费者。

---

拓展消息应答策略的选择讨论

自动应答手动应答的好处是可以批量应答并且减少网络拥堵

        消息发送后立即被认为已经传送成功这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前消费者那边出现连接或者 channel 关闭那么消息就丢失了,当然另一方面这种模式消费者那边可以传递过载的消息没有对传递的消息数量进行限制当然这样有可能使得消费者这边由于接收太多还来不及处理的消息导致这些消息的积压最终使得内存耗尽最终这些消费者线程被操作系统杀死所以这种模式仅适用在消费者可以高效并以 某种速率能够处理这些消息的情况下使用。自动应答没有什么批量的概念因为接收到就立马应答了哪里会等到后边再一起应答

        设置了自动应答后消费者在接收到消息后就会立马自动应答然后rabbitMQ就会立马将该消息从队列中删除但是这是不安全的要知道我们接收消息的目的还是使用消息如果说在接收到消息后在时候消息的时候由于报错等一系列因素导致的数据丢失了那么我们就没有后悔的余地了因为队列中的消息已经被删除。所以说我们最好采用手动应答的方式在接收到消息并处理了消息的流程的最后再进行手动应答进行消息在队列中删除的过程这样子哪怕在处理消息的过程中丢失了消息都能在队列中再拿一份出来

---

3.2.失败重试机制

        当消费者出现异常后消息会不断requeue重入队到队列再重新发送给消费者。如果消费者再次执行依然出错消息会再次requeue到队列再次投递直到消息处理成功为止。 极端情况就是消费者一直无法执行成功那么消息requeue就会无限循环导致mq的消息处理飙升带来不必要的压力

        当然上述极端情况发生的概率还是非常低的不过不怕一万就怕万一。为了应对上述情况Spring又提供了消费者失败重试机制

在消费者出现异常时利用本地重试而不是无限制的requeue到mq队列。

修改consumer服务的application.yml文件添加内容

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 1000ms # 初识的失败等待时长为1秒
          multiplier: 1 # 失败的等待时长倍数下次等待时长 = multiplier * last-interval
          max-attempts: 3 # 最大重试次数
          stateless: true # true无状态false有状态。如果业务中包含事务这里改为false

stateless 配置是 RabbitMQ 中一个重要的属性用于定义消息的传递方式。当将队列和交换机配置为无状态时它们不会在磁盘上持久化消息。这意味着如果 RabbitMQ 服务被重启或崩溃所有未处理的消息都会丢失。Stateless 配置适用于那些对消息持久性要求不高、可丢失的场景例如临时数据的传递或不需要保留的日志信息等。stateless 为false便是不加持久化这也是为什么说如果包含了事务就不需要考虑用持久化来确保数据的完整性

重启consumer服务重复之前的测试。可以发现

• 消费者在失败后消息没有重新回到MQ无限重新投递而是在本地重试了3次

• 本地重试3次以后抛出了AmqpRejectAndDontRequeueException异常。查看RabbitMQ控制台发现消息被删除了说明最后SpringAMQP返回的是reject

结论

• 开启本地重试时消息处理过程中抛出异常不会requeue到队列而是在消费者本地重试

• 重试达到最大次数后Spring会返回reject消息会被丢弃 

---

3.3.失败处理策略

在之前的测试中本地测试达到最大重试次数后消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下显然不太合适了。 因此Spring允许我们自定义重试次数耗尽后的消息处理策略这个策略是由MessageRecovery接口来定义的它有3个不同实现策略模式

• RejectAndDontRequeueRecoverer重试耗尽后直接reject丢弃消息。默认就是这种方式

• ImmediateRequeueMessageRecoverer重试耗尽后返回nack消息重新入队

• RepublishMessageRecoverer重试耗尽后将失败消息投递到指定的交换机

比较优雅的一种处理方案是RepublishMessageRecoverer失败后将消息投递到一个指定的专门存放异常消息的队列后续由人工集中处理。

1在consumer服务中定义处理失败消息的交换机和队列

@Bean
public DirectExchange errorMessageExchange(){
    return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue(){
    return new Queue("error.queue", true);
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
    return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
}

 2定义一个RepublishMessageRecoverer关联队列和交换机

@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
    return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}

完整代码如下

@Configuration
@ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled", havingValue = "true")
public class ErrorMessageConfig {
    @Bean
    public DirectExchange errorMessageExchange(){
        return new DirectExchange("error.direct");
    }
    @Bean
    public Queue errorQueue(){
        return new Queue("error.queue", true);
    }
    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
        return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
    }

    @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }
}

---

3.4.业务幂等性

何为幂等性 幂等是一个数学概念用函数表达式来描述是这样的f(x) = f(f(x))例如求绝对值函数。 在程序开发中则是指同一个业务执行一次或多次对业务状态的影响是一致的。例如

• 根据id删除数据

• 查询数据

• 新增数据

但数据的更新往往不是幂等的如果重复执行可能造成不一样的后果。比如

• 取消订单恢复库存的业务。如果多次恢复就会出现库存重复增加的情况

• 退款业务。重复退款对商家而言会有经济损失。

所以我们要尽可能避免业务被重复执行。 然而在实际业务场景中由于意外经常会出现业务被重复执行的情况例如

• 页面卡顿时频繁刷新导致表单重复提交

• 服务间调用的重试

• MQ消息的重复投递

我们在用户支付成功后会发送MQ消息到交易服务修改订单状态为已支付就可能出现消息重复投递的情况。如果消费者不做判断很有可能导致消息被消费多次出现业务故障。 举例

1. 假如用户刚刚支付完成并且投递消息到交易服务交易服务更改订单为已支付状态。

2. 由于某种原因例如网络故障导致生产者没有得到确认隔了一段时间后重新投递给交易服务。

3. 但是在新投递的消息被消费之前用户选择了退款将订单状态改为了已退款状态。

4. 退款完成后新投递的消息才被消费那么订单状态会被再次改为已支付。业务异常。

因此我们必须想办法保证消息处理的幂等性。这里给出两种方案

• 唯一消息ID

• 业务状态判断

---

3.4.1.唯一消息ID

这个思路非常简单

1. 每一条消息都生成一个唯一的id与消息一起投递给消费者。

2. 消费者接收到消息后处理自己的业务业务处理成功后将消息ID保存到数据库

3. 如果下次又收到相同消息去数据库查询判断是否存在存在则为重复消息放弃处理。

我们该如何给消息添加唯一ID呢 其实很简单SpringAMQP的MessageConverter自带了MessageID的功能我们只要开启这个功能即可。

以Jackson的消息转换器为例

@Bean
public MessageConverter messageConverter(){
    // 1.定义消息转换器
    Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
    // 2.配置自动创建消息id用于识别不同消息也可以在业务中基于ID判断是否是重复消息
    jjmc.setCreateMessageIds(true);
    return jjmc;
}

3.4.2.业务判断

业务判断就是基于业务本身的逻辑或状态来判断是否是重复的请求或消息不同的业务场景判断的思路也不一样。 例如我们当前案例中处理消息的业务逻辑是把订单状态从未支付修改为已支付。因此我们就可以在执行业务时判断订单状态是否是未支付如果不是则证明订单已经被处理过无需重复处理。

相比较而言消息ID的方案需要改造原有的数据库所以我更推荐使用业务判断的方案。

以支付修改订单的业务为例我们需要修改OrderServiceImpl中的markOrderPaySuccess方法

 @Override
    public void markOrderPaySuccess(Long orderId) {
        // 1.查询订单
        Order old = getById(orderId);
        // 2.判断订单状态
        if (old == null || old.getStatus() != 1) {
            // 订单不存在或者订单状态不是1放弃处理
            return;
        }
        // 3.尝试更新订单
        Order order = new Order();
        order.setId(orderId);
        order.setStatus(2);
        order.setPayTime(LocalDateTime.now());
        updateById(order);
    }

上述代码逻辑上符合了幂等判断的需求但是由于判断和更新是两步动作因此在极小概率下可能存在线程安全问题。

我们可以合并上述操作为这样

@Override
public void markOrderPaySuccess(Long orderId) {
    // UPDATE `order` SET status = ? , pay_time = ? WHERE id = ? AND status = 1
    lambdaUpdate()
            .set(Order::getStatus, 2)
            .set(Order::getPayTime, LocalDateTime.now())
            .eq(Order::getId, orderId)
            .eq(Order::getStatus, 1)
            .update();
}

注意看上述代码等同于这样的SQL语句

UPDATE `order` SET status = ? , pay_time = ? WHERE id = ? AND status = 1

我们在where条件中除了判断id以外还加上了status必须为1的条件。如果条件不符说明订单已支付则SQL匹配不到数据根本不会执行。

---

3.5.兜底方案

虽然我们利用各种机制尽可能增加了消息的可靠性但也不好说能保证消息100%的可靠。万一真的MQ通知失败该怎么办呢 有没有其它兜底方案能够确保订单的支付状态一致呢

其实思想很简单既然MQ通知不一定发送到交易服务那么交易服务就必须自己主动去查询支付状态。这样即便支付服务的MQ通知失败我们依然能通过主动查询来保证订单状态的一致。 流程如下

图中黄色线圈起来的部分就是MQ通知失败后的兜底处理方案由交易服务自己主动去查询支付状态。

不过需要注意的是交易服务并不知道用户会在什么时候支付如果查询的时机不正确比如查询的时候用户正在支付中可能查询到的支付状态也不正确。 那么问题来了我们到底该在什么时间主动查询支付状态呢

这个时间是无法确定的因此通常我们采取的措施就是利用定时任务定期查询例如每隔20秒就查询一次并判断支付状态。如果发现订单已经支付则立刻更新订单状态为已支付即可。 定时任务大家之前学习过具体的实现这里就不再赘述了。

至此消息可靠性的问题已经解决了。

综上支付服务与交易服务之间的订单状态一致性是如何保证的

• 首先支付服务会正在用户支付成功以后利用MQ消息通知交易服务完成订单状态同步。

• 其次为了保证MQ消息的可靠性我们采用了生产者确认机制、消费者确认、消费者失败重试等策略确保消息投递的可靠性

• 最后我们还在交易服务设置了定时任务定期查询订单支付状态。这样即便MQ通知失败还可以利用定时任务作为兜底方案确保订单支付状态的最终一致性。

---

拓展应答和发布区别

应答功能属于消费者消费完消息告诉 RabbitMQ 已经消费成功。可以删了

发布功能属于生产者生产消息到 RabbitMQRabbitMQ 需要告诉生产者是否收到消息。生产者根据该消息来进行考虑要不要再发一次还是怎么样

        其实二者并没有什么关联应答是为了确保消息真正被用好了确保传递过来的消息被删除前都是合理处理后再进行删除操作与生产者无关它只要管自己要的数据处理好就行而发布是为了确保生产者自身的消息没有发送错误确保生产者消息传输的安全性它不用管消费者处理得怎么样它只需要管自己的消息都发出去了。

---

拓展备份交换机 

        有了 mandatory 参数和回退消息我们获得了对无法投递消息的感知能力有机会在生产者的消息无法被投递时发现并处理。但有时候我们并不知道该如何处理这些无法路由的消息将回退的消息的处理麻烦最多打个日志然后触发报警再来手动处理。而通过日志来处理这些无法路由的消息是很不优雅的做法特别是当生产者所在的服务有多台机器的时候手动复制日志会更加麻烦而且容易出错。而且设置 mandatory 参数会增加生产者的复杂性对备份消息的处理麻烦需要添加处理这些被退回的消息的逻辑。如果既不想丢失消息又不想增加生产者的复杂性该怎么做呢

        前面在设置死信队列的文章中我们提到可以为队列设置死信交换机来存储那些处理失败的消息可是这些不可路由消息根本没有机会进入到队列(因为进死信队列的前提是先进入队列中再进行判断是否进入死信队列你现在队列都进不了何来死信队列的说法呢)因此无法使用死信队列来保存消息。 在 RabbitMQ 中有一种备份交换机的机制存在可以很好的应对这个问题。

什么是备份交换机呢

备份交换机可以理解为 RabbitMQ 中交换机的“备胎”

原理步骤

当我们为某一个交换机声明一个对应的备份交换机时就是为它创建一个备胎当交换机接收到一条不可路由消息时将会把这条消息转发到备份交换机中由备份交换机来进行转发和处理通常备份交换机的类型为 Fanout 这样就能把所有消息都投递到与其绑定的队列中然后我们在备份交换机下绑定一个队列这样所有那些原交换机无法被路由的消息就会都进入这个队列了。当然我们还可以建立一个报警队列用独立的消费者来进行监测和报警。

当 RabbitMQ 发现一个消息无法被投递到目标队列时它会将该消息发送到备份交换机。备份交换机是一种特殊的交换机可以通过配置让 RabbitMQ 发送无法被路由的消息到指定的备份交换机。

Mandatory 参数与备份交换机可以一起使用的时候如果两者同时开启消息究竟何去何从谁优先级高经过上面结果显示答案是备份交换机优先级高。

---

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.fanoutRoutingKey为blue并设置消息的有效期为5000毫秒

注意尽管这里的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 由于我们安装的MQ是3.8版本因此这里下载3.8.17版本

具体的安装教程这里就不再赘述了网上这种配置类的文章很多大家可以去搜刮一波

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的方式

@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;
        }
    });
}

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

---

拓展优先级队列 

使用场景

        在我们系统中有一个订单催付的场景我们的客户在天猫下的订单淘宝会及时将订单推送给我们如果在用户设定的时间内未付款那么就会给用户推送一条短信提醒很简单的一个功能对吧。

        但是tmall 商家对我们来说肯定是要分大客户和小客户的对吧比如像苹果小米这样大商家一年起码能给我们创造很大的利润所以理应当然他们的订单必须得到优先处理而曾经我们的后端系统是使用 redis 来存放的定时轮询大家都知道 redis 只能用 List 做一个简简单单的消息队列并不能实现一个优先级的场景所以订单量大了后采用 RabbitMQ 进行改造和优化如果发现是大客户的订单给一个相对比较高的消息优先级 否则就是默认优先级

实现步骤

声明队列的时候添加优先级

设置队列的最大优先级 最大可以设置到 255 官网推荐 1-10 如果设置太高比较吃内存和 CPU

        RabbitMQ的优先级队列将消息的优先级与队列的优先级相结合以确保高优先级的消息首先被处理。这样做的确切原因是优先级队列的作用是确保高优先级的消息在队列中具有更高的权重以便它们更快地被处理。如果只设置了消息的优先级而没有为队列设置优先级则消息可能会被放置在具有低优先级的队列中从而导致高优先级的消息被延迟处理。因此为队列设置优先级是允许正确的优先级排序的必要步骤。就是说你如果想要设置优先级消息那你只能在优先级队列中操作

        注意要等所有的消息都发到MQ队列中才能开放消费者获取消息因为优先机队列的原理是见给所有的消息再MQ队列中进行排序将优先级高的消息排序到队列的前端这样子消费者以获取的就是队列最前的消息也就是优先级最高的消息

---

4.5.订单状态同步问题

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

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

优化后的实现思路如下

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

@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等常量

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逻辑

PayOrderDTO代码如下

@Data
@ApiModel(description = "支付单数据传输实体")
public class PayOrderDTO {
    @ApiModelProperty("id")
    private Long id;
    @ApiModelProperty("业务订单号")
    private Long bizOrderNo;
    @ApiModelProperty("支付单号")
    private Long payOrderNo;
    @ApiModelProperty("支付用户id")
    private Long bizUserId;
    @ApiModelProperty("支付渠道编码")
    private String payChannelCode;
    @ApiModelProperty("支付金额单位分")
    private Integer amount;
    @ApiModelProperty("付类型1h5,2:小程序3公众号4扫码5余额支付")
    private Integer payType;
    @ApiModelProperty("付状态0待提交1:待支付2支付超时或取消3支付成功")
    private Integer status;
    @ApiModelProperty("拓展字段用于传递不同渠道单独处理的字段")
    private String expandJson;
    @ApiModelProperty("第三方返回业务码")
    private String resultCode;
    @ApiModelProperty("第三方返回提示信息")
    private String resultMsg;
    @ApiModelProperty("支付成功时间")
    private LocalDateTime paySuccessTime;
    @ApiModelProperty("支付超时时间")
    private LocalDateTime payOverTime;
    @ApiModelProperty("支付二维码链接")
    private String qrCodeUrl;
    @ApiModelProperty("创建时间")
    private LocalDateTime createTime;
    @ApiModelProperty("更新时间")
    private LocalDateTime updateTime;
}

PayClient代码如下

@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代码如下

@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);
    }
}