微服务技术栈之rabbitMQ高级（二）

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

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

这些问题，在这一次的学习中都会找到答案。

生产者的可靠性

首先，我们一起分析一下消息丢失的可能性有哪些。

消息从发送者发送消息，到消费者处理消息，需要经过的流程是这样的：

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

• 发送消息时丢失：

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

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

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

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

• MQ导致消息丢失：

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

• 消费者处理消息时：

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

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

所以，我们要解决消息丢失问题，保证MQ的可靠性，就必须从三个方面入手

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

• 确保MQ不会将消息弄丢

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

生产者重试机制

生产者发送消息时，出现了网络故障，导致与MQ的连接中断。

为了解决这一问题，SpringAMQP提供的消息发送时的重试机制。即：当RabbitTemplate与MQ连接超时后，多次重试。

修改publisher工程的yml配置文件

配置文件： 

  rabbitmq:
    host: 192.168.200.131
    port: 5672
    virtual-host: /hmall
    username: hmall
    password: 123
    connection-timeout: 1S # 设置连接超时时间
    template:
      retry:
        enabled: true # 开启超时重试机制
        initial-interval: 1000ms #失败后的初始等待时间
        multiplier: 1 #失败后下次的等待时长倍数，下次等待时长 = 失败后的初始等待时间（initial-interval） * multiplier
        max-attempts: 3 # 最大重试次数

停止mq服务：

docker stop mq

命令： 

[root@localhost ~]# docker stop mq
mq
[root@localhost ~]#
 

启动测试：

然后我们测试TestQueue方法。测试发送一条消息。

代码：

    @Test
    @DisplayName("简单的队列测试")
   public void testSimpleQueue(){
        //1，队列名称
        String queueName = "simple.queue";
 
        //2，消息
        String msg = "hello world";
        //3，发送消息
        rabbitTemplate.convertAndSend(queueName,msg);
 
    }

看结果：

 发现会每隔1秒（连接超时1秒，隔1秒；所以输出的时候显示每隔2秒）重试1次，总共重试了3次。消息发送的超时重试机制配置成功了！

注意事项： 

当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的。

如果对于业务性能有要求，建议禁用重试机制。如果一定要使用，请合理配置等待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码。

生产者确认机制

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

但是，不排除在下面的情况也会出现消息发送到MQ之后丢失的现象，例如：

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

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

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

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

示意图：

总结：

• 当消息投递到MQ，但是路由失败时，通过Publisher Return返回异常信息，同时返回ack的确认信息，代表投递成功

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

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

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

其中ack和nack属于Publisher Confirm机制，ack是投递成功；nack是投递失败。而return则属于Publisher Return机制。

默认两种机制都是关闭状态，需要通过配置文件来开启。

实现生产者确认

开启生产者确认：

在publisher生产者工程的yml添加下面的配置

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

代码

  rabbitmq:
    host: 192.168.200.131
    port: 5672
    virtual-host: /hmall
    username: hmall
    password: 123
    publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
    publisher-returns: true # 开启publisher return机制

publisher-confirm-type有三种模式可选：

• none：关闭confirm

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

• correlated：MQ异步回调返回回执

一般推荐使用correlated，回调机制

定义ReturnCallback：

每个RabbitTemplate只能配置一个ReturnCallback，因此我们可以在一个配置类中，统一进行配置。我在publisher模板定义一个配置类：

Mqconfig配置类

代码：

@Slf4j
@Configuration
@RequiredArgsConstructor
public class MqConfig {
 
    private final RabbitTemplate rabbitTemplate;
 
    @PostConstruct
    public void init(){
       rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
           @Override
           public void returnedMessage(ReturnedMessage returnedMessage) {
               log.debug("触发了return callback");
               System.out.println("交换机exchange："+returnedMessage.getExchange());
               System.out.println("routingKey路由key："+returnedMessage.getRoutingKey());
               System.out.println("消息体message："+returnedMessage.getMessage());
               System.out.println("应答码：replayCode："+returnedMessage.getReplyCode());
               System.out.println("应答信息：replayText："+returnedMessage.getReplyText());
           }
       });
    }
 
}

 定义ConfirmCallback：

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

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

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

• SettableListenableFuture：回执结果的Future对象

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

新增一个测试方法，向交换机发送消息，并且routingkey是不存在的，来进行测试，并且添加ConfirmCallback：

我在一个测试类下，新增一个testConfirmCallback方法：