基于SpringBoot解决RabbitMQ消息丢失问题
- 一、RabbitMQ解决消息丢失问题
- 二、方案实践
- 1、在生产者服务相关配置
- 2、在消费者服务相关配置
- 三、测试验证
- 1、依次启动RabbitMQ、producer(建议先清空队列里面旧的测试消息再启动consumer)和consumer
- 2、在producer中调用接口，发送消息
- 3、观察控制台打印日志
- 四、项目结构及源码
- 1、项目结构
- 2、源码下载
一、RabbitMQ解决消息丢失问题
RabbitMQ通过以下机制来保证消息的可靠性，从而解决消息丢失问题：
（1）消息持久化：RabbitMQ支持将消息持久化到磁盘，即使RabbitMQ服务器宕机或重启，消息也不会丢失。在发布消息时，可以设置消息的持久化标志，这样消息就会被写入磁盘中，而不是仅仅保存在内存中。在自定义MQ的配置中，设置消息队列和交换机，默认为true代表持久化。
（2）消息确认机制：RabbitMQ提供了消息确认机制，即生产者在发送消息后，可以等待RabbitMQ服务器返回确认信息，以确保消息已经被正确地接收和处理。如果RabbitMQ服务器没有返回确认信息，生产者可以选择重新发送消息或者采取其他的补救措施。在自定义MQ的配置中，配置RabbitTemplate，设置setConfirmCallback和setReturnCallback进行确认。
（3）事务机制：RabbitMQ还支持事务机制，即生产者可以将多个操作封装在一个事务中，只有当所有的操作都成功完成后，才提交事务。如果某个操作失败，整个事务会被回滚，从而保证消息的完整性和一致性。
（4）消息重试机制：如果消息在传输过程中出现异常，RabbitMQ会自动进行消息重试，直到消息被正确地处理为止。可以通过设置重试次数和重试时间间隔来控制消息重试的行为。可以在application.yaml配置文件中设置
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
virtual-host: /
#1、确保消息从发送端到服务端投递可靠（分为以下两个步骤）
#1.1、确认消息已发送到交换机(Exchange) 可以把publisher-confirms: true 替换为 publisher-confirm-type: correlate
publisher-confirm-type: correlated
#1.2、确认消息从交换机中到队列中
publisher-returns: true
综上所述，RabbitMQ通过持久化、确认、事务和重试等机制来保证消息的可靠性，从而解决消息丢失的问题。下面按照上述4个机制进行代码实践。
二、方案实践
1、在生产者服务相关配置
1.1 通过自定义RabbitConfig设置消息投递失败的策略为返回到客户端，处理返回的消息（请注意!如果你使用了延迟队列插件，那么一定会调用该callback方法，因为数据并没有提交上去）。
@Slf4j
@Configuration
public class RabbitConfig {
@Bean
public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory);
//设置消息投递失败的策略，有两种策略：自动删除或返回到客户端。
//我们既然要做可靠性，当然是设置为返回到客户端(true是返回客户端，false是自动删除)
rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) {
log.info("ConfirmCallback 关联数据：{},投递成功,确认情况：{}", correlationData, ack);
} else {
log.info("ConfirmCallback 关联数据：{},投递失败,确认情况：{}，原因：{}", correlationData, ack, cause);
}
}
});
rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
// 请注意!如果你使用了延迟队列插件，那么一定会调用该callback方法，因为数据并没有提交上去，
// 而是提交在交换器中，过期时间到了才提交上去，并非是bug！你可以用if进行判断交换机名称来捕捉该报错
/*if (exchange.equals("你声明的延迟队列的交换机")) {
return;
}*/
log.info("ReturnsCallback 消息被退回：{},回应码：{},回应信息：{},交换机：{},路由键：{}"
, returnedMessage.getMessage(), returnedMessage.getReplyCode()
, returnedMessage.getReplyText(), returnedMessage.getExchange()
, returnedMessage.getRoutingKey());
}
});
return rabbitTemplate;
}
}
DirectRabbitConfig直连交换机配置
@Slf4j
@Configuration
public class DirectRabbitConfig {
private static final String QUEUE = "TestDirectQueue";
private static final String EXCHANGE = "TestDirectExchange";
private static final String ROUTING_KEY = "TestDirectRouting";
/**
* 创建一个名为TestDirectQueue的队列
*
* @return
*/
@Bean
public Queue testDirectQueue() {
// durable:是否持久化,默认为true,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
// autoDelete:是否自动删除，有消息者订阅本队列，然后所有消费者都解除订阅此队列，会自动删除。
// arguments：队列携带的参数，比如设置队列的死信队列，消息的过期时间等等。
return new Queue(QUEUE);
}
/**
* 创建一个名为TestDirectExchange的Direct类型的交换机
*
* @return
*/
@Bean
public DirectExchange testDirectExchange() {
// durable:是否持久化,true,持久化交换机。
// autoDelete:是否自动删除，交换机先有队列或者其他交换机绑定的时候，然后当该交换机没有队列或其他交换机绑定的时候，会自动删除。
// arguments：交换机设置的参数，比如设置交换机的备用交换机（Alternate Exchange），当消息不能被路由到该交换机绑定的队列上时，会自动路由到备用交换机
return new DirectExchange(EXCHANGE);
}
/**
* 绑定交换机和队列
*
* @return
*/
@Bean
public Binding bindingDirect() {
//bind队列to交换机中with路由key（routing key）
return BindingBuilder.bind(testDirectQueue()).to(testDirectExchange()).with(ROUTING_KEY);
}
}
2、在消费者服务相关配置
2.1 配置DirectConsumer
/**
* 直连交换机消息
*
* @author
* @DATE 2025/6/2
**/
@RabbitListener(queues = "TestDirectQueue")
@Component
@Slf4j
public class DirectConsumer {
@RabbitListener(queues = "TestDirectQueue")
@Component
@Slf4j
public class DirectConsumer {
/*@RabbitHandler
public void process(Object data, Channel channel, Message message) throws IOException {
log.info("消费者接受到的消息是：{},消息体为：{}", data, message);
//由于配置设置了手动应答，所以这里要进行一个手动应答。注意：如果设置了自动应答，这里又进行手动应答，会出现double ack，那么程序会报错。
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}*/
@RabbitHandler
public void process(Object message, Channel channel,@Headers Map<String,Object> map) {
System.out.println(message);
//这里只是模拟业务，其实还有很多可能，比如验证用户的银行卡号已经过期等等，都可以发出nack
if (map.get("error")!= null){
System.out.println("错误的消息");
try {
channel.basicNack((Long)map.get(AmqpHeaders.DELIVERY_TAG),false,true); //否认消息
return;
} catch (IOException e) {
e.printStackTrace();
}
}
try {
System.out.println("业务在这里执行！");
channel.basicAck((Long)map.get(AmqpHeaders.DELIVERY_TAG),false); //确认消息
} catch (IOException e) {
//实际业务场景，可能需要将上面的channel.basicNack，放到异常里面进行重试！
e.printStackTrace();
}
}
}
三、测试验证
1、依次启动RabbitMQ、producer(建议先清空队列里面旧的测试消息再启动consumer)和consumer
首先调用接口，生成交换机和队列，否则，在启动consumer时候会报找不到交换机和队列错误。
然后在RabbitMQ界面手动清空队列消息，防止干扰本次实验。
2、在producer中调用接口，发送消息
3、观察控制台打印日志
生产者控制台
消费者控制台
可以看到由于接口中第1，4，5条消息会正常发送，所以在consumer已经进行了正常消费，并且针对第5条进行了业务重试。
由于第2，3条分别由于交换机错误或者队列错误，导致消息发送失败。
四、项目结构及源码
1、项目结构
2、源码下载
RabbitMQ，欢迎Star
