RabbitMQ----生产者可靠性-Toy模板网

这篇具有很好参考价值的文章主要介绍了RabbitMQ----生产者可靠性。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

RabbitMQ----生产者可靠性

生产者可靠性主要分为两个方面：

生产者重连
生产者确认

一、生产者重连

有的时候由于网络波动，可能会出现客户端连接MO失败的情况。通过配置我们可以开启连接失败后的重连机制：

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

注意：

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

二、生产者确认

RabbitMQ有Publisher Confirm和Publisher Return两种确认机制。开启确机制认后，在MO成功收到消息后会返回确认消息给生产者。返回的结果有以下几种情况：

消息投递到了MQ，但是路由失败。此时会通过Publisher Return返回路由异常原因，然后返回ACK，告知投递成功
临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
持久消息投递到了MO，并且入队完成持久化，返回ACK ，告知投递成功
其它情况都会返回NACK，告知投递失败

Publisher Return确认机制只有在交换机路由失败的时候才return返回ACK

RabbitMQ----生产者可靠性,RabbitMQ,java-rabbitmq,rabbitmq,java

SpringAMQP实现生产者确认

1、在publisher（生产者）这个微服务的application.yml中添加配置：

spring:
  rabbitmq:
    publisher-confirm-type: correlated # 开ublisher confirm机制，并设置confirm类型
    publisher-returns: true # 开publisher return机制(交换机路由失败时返回ACK，大部分原因是开发问题)

配置说明：这里publisher-confirm-type有三种模式可选

none：关闭confirm机制
simple：同步阻塞等待MO的回执消息
correlated：MO异步回调方式返回回执消息

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

@Slf4j
@Configuration
public class configConfirmConfig implements ApplicationContextAware {
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        //huoRabbitTemplate
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        //设置ReturnsCallback
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                log.debug("收到return callback,exchange:{},key:{},msg:{},code:{},text:{}",
                        returned.getExchange(), returned.getRoutingKey(),returned.getMessage(),
                        returned.getReplyCode(),returned.getReplyText());
            }
        });
    }
}

3、发送消息，指定消息ID，消息ConfirmCallback

@Test
    void testConfirmCallback() {
        //1、创建CorrelationData
        CorrelationData data = new CorrelationData(UUID.randomUUID().toString());
        //2、添加ConfirmCallback
        data.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("消息回调失败",ex);
            }

            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                log.debug("收到confirm callback 回执");
                if (result.isAck()){
                    //消息发送成功
                    log.debug("消息发送成功，收到ack");
                }else {
                    //消息发送失败，这里需要重发消息
                    log.error("消息发送失败，收到nack, 原因：{}",result.getReason());
                }
            }
        });
        rabbitTemplate.convertAndSend("hmall.direct","blue","hello",data);
    }