rabbitMq确认机制之ConfirmType_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更新为发送成功，或者发送失败。
对于发送失败的安排重试，可以在消息头加上重试次数记录重试次数，达到指定次数，更新为发送失败。