0
点赞
收藏
分享

微信扫一扫

SQLite—免费开源数据库系列文章目录

cnlinkchina 03-14 09:00 阅读 4
  • 我们该如何确保MQ消息的可靠性

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

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

生产者的可靠性

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

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

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

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

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

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

    • 消息到达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服务:

命令: 

[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次。消息发送的超时重试机制配置成功了!

注意事项: 

生产者确认机制

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

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

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

示意图:

总结:

其中acknack属于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有三种模式可选:

一般推荐使用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包含两个核心的东西:

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

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

我在一个测试类下,新增一个testConfirmCallback方法:

举报

相关推荐

0 条评论