当前位置:   article > 正文

rabbitMq确认机制之ConfirmType_publisher-confirm-type

publisher-confirm-type

配置方式

 @Bean(name = "connectionFactory")
    @Primary
    public ConnectionFactory normalConnectionFactory(
   
            @Value("${spring.rabbitmq.username}") String username,
            @Value("${spring.rabbitmq.password}") String password,
            @Value("${spring.rabbitmq.addresses}") String address) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
 
        connectionFactory.setAddresses(address);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
     //   connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.setExecutor(createThreadPool(10,
                20,
                "mq-connection-",
                "mq-connection-group"));
        return connectionFactory;
    }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21

或者配置文件里配置

spring:
  # RabbitMQ 配置项,对应 RabbitProperties 配置类
  rabbitmq:
    publisher-confirm-type: correlated
  • 1
  • 2
  • 3
  • 4

publisher-confirm-type属性有三个可选值:

  1. none(默认):关闭发布确认模式。
  2. correlated:消息从生产者发送到交换机后触发回调方法。
  3. simple:会触发回调方法,相当于单个确认(发一条确认一条)。在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果,根据返回结果来判定下一步的逻辑,要注意的点是 waitForConfirmsOrDie方法如果返回false则会关闭channel,则接下来无法发送消息到broker。

使用方式

SIMPLE模式

开启simple模式需要在invoke方法中一起执行 rabbitTemplate.waitForConfirms
同时也会收到回调,回调后结束阻塞,同时可以获取到返回结果

  RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack){
                    System.out.println(correlationData.toString() + "发送成功");
                }else {
                    System.out.println(correlationData.toString() + "发送失败, 原因: " + cause);
                }
            }
        };
        rabbitTemplate.setConfirmCallback(confirmCallback);
        Boolean invoke = rabbitTemplate.invoke(operations -> {
            rabbitTemplate.convertAndSend("direct_exchange", "ROUTING_KEY_01", message, correlationData);
            return rabbitTemplate.waitForConfirms(1000l);
        });

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16

CORRELATED模式

     RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack){
                    System.out.println(correlationData.toString() + "发送成功");
                }else {
                    System.out.println(correlationData.toString() + "发送失败, 原因: " + cause);
                }
            }
        };
        rabbitTemplate.setConfirmCallback(confirmCallback);
    
        rabbitTemplate.convertAndSend("direct_exchange", "ROUTING_KEY_01", message, correlationData);
   


    //    correlationData.getFuture().get();
        sleep(1000*60);
        System.out.println("发送消息boot mq hello Direct成功");
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

实现通过callback实现保证消息发送成功。

可以看出来,在开启publisher-confirm的情况下,如果不自行实现ConfirmCallback的逻辑,也无法做到保证消息成功发送。

可以在发送消息时更新为发送中。
收到callback更新为发送成功,或者发送失败。
对于发送失败的安排重试,可以在消息头加上重试次数记录重试次数,达到指定次数,更新为发送失败。

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/weixin_40725706/article/detail/689951
推荐阅读
相关标签
  

闽ICP备14008679号