做IP授权的一般看什么网站,建站工具cms,永久免费不收费的聊天软件app,wordpress是不是做网页这篇文章是了解MQ消息的可靠性#xff0c;即#xff1a;消息应该至少被消费者处理1次
那么问题来了#xff1a;
我们该如何确保MQ消息的可靠性#xff1f;如果真的发送失败#xff0c;有没有其它的兜底方案#xff1f; 首先#xff0c;我们一起分析一下消息丢失的可能…这篇文章是了解MQ消息的可靠性即消息应该至少被消费者处理1次
那么问题来了
我们该如何确保MQ消息的可靠性如果真的发送失败有没有其它的兜底方案 首先我们一起分析一下消息丢失的可能性有哪些。消息从发送者发送消息到消费者处理消息需要经过的流程是这样的 消息从生产者到消费者的每一步都可能导致消息丢失
发送消息时丢失 生产者发送消息时连接MQ失败生产者发送消息到达MQ后未找到Exchange生产者发送消息到达MQ的Exchange后未找到合适的Queue消息到达MQ后处理消息的进程发生异常MQ导致消息丢失 消息到达MQ保存到队列后尚未消费就突然宕机消费者处理消息时 消息接收后尚未处理突然宕机消息接收后处理过程中抛出异常
综上我们要解决消息丢失问题保证MQ的可靠性就必须从3个方面入手
确保生产者一定把消息发送到MQ确保MQ不会将消息弄丢确保消费者一定要处理消息
这一章我们先来看如何确保生产者一定能把消息发送到MQ。
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 * multipliermax-attempts: 3 # 最大重试次数
停掉RabbitMQ服务
然后测试发送一条消息会发现会每隔1秒重试1次总共重试了3次。消息发送的超时重试机制配置成功了
注意当网络不稳定的时候利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试也就是说多次重试等待的过程中当前线程是被阻塞的。如果对于业务性能有要求建议禁用重试机制。如果一定要使用请合理配置等待时长和重试次数当然也可以考虑使用异步线程来执行发送消息的代码。
2.生产者确认机制 一般情况下只要生产者与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机制。默认两种机制都是关闭状态需要通过配置文件来开启。
3.实现生产者确认 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回调机制。
3.2.定义ReturnCallback
每个RabbitTemplate只能配置一个ReturnCallback因此我们可以在配置类中统一设置。我们在publisher模块定义一个配置类 内容如下
package com.itheima.publisher.config;import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;import javax.annotation.PostConstruct;Slf4j
AllArgsConstructor
Configuration
public class MqConfig {private final RabbitTemplate rabbitTemplate;PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {Overridepublic 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());}});}
}
3.3.定义ConfirmCallback
由于每个消息发送时的处理逻辑不一定相同因此ConfirmCallback需要在每次发消息时定义。具体来说是在调用RabbitTemplate中的convertAndSend方法时多传递一个参数 这里的CorrelationData中包含两个核心的东西
id消息的唯一标示MQ对不同的消息的回执以此做判断避免混淆SettableListenableFuture回执结果的Future对象
将来MQ的回执就会通过这个Future来返回我们可以提前给CorrelationData中的Future添加回调函数来处理消息回执 我们新建一个测试向系统自带的交换机发送消息并且添加ConfirmCallback
Test
void testPublisherConfirm() {// 1.创建CorrelationDataCorrelationData cd new CorrelationData();// 2.给Future添加ConfirmCallbackcd.getFuture().addCallback(new ListenableFutureCallbackCorrelationData.Confirm() {Overridepublic void onFailure(Throwable ex) {// 2.1.Future发生异常时的处理逻辑基本不会触发log.error(send message fail, ex);}Overridepublic 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。
:::warning注意开启生产者确认比较消耗MQ性能一般不建议开启。而且大家思考一下触发确认的几种情况
路由失败一般是因为RoutingKey错误导致往往是编程导致交换机名称错误同样是编程错误导致MQ内部故障这种需要处理但概率往往较低。因此只有对消息可靠性要求非常高的业务才需要开启而且仅仅需要开启ConfirmCallback处理nack就可以了。
