RabbitMQ之如何保证发送消息的可靠性?

1.消费发送的机制

    1.1消息发送我们都知道会先发送到交换机上，然后再根据定的路由规则，由交换机将消息路由到不同的 Queue（队列）中，再由不同的消费者去消费。如下图

所以我们就应该保证消息成功到达交换机 和对列,如果都做到了纳闷我们消息就发送成功了对吧

2.常见的方案

    2.1开启事务机制

    2.2 发送方确认机制

3.MQ事务机制确保消息可靠性（不建议）

    3.1 在RabbitMQ的配置类中，准备一个事务管理器（这里我只贴出MQ配置类关于事务的）

 /**
     * 事务管理器
     * @param connectionFactory
     * @return
     */
    @Bean
    RabbitTransactionManager transactionManager(ConnectionFactory connectionFactory) {
        return new RabbitTransactionManager(connectionFactory);
    }

    3.2 在生产者（Product）打上@Transactional注解（事务） 然后我们在发送消息的时候调用 setChannelTransacted 方法设置为 true 开启事务模式

  rabbitTemplate.setChannelTransacted(true);

3.3 这个时候我们的事务就已经OK了,我们模拟一下，在发送消息的时候制造一个错误

 @Autowired
    RabbitTemplate rabbitTemplate;
    @RequestMapping("/direct")
    @Transactional
    public String faSon(){
            rabbitTemplate.setChannelTransacted(true);
            rabbitTemplate.convertAndSend(RabbitMqConfig.DIRECT, "direct", "direct");
            int i = 1/0;
        return "消费者消息发送成功";
      }

由上图我们可以看见我们制造了一个int i = 1/0的错误，所以当我们去发送消息的时候,我们后台抛出了异常，我们的对列也没有消息投送过来

这个时候可以看出来我们的事务启效果了,接下来我们修改下代码，不开启事务机制，

把事务管理器注销,然后在发送者这边修改代码

 public String faSon(){
            rabbitTemplate.convertAndSend(RabbitMqConfig.DIRECT, "direct", "direct");
            int i = 1/0;
        return "消费者消息发送成功";
      }

这个时候可以看出来我们已经去掉了事务,然后我们启动，访问在去看看效果

这里后台肯定会报错抛异常的,主要是我们去看看MQ的可视化页面

可以从上图看出,我们的消息依然发送成功了,说明我们这个消息不可靠,尽管我们后台出错了,也发了,所以我们就可以通过事务来解决了消息的可靠性，但是不建议，为什么了,因为事务模式其实效率有点低，这并非一个最佳解决方案。

4.发送方确认机制

    4.1配置yml

   publisher-confirm-type: correlated #到达交换器的确认回调 none：表示禁用发布确认模式，默认即此。correlated：表示成功发布消息到交换器后会触发的回调方法。 simple：类似 correlated，并且支持 waitForConfirms() 和 waitForConfirmsOrDie() 方法的调用。
    publisher-returns: true #消息到达队列的回调

    4.2 然后写配置类

    

Configuration
public class RabbitMqConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    public  static  final  String DIRECT = "DIRECT";
    public  static  final  String  TEST_QUEUE= "TopicExchange";
    @Autowired
    RabbitTemplate rabbitTemplate;
    /**
     * 声明交换机
     */
    @Bean
    public DirectExchange getExchange(){
        return new DirectExchange(DIRECT,true,false);
    }
    /**
     * 声明对列
     */
    @Bean
    public Queue getQueue(){
        return new Queue(TEST_QUEUE,true,false,false);
    }

    /**
     * 绑定交换机对对列
     */
    @Bean
    public Binding bindingQueueExchange(){
        return BindingBuilder.bind(getQueue()).to(getExchange()).with("direct");
    }

  
    /**
     * 消息到达交换机回调
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            if (ack){
                System.out.println("消息成功到达交换器");
            }else {
                System.out.println("消息发送失败");
            }
    }

    /**
     * 消息到达对列的回调
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("消息未成功路由到队列");
    }
}

这里我们实现了

RabbitTemplate.ConfirmCallback

和

RabbitTemplate.ReturnsCallback

两个接口，这两个接口，前者的回调用来确定消息到达交换器，后者则会在消息路由到队列失败时被调用。

然后在配置类中定义 initRabbitTemplate 方法并添加 @PostConstruct 注解，在该方法中为 rabbitTemplate 分别配置这两个 Callback。

@PostConstruct
    public void initRabbitTemplate() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

接下来我们就去发送消息，我们先尝试发送到一个没有的交换机上面看看

我们后台打印了出结果

由此可见我们发送到交换机失败了，确认机制起了作用，我们在发到没有的对列看看效果

我们后台效果看出了发送到对列失败了，我们的回调监听到了,这就是 publisher-confirm 模式。相比于事务，这种模式下的消息吞吐量会得到极大的提升

5.重试机制

    5.1自带的重试机制

前面所说的事务机制和发送方确认机制，都是发送方确认消息发送成功的办法。如果发送方一开始就连不上 MQ，那么 Spring Boot 中也有相应的重试机制，但是这个重试机制就和 MQ 本身没有关系了，这是利用 Spring 中的 retry 机制来完成的，具体配置如下：

template:
      retry:
        enabled: true #开启重试机制
        initial-interval: 1000ms #重试起始间隔时间
        max-interval: 10 #最大重试间隔时间。
        max-attempts: 10 #重试次数
        multiplier: 2 #间隔时间乘数。（这里配置间隔时间乘数为 2，则第一次间隔时间 1 秒，第二次重试间隔时间 2 秒，第三次 4 秒，以此类推）

再次启动 Spring Boot 项目，然后关掉 MQ，此时尝试发送消息，就会发送失败，进而导致自动重试。

好了，今天就先写到这来了,下一章会讲如何确保消息的成功消息，嘿嘿觉得可以的老铁点个赞！