SpringBoot集成RabbitMQ实现发布确认
发布日期:2022-02-10 08:11:08 浏览次数:5 分类:技术文章

本文共 4796 字,大约阅读时间需要 15 分钟。

前言

何为发布确认?

image.png

字面意思:P发送消息之后,能够确定C收到消息了

问题

P发送消息要通过交换机和队列,所以只需要确保这二者都不出问题,或者有PlanB来防止出意外即可

问题一:交换机出问题

不明原因,导致交换机重启,在重启过程中,生产者投递消息失败,导致消息丢失;

交换机宕机了,没有集群

。。。

问题二:队列出问题

交换机和队列之间的bingKey错误,导致找不到指定队列;

队列TTL已到

。。。

解决方案

交换机问题解决

交换机是否收到消息,生产者是不知道的,所以需要写一个回调方法,通知生产者,消息是否被交换机接收到

队列问题解决

交换机是否找到对应队列,交换机如果找不到对应队列,会将消息丢掉,这种情况是不允许发生的,所以需要把不可达的消息返回交换机,交给其他(备份)交换机重新发送

流程图

备份交换机架构图.png

实现步骤

声明配置
spring:  rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
publisher-confirm-type: correlated
   #开启发布确认 默认是关闭的
publisher-returns: true
  #开启消息回退  交换机找不到队列时,会直接丢弃消息,回退给生产者
声明 确定队列,备份队列,警告队列,业务交换机,备份交换机,以及它们之间的绑定关系
@Configurationpublic class ConfirmConfig {

   public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";
public static final String CONFIRM_QUEUE_NAME = "confirm.queue";
public static final String BACKUP_EXCHANGE_NAME = "backup.exchange";
public static final String BACKUP_QUEUE_NAME = "backup.queue";
public static final String WARNING_QUEUE_NAME = "warning.queue";
//声明确定队列
@Bean("confirQueue")
public Queue queue(){
   return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
}
//声明业务交换机
@Bean("confirExchange")
public DirectExchange directExchange(){
   return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true).withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
}
//绑定确定队列和业务交换机
@Bean
public Binding bindingQueueToExchange(@Qualifier("confirQueue") Queue confirQueue,
  @Qualifier("confirExchange")DirectExchange confirExchange){
   return BindingBuilder.bind(confirQueue).to(confirExchange).with("key1");
}
//声明备份 Exchange
@Bean("backupExchange")
public FanoutExchange backupExchange(){
   return new FanoutExchange(BACKUP_EXCHANGE_NAME);
}
//声明备份队列
@Bean("backQueue")
public Queue backQueue(){
   return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
}
//声明报警队列
@Bean("warningQueue")
public Queue warningQueue(){
   return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
}
//声明备份队列和 备份交换机之间的绑定关系
@Bean
public Binding backupBinding(@Qualifier("backQueue") Queue queue,
 @Qualifier("backupExchange") FanoutExchange backupExchange){
   return BindingBuilder.bind(queue).to(backupExchange);
}
//声明报警队列和 备份交换机之间的绑定关系
@Bean
public Binding warningBinding(@Qualifier("warningQueue") Queue queue,
  @Qualifier("backupExchange") FanoutExchange
  backupExchange){
   return BindingBuilder.bind(queue).to(backupExchange);
}}
声明回滚类
@Component@Slf4jpublic class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback{

   /**
 *  交换机是否收到消息的一个回调方法
 *
 * @param correlationData  消息相关数据
 * @param ack
  交换机是否收到消息
 * @param s
 失败 的原因,成功的话为null
 */
@Override
public void confirm(CorrelationData correlationData, boolean ack, String s) {
   String id = correlationData!=null?correlationData.getId():"";
if (ack){
  log.info("交换机已收到id为:{}",id);
}else{
   log.info("交换机还未收到 id 为:{}消息,由于原因:{}",id,s);
}
}
/**
 * 只有当消息传递过程中不可达目的地时,将消息返回给生产者
 * @param returnedMessage
 */
@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
   log.error("消息:{},被交换机:{}退回,退回原因:{},路由key:{}",returnedMessage.getMessage(),returnedMessage.getExchange(),returnedMessage.getReplyText(),returnedMessage.getRoutingKey());
}}
声明生产者
@RestController@RequestMapping("/confirm")@Slf4jpublic class ConfirmProducer {

   public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private MyCallBack myCallBack;
//依赖注入rabbitMQ之后再设置它的回调对象
@PostConstruct
public void init(){
   rabbitTemplate.setConfirmCallback(myCallBack);
rabbitTemplate.setReturnsCallback(myCallBack);
}
//发送消息
@GetMapping("/sendMessage/{message}")
public void sendMessage(@PathVariable String message){
   //指定消息id为1
CorrelationData correlationData1 = new CorrelationData("1");
String routintKey = "key1";
rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routintKey,message+routintKey,correlationData1);
log.info("发送的消息是:{}",message);
//指定消息id为2
CorrelationData correlationData2 = new CorrelationData("2");
rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routintKey+123,message+routintKey,correlationData2);
log.info("发送的消息是:{}",message);
}}
声明确认消费者
@Component@Slf4jpublic class ConfirmCustomer {

   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("接收确定队列传来的消息:{}",msg);
}}
声明警告消费者
@Component@Slf4jpublic class WarningConsumer {

   public static final String WARNING_QUEUE_NAME = "warning.queue";
@RabbitListener(queues = WARNING_QUEUE_NAME)
public void receiveWarningMsg(Message message) {
   String msg = new String(message.getBody());
log.error("报警发现不可路由消息:{}", msg);
}}
启动测试
http://localhost:8080/confirm/sendMessage/你爱我,我爱你

转载地址:https://blog.csdn.net/qq_43612495/article/details/120783017 如侵犯您的版权,请留言回复原文章的地址,我们会给您删除此文章,给您带来不便请您谅解!

上一篇:SpringBoot集成RabbitMQ实现延迟队列
下一篇:MarkDown学习

发表评论

最新留言

第一次来,支持一个
[***.191.171.21]2022年08月17日 11时06分59秒