RabbitMQ学习-发布确认高级

发布确认springboot版本

确认机制方案：

 代码架构图：

 配置文件：

在application.properties全局配置文件中添加spring.rabbitmq.publish-confirm-type属性，这个属性有以下几种值

none:禁用发布确认模式(默认)0
correlated:发布消息成功到交换机后会触发回调方法
simple：有两种效果
第一种效果是和correlated一样会触发回调方法
第二种效果是在发布消息成功以后使用rabbitTemplate调用waitForConfirms或者waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判单下一步的逻辑
waitForConfirmsOrDie方法如果返回false则会关闭信道，那么接下来就无法发送消息到broker

spring.rabbitmq.host=182.92.234.71
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123
spring.rabbitmq.publisher-confirm-type=correlated

添加配置类

@Configuration
public class ConfirmConfig {public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";public static final String CONFIRM_QUEUE_NAME = "confirm.queue";//声明业务 Exchange@Bean("confirmExchange")public DirectExchange confirmExchange(){return new DirectExchange(CONFIRM_EXCHANGE_NAME);}// 声明确认队列@Bean("confirmQueue")public Queue confirmQueue(){return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();}// 声明确认队列绑定关系@Beanpublic Binding queueBinding(@Qualifier("confirmQueue") Queue queue,@Qualifier("confirmExchange") DirectExchange exchange){return BindingBuilder.bind(queue).to(exchange).with("key1");}
}

消息生产者

@RestController
@RequestMapping("/confirm")
@Slf4j
public class Producer {public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";@Autowiredprivate RabbitTemplate rabbitTemplate;@Autowiredprivate MyCallBack myCallBack;//依赖注入 rabbitTemplate 之后再设置它的回调对象@PostConstructpublic void init(){rabbitTemplate.setConfirmCallback(myCallBack);}@GetMapping("sendMessage/{message}")public void sendMessage(@PathVariable String message){//指定消息 id 为 1CorrelationData correlationData1=new CorrelationData("1");String routingKey="key1";rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey,message+routingKey,correl
ationData1);CorrelationData correlationData2=new CorrelationData("2");routingKey="key2";rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey,message+routingKey,correl
ationData2);log.info("发送消息内容:{}",message);}
}

回调接口

@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback {/*** 交换机不管是否收到消息的一个回调方法* CorrelationData* 消息相关数据* ack* 交换机是否收到消息*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {String id=correlationData!=null?correlationData.getId():"";if(ack){log.info("交换机已经收到 id 为:{}的消息",id);}else{log.info("交换机还未收到 id 为:{}消息,由于原因:{}",id,cause);}}
}

消息消费者

@Component
@Slf4j
public class ConfirmConsumer {public static final String CONFIRM_QUEUE_NAME = "confirm.queue";@RabbitListener(queues =CONFIRM_QUEUE_NAME)public void receiveMsg(Message message){String msg=new String(message.getBody());log.info("接受到队列 confirm.queue 消息:{}",msg);}
}

结果分析

可以看到，发送了两条消息，第一条消息的 RoutingKey 为 "key1" ，第二条消息的 RoutingKey 为 "key2"，两条消息都成功被交换机接收，也收到了交换机的确认回调，但消费者只收到了一条消息，因为 第二条消息的 RoutingKey 与队列的 BindingKey 不一致，也没有其它队列能接收这个消息，所有第二条 消息被直接丢弃了。

 

回退消息

mandatory参数

如果我们仅仅开启了生产者确认机制，那么当交换机接受到消息后，会直接给生产者发送确认消息，但是如果发现消息不可以路由，就会直接把消息丢弃，此时消费者接受不到消息，而且这个时候生产者也不知道消息被丢弃了，这样就导致消息丢失，我们可以通过设置mandatory参数，是的消息在传递过程中出现不可到达的目的的时候可以把消息返回给生产者

@Slf4j
@Component
public class MessageProducer implements RabbitTemplate.ConfirmCallback , 
RabbitTemplate.ReturnCallback {@Autowiredprivate RabbitTemplate rabbitTemplate;//rabbitTemplate 注入之后就设置该值@PostConstructprivate void init() {rabbitTemplate.setConfirmCallback(this);/*** true：* 交换机无法将消息进行路由时，会将该消息返回给生产者* false：* 如果发现消息无法进行路由，则直接丢弃*/rabbitTemplate.setMandatory(true);//设置回退消息交给谁处理rabbitTemplate.setReturnCallback(this);}@GetMapping("sendMessage")
public void sendMessage(String message){//让消息绑定一个 id 值CorrelationData correlationData1 = new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend("confirm.exchange","key1",message+"key1",correlationData1)
;log.info("发送消息 id 为:{}内容为{}",correlationData1.getId(),message+"key1");CorrelationData correlationData2 = new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend("confirm.exchange","key2",message+"key2",correlationData2)
;log.info("发送消息 id 为:{}内容为{}",correlationData2.getId(),message+"key2");
}@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {String id = correlationData != null ? correlationData.getId() : "";if (ack) {log.info("交换机收到消息确认成功, id:{}", id);} else {log.error("消息 id:{}未成功投递到交换机,原因是:{}", id, cause);}}@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String 
exchange, String routingKey) {log.info("消息:{}被服务器退回，退回原因:{}, 交换机是:{}, 路由 key:{}",new String(message.getBody()),replyText, exchange, routingKey);}
}

回调接口：

@Component
@Slf4j
public class MyCallBack implements 
RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {/*** 交换机不管是否收到消息的一个回调方法* CorrelationData* 消息相关数据* ack* 交换机是否收到消息*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {String id=correlationData!=null?correlationData.getId():"";if(ack){log.info("交换机已经收到 id 为:{}的消息",id);}else{log.info("交换机还未收到 id 为:{}消息,由于原因:{}",id,cause);}}//当消息无法路由的时候的回调方法@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String 
exchange, String routingKey) {log.error(" 消 息 {}, 被交换机 {} 退回，退回原因 :{}, 路 由 key:{}",new 
String(message.getBody()),exchange,replyText,routingKey);}
}

结果分析：