确认机制方案
代码架构图
配置文件
spring.rabbitmq.publisher-confirm-type=correlated
⚫ NONE
禁用发布确认模式，是默认值
⚫ CORRELATED
发布消息成功到交换器后会触发回调方法
⚫ SIMPLE
经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法， 其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法 等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker
spring.application.name=demo6 spring.rabbitmq.host=192.168.231.135 spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=123 server.port=8989 spring.rabbitmq.publisher-confirm-type=CORRELATED
添加配置类
/** * 发布确认的配置类 */ @Configuration @Slf4j public class config { public static final String CONFIRM_EXCHANGE_NAME="confirm.exchange"; public static final String CONFIRM_QUEUE_NAME="confirm.queue"; //声明业务交换机 @Bean("confirmExchange") public DirectExchange confirmExchange() { return new DirectExchange(CONFIRM_EXCHANGE_NAME); } //声明确认队列 @Bean("confirmQueue") public Queue confirmQueue() { return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build(); } //声明确认队列和绑定关系 @Bean public Binding queueBinding(@Qualifier("confirmQueue") Queue queue,@Qualifier("confirmExchange") DirectExchange exchange) { return BindingBuilder.bind(queue).to(exchange).with("key1"); } }
回调接口
@Component @Slf4j public class MyCallBack implements RabbitTemplate.ConfirmCallback { @Override public void confirm(CorrelationData correlationData, boolean ack ,String cause) { String id=correlationData!=null?correlationData.getId():""; if(ack) { log.info("交换机已经收到id为{}的消息",id); } else { log.info("交换机还未收到id未：{}的消息，原因是{}",cause); } } }
消息生产者
@RestController @RequestMapping("/confirm") @Slf4j public class produce { public static final String confirm_exchange_name="confirm.exchange"; @Autowired private RabbitTemplate rabbitTemplate; @Autowired private MyCallBack myCallBack; @PostConstruct public void init() { rabbitTemplate.setConfirmCallback(myCallBack); } @GetMapping("sendMessage1/{message}") public void sendMessage1(@PathVariable String message) { //指定消息id为1 CorrelationData correlationData1=new CorrelationData("1"); String routingKey="key1"; rabbitTemplate.convertAndSend(confirm_exchange_name,routingKey,message+routingKey,correlationData1); log.info("发送的消息的内容{}",message); CorrelationData correlationData2=new CorrelationData("2"); routingKey="key2"; rabbitTemplate.convertAndSend(confirm_exchange_name,routingKey,message+routingKey,correlationData2); log.info("发送的消息的内容{}",message); } }
消息消费者
@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 s = new String(message.getBody()); log.info("接收到队列confirm.queue消息：{}",s); } }
发送消息
http://localhost:8989/confirm/sendMessage1/3322
可以看到，发送了两条消息，第一条消息的 RoutingKey 为 "key1"，第二条消息的 RoutingKey 为 "key2"，两条消息都成功被交换机接收，也收到了交换机的确认回调，但消费者只收到了一条消息，因为第二条消息的 RoutingKey 与队列的 BindingKey 不一致，也没有其它队列能接收这个消息，所有第二条消息被直接丢弃了。
