-
我们该如何确保MQ消息的可靠性?
-
如果真的发送失败,有没有其它的兜底方案?
这些问题,在这一次的学习中都会找到答案。
生产者的可靠性
首先,我们一起分析一下消息丢失的可能性有哪些。
消息从发送者发送消息,到消费者处理消息,需要经过的流程是这样的:
消息从生产者到消费这的每一步都有可能导致消息丢失:
-
发送消息时丢失:
-
生产者发送消息时连接MQ失败
-
生产者发送消息到达MQ后未找到
Exchange
-
生产者发送消息到达MQ的
Exchange
后,未找到合适的Queue
-
消息到达MQ后,处理消息的进程发生异常
-
-
MQ导致消息丢失:
-
消息到达MQ,保存到队列后,尚未消费就突然宕机
-
-
消费者处理消息时:
-
消息接收后尚未处理突然宕机
-
消息接收后处理过程中抛出异常
-
所以,我们要解决消息丢失问题,保证MQ的可靠性,就必须从三个方面入手
确保生产者一定把消息发送到MQ
确保MQ不会将消息弄丢
确保消费者一定要处理消息
生产者重试机制
生产者发送消息时,出现了网络故障,导致与MQ的连接中断。
为了解决这一问题,SpringAMQP提供的消息发送时的重试机制。即:当RabbitTemplate
与MQ连接超时后,多次重试。
修改publisher工程的yml配置文件
配置文件:
rabbitmq:host: 192.168.200.131port: 5672virtual-host: /hmallusername: hmallpassword: 123connection-timeout: 1S # 设置连接超时时间template:retry:enabled: true # 开启超时重试机制initial-interval: 1000ms #失败后的初始等待时间multiplier: 1 #失败后下次的等待时长倍数,下次等待时长 = 失败后的初始等待时间(initial-interval) * multipliermax-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.131port: 5672virtual-host: /hmallusername: hmallpassword: 123publisher-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;@PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic 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方法: