当前位置：首页 > news >正文

RabbitMQ发送者重连、发送者确认

news 2025/7/9 7:08:35

RabbitMQ发送者重连、发送者确认

一、发送者重连

spring:rabbitmq:connection-timeout: 1s #设置MQ的连接超时时间template:retry:enabled: true #开启超时重试机制initial-interval: 1000ms #失败后的初始等待时间multiplier: 1 #失败后下次的等待时长倍数，下次等待时长=initial-interva * multipliermax-attempts: 3 #最大重试次数

当网络不稳定的时候，利用重试机制可以有效提高消息发送成功的的成功率，不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的，会影响业务性能。
如果对于业务性能有要求，建议禁用重试机制。如果一定要用，请合理配置等待时长和重试次数，当然也可考虑使用异步线程来执行发送消息的代码。

二、发送者确认
SpringAMQP提供了Publisher Confirm和 Publisher Return两种确认机制。开启确认机制以后，当发送者发送消息给MQ后，MQ会返回确认结果给发送者。返回的结果有以下几种情况：

消息投递到MQ，但是路由失败。此时会通过PublisherReturn返回路由异常原因，然后返回ＡＣＫ，告知投递成功。
临时消息投递到了ＭＱ，并且入队成功，返回ＡＣＫ，告知投递成功
持久消息投递到了ＭＱ，并且入队完成持久化，返回ＡＣＫ，告知投递成功
其他情况都会返回ＮＡＣＫ，告知投递失败。

１.在ｐｕｂｌｉｃｈｅｒ这个微服务的ａｐｐｌｉｃａｔｉｏｎ．ｙｍｌ中添加配置

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

配置说明这里ｐｕｂｌｉｓｈｅｒ－ｃｏｎｆｉｒｍ－ｔｙｐｅ有三种模式可选

ｎｏｎｅ：关闭ｃｏｎｆｉｒｍ机制
ｓｉｍｐｌｅ：同步阻塞等待ＭＱ的回执消息
ｃｏｒｒｅｌａｔｅｄ：ＭＱ异步回调方式返回回执消息

2.ReturnCallback:每个RabbitTemplate只能配置一个ReturnCallback，因此需要在项目启动过程中配置：

@Slf4j
@Configuration
@RequiredArgsConstructor
public class MqConfig {private final RabbitTemplate rabbitTemplate;@PostConstructpublic void init() {rabbitTemplate.setReturnsCallback(returnedMessage -> {log.error("监听到消息return callback");log.debug("exchange: {}", returnedMessage.getExchange());log.debug("routingKey: {}", returnedMessage.getRoutingKey());log.debug("message: {}", returnedMessage.getMessage());log.debug("replyCode: {}", returnedMessage.getReplyCode());log.debug("replyText: {}", returnedMessage.getReplyText());});}
==============================上面简化的是这部分内容=========================================================rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {//TODO}});

3.ConfirmCallback：发送消息，指定消息ID，消息ConfirmCallback

@Testpublic void testConfirmCallback() throws InterruptedException {//创建correlationDataCorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());correlationData.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {@Overridepublic void onFailure(Throwable ex) {log.error("spring amqp 处理确认结果异常", ex);}@Overridepublic void onSuccess(CorrelationData.Confirm result) {if (result.isAck()) {log.debug("收到ConfirmCallback ack,消息发送成功!!!");} else {log.error("收到ConfirmCallback nack,消息发送失败！reason：{}", result.getReason());}}});//交换机名称String exChangeNange = "hmall.topic";//消息String message = "TopicTopic！！！Topic！！！ ";//发送消息rabbitTemplate.convertAndSend(exChangeNange, "hina.new", message, correlationData);Thread.sleep(3000);}