rabbitMq确认机制之ConfirmType

配置方式

 @Bean(name = "connectionFactory")@Primarypublic 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;}

或者配置文件里配置

spring:# RabbitMQ 配置项，对应 RabbitProperties 配置类rabbitmq:publisher-confirm-type: correlated

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

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

使用方式

SIMPLE模式

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

  RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {@Overridepublic 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);});

CORRELATED模式

     RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {@Overridepublic 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成功");