程序员 给老婆做网站,网站建设课程体会,wordpress导出淘宝,seo网站计划书文章目录 前言一、生产者重试机制二、生产者确认机制实现生产者确认#xff08;1#xff09;定义ReturnCallback#xff08;2#xff09;定义ConfirmCallback 总结 前言
生产者重试机制、生产者确认机制。 一、生产者重试机制 问题#xff1a;生产者发送消息时#xff0… 文章目录 前言一、生产者重试机制二、生产者确认机制实现生产者确认1定义ReturnCallback2定义ConfirmCallback 总结 前言
生产者重试机制、生产者确认机制。 一、生产者重试机制 问题生产者发送消息时出现了网络故障导致与MQ的连接中断。解决SpringAMQP提供的消息发送时的重试机制。即当RabbitTemplate与MQ连接超时后多次重试。 实现 需要配置application.yaml文件
spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled: true # 开启超时重试机制initial-interval: 1000ms # 失败后的初始等待时间multiplier: 1 # 失败后下次的等待时长倍数下次等待时长 initial-interval * multipliermax-attempts: 3 # 最大重试次数我们利用命令停掉RabbitMQ服务
docker stop mq然后测试发送一条消息会发现会每隔1秒重试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机制。 默认两种机制都是关闭状态需要通过配置文件来开启。
实现生产者确认
发送者的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回调机制。
1定义ReturnCallback
每个RabbitTemplate只能配置一个ReturnCallback因此我们可以在配置类中统一设置。我们在发送者定义一个配置类
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;Slf4j
Configuration
public class MqConfirmConfig implements ApplicationContextAware {Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取RabbitTemplateRabbitTemplate rabbitTemplate applicationContext.getBean(RabbitTemplate.class);// 设置ReturnCallback(回调)rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {log.debug(收到消息的return callback, exchange:{},key:{}, msg:{}, code:{},text:{}, returnedMessage.getExchange(),returnedMessage.getRoutingKey(),returnedMessage.getMessage(),returnedMessage.getReplyCode(),returnedMessage.getReplyText());}});}
}2定义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.发送消息 RoutingKey是错误的rabbitTemplate.convertAndSend(dragon.direct, q, hello, cd);
}由于传递的RoutingKey是错误的路由失败后触发了return callback同时也收到了ack。当我们修改为正确的RoutingKey以后就不会触发return callback了只收到ack。而如果连交换机都是错误的则只会收到nack。 注意
开启生产者确认比较消耗MQ性能一般不建议开启。而且大家思考一下触发确认的几种情况
路由失败一般是因为RoutingKey错误导致往往是编程导致交换机名称错误同样是编程错误导致MQ内部故障这种需要处理但概率往往较低。因此只有对消息可靠性要求非常高的业务才需要开启而且仅仅需要开启ConfirmCallback处理nack就可以了。 总结
以上就是全部讲解。
