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

news2025/4/21 13:33:32

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

一、发送者重连

spring:
  rabbitmq:
    connection-timeout: 1s #设置MQ的连接超时时间
    template:
      retry:
        enabled: true #开启超时重试机制
        initial-interval: 1000ms #失败后的初始等待时间
        multiplier: 1 #失败后下次的等待时长倍数，下次等待时长=initial-interva * multiplier
        max-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;
    @PostConstruct
    public 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() {
       @Override
       public void returnedMessage(ReturnedMessage returnedMessage) {
           //TODO
       }
   });

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

@Test
    public void testConfirmCallback() throws InterruptedException {
        //创建correlationData
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        correlationData.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("spring amqp 处理确认结果异常", ex);
            }
            @Override
            public 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);
    }