RabbitMQ 消息丢失解决（高级发布确认、消息回退与重发、备份交换机）

一、发布确认SpringBoot版本

确认机制图例：

代码实战：

代码架构图：

1.1交换机的发布确认

添加配置类

消息消费者

消息生产者发布消息后的回调接口

测试：

1.2回退消息并重发（队列的发布确认）

修改回调接口

生产者：

测试：

二、备份交换机

实战

生产者

报警消费者：

测试：

一、发布确认SpringBoot版本

首先发布消息后进行备份在缓存里，如果消息成功发布确认到交换机，则从缓存里删除该消息，如果没有成功发布，则设置一个定时任务，重新从缓存里获取消息发布到交换机，直到成功发布到交换机。

确认机制图例：

代码实战：

一个交换机：confirm.exchange，一个队列：confirm.queue，一个消费者：confirm.consumer

其中交换机类型时 direct，与队列关联的 routingKey 是 key1

代码架构图：

1.1交换机的发布确认

配置文件中添加：

server:port: 8888
spring:rabbitmq:host: 192.168.163.132port: 5672username: 2252631565password: 2252631565
#    高级发布确认 发布消息成功后将会触发回调方法publisher-confirm-type: correlated

NONE 值是禁用发布确认模式，是默认值
CORRELATED 值是发布消息成功到交换器后会触发回调方法
SIMPLE 值经测试有两种效果，其一效果和 CORRELATED 值一样会触发回调方法，其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker;

添加配置类

声明交换机和队列，并且将交换机和队列进行绑定:

@Configuration
public class ConfirmConfig {public static final String CONFIRM_EXCHANGE="confirm.exchange";public static final String CONFIRM_QUEUE="confirm.queue";public static final String ROUTING_KEY="key1";@Beanpublic DirectExchange confirmExchange(){return new DirectExchange(CONFIRM_EXCHANGE,false,false);}@Beanpublic Queue confirmQueue(){return QueueBuilder.durable(CONFIRM_QUEUE).build();}@Beanpublic Binding EAndQBind(@Qualifier("confirmExchange") DirectExchange exchange,@Qualifier("confirmQueue")Queue queue){return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);}
}

消息生产者

也可以说是 Controller 层，在这里发送两条消息给两个交换机，其中一个交换机是我们设置好的，另一个交换机不存在；这样就可以清晰看出交换机应答效果。

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {@Autowiredprivate RabbitTemplate rabbitTemplate;//高级发布确认模式@GetMapping("/sendConfirmMsg/{message}")public void sendConfirmMsg(@PathVariable String message){log.info("发送一条时长为的消息给第一个队列内容是:{}",new Date().toString(),message);CorrelationData correlationData=new CorrelationData("1");correlationData.setReturnedMessage(new org.springframework.amqp.core.Message(message.getBytes()));rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY,message,correlationData);//向一个不存在的交换机发送消息log.info("发送一条时长为的消息给第一个队列内容是:{}",new Date().toString(),message);CorrelationData correlationData2=new CorrelationData("2");correlationData2.setReturnedMessage(new org.springframework.amqp.core.Message(message.getBytes()));rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE+123,ConfirmConfig.ROUTING_KEY,message,correlationData2);}}

消息消费者

监听 confirm.queue 队列

@Slf4j
@Component
public class ConfirmLetterQueue {@RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE)public void confirmConsumer(Message message, Channel channel){log.info("收到了消息：{}",new String(message.getBody()));}
}

消息生产者发布消息后的回调接口

只要生产者发布消息，交换机不管是否收到消息，都会调用该类的 confirm 方法

@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback {@Autowiredprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init(){//注入rabbitTemplate.setConfirmCallback(this::confirm);}/*交换机确认回调1.交换机收到了消息 触发回调1.1 correlationData(我们在发消息的时候自己创建的) 消息的ID以及消息内容1.2 ack 交换机收到消息 true1.3 cause 交换机收到消息的原因 null---------------------------------2.交换机未收到消息 触发回调2.1 correlationData 消息的ID以及消息内容2.2 ack 交换机未收到消息 false2.3 cause 失败原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if(ack){log.info("发送消息到交换机成功！消息体为：{}",new String(correlationData.getReturned().getMessage().getBody()));}else {log.info("发送消息到交换机失败！原因为：{}",cause.toString());}}
}

测试：

效果很明显，我们配置的交换机成功收到消息并转发给队列；不存在的交换机没有接受到消息并作出反应。

1.2回退消息并重发（队列的发布确认）

在配置文件中开启消息回退功能

server:port: 8888
spring:rabbitmq:host: 192.168.163.133port: 5672username: 2252631565password: 2252631565
#    高级发布确认 发布消息成功后将会触发回调方法publisher-confirm-type: correlated#    消息回退 当消息未路由至队列时触发publisher-returns: true

修改回调接口

实现 RabbitTemplate.ReturnsCallback 接口，并实现方法

@Slf4j
@Component
public class MyCallBack  implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback{@Autowiredprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init(){//注入rabbitTemplate.setConfirmCallback(this::confirm);rabbitTemplate.setReturnsCallback(this::returnedMessage);}/*交换机确认回调1.交换机收到了消息 触发回调1.1 correlationData(我们在发消息的时候自己创建的) 消息的ID以及消息内容1.2 ack 交换机收到消息 true1.3 cause 交换机收到消息的原因 null---------------------------------2.交换机未收到消息 触发回调2.1 correlationData 消息的ID以及消息内容2.2 ack 交换机未收到消息 false2.3 cause 失败原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if(ack){log.info("时间:{}发送消息到交换机成功！",new Date());}else {log.info("发送消息到交换机失败！原因为：{}",cause.toString());}}//当消息未路由到队列时触发 只有失败时才触发 若消息发送至延迟队列则一定会触发回退 记得根据交换机名称排除延迟队列@Overridepublic void returnedMessage(ReturnedMessage returned) {log.error("消息：'{}',被交换机：{}回退，回退原因为：{}，routingKey为：{}",new String(returned.getMessage().getBody()),returned.getExchange(),returned.getReplyText(),returned.getRoutingKey());//10s后消息重发try {Thread.sleep(10000);log.info("时间:{}，生产者重新发消息",new Date());rabbitTemplate.convertAndSend(returned.getExchange(),ConfirmConfig.ROUTING_KEY,new String(returned.getMessage().getBody()));}catch (InterruptedException e) {throw new RuntimeException(e);}}
}

生产者：

向交换机中发送消息，指定错误的routingkey,触发队列回退消息并重发消息。

    //高级发布确认模式@GetMapping("/sendConfirmMsg/{message}")public void sendConfirmMsg(@PathVariable String message){//向一个不存在的队列发送消息log.info("时间:{}生产者发送一条的消息给第一个队列内容是:{}",new Date().toString(),message);CorrelationData correlationData2=new CorrelationData("2");rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY+123,message,correlationData2);}

测试：

回退未进入队列的消息并重新发送消息。

二、备份交换机

什么是备份交换机呢？备份交换机可以理解为 RabbitMQ 中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout （扇出），这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警并可以重发消息。

实战

需要一个备份交换机 backup.exchange，类型为 fanout，该交换机发送消息到队列 backup.queue 和 warning.queue。

修改高级确认发布配置类

@Configuration
public class ConfirmConfig {public static final String CONFIRM_EXCHANGE="confirm.exchange";public static final String CONFIRM_QUEUE="confirm.queue";public static final String ROUTING_KEY="key1";//备份交换机public static final String BACKUP_EXCHANGE="backup.exchange";//备份队列public static final String BACKUP_QUEUE="backup.queue";//报警队列public static final String WARNING_QUEUE="warning.queue";@Beanpublic DirectExchange confirmExchange(){//绑定确认交换机与备份交换机Map<String,Object> argument=new HashMap<>();argument.put("alternate-exchange",BACKUP_EXCHANGE);return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).withArguments(argument).build();}//备份交换机@Beanpublic FanoutExchange backupExchange(){return new FanoutExchange(BACKUP_EXCHANGE);}@Beanpublic Queue confirmQueue(){return QueueBuilder.durable(CONFIRM_QUEUE).build();}//备份队列@Beanpublic Queue backupQueue(){return QueueBuilder.durable(BACKUP_QUEUE).build();}//警告队列@Beanpublic Queue warningQueue(){return QueueBuilder.durable(WARNING_QUEUE).build();}@Beanpublic Binding EAndQBind(@Qualifier("confirmExchange") DirectExchange exchange,@Qualifier("confirmQueue")Queue queue){return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);}//绑定备份交换机与两个队列@Beanpublic Binding BAndBBing(@Qualifier("backupExchange") FanoutExchange exchange,@Qualifier("backupQueue")Queue queue){return BindingBuilder.bind(queue).to(exchange);}@Beanpublic Binding BAndWBing(@Qualifier("backupExchange") FanoutExchange exchange,@Qualifier("warningQueue")Queue queue){return BindingBuilder.bind(queue).to(exchange);}
}

生产者

生产者发送两条消息一个配置正确的路由，另一个是错误的路由。预期目标是正确路由正常接收消息，错误路由传输的信息由警告队列接收。

    //高级发布确认模式@GetMapping("/sendConfirmMsg/{message}")public void sendConfirmMsg(@PathVariable String message){log.info("时间:{}生产者发送两条消息队列内容是:{}",new Date().toString(),message);CorrelationData correlationData=new CorrelationData("1");rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY,message,correlationData);//向一个不存在的队列发送消息CorrelationData correlationData2=new CorrelationData("2");rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY+123,message,correlationData2);}

报警消费者：

接收不可路由的消息

/*** 报警消费者*/
@Slf4j
@Component
public class WarningConsumer {//监听报警消息@RabbitListener(queues = ConfirmConfig.WARNING_QUEUE)public void receiveWarningMsg(Message message){String msg=new String(message.getBody());log.info("报警发现不可路由消息：{},重发消息",msg);}
}