基于SpringBoot解决RabbitMQ消息丢失问题

一、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.1port: 5672username: guestpassword: guestvirtual-host: /#1、确保消息从发送端到服务端投递可靠（分为以下两个步骤）#1.1、确认消息已发送到交换机(Exchange) 可以把publisher-confirms: true 替换为  publisher-confirm-type: correlatepublisher-confirm-type: correlated#1.2、确认消息从交换机中到队列中publisher-returns: true

综上所述，RabbitMQ通过持久化、确认、事务和重试等机制来保证消息的可靠性，从而解决消息丢失的问题。下面按照上述4个机制进行代码实践。

二、方案实践

1、在生产者服务相关配置

1.1 通过自定义RabbitConfig设置消息投递失败的策略为返回到客户端，处理返回的消息（请注意!如果你使用了延迟队列插件，那么一定会调用该callback方法，因为数据并没有提交上去）。

@Slf4j
@Configuration
public class RabbitConfig {@Beanpublic RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory);//设置消息投递失败的策略，有两种策略：自动删除或返回到客户端。//我们既然要做可靠性，当然是设置为返回到客户端(true是返回客户端，false是自动删除)rabbitTemplate.setMandatory(true);rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic 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() {@Overridepublic 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*/@Beanpublic Queue testDirectQueue() {// durable:是否持久化,默认为true,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除，有消息者订阅本队列，然后所有消费者都解除订阅此队列，会自动删除。// arguments：队列携带的参数，比如设置队列的死信队列，消息的过期时间等等。return new Queue(QUEUE);}/*** 创建一个名为TestDirectExchange的Direct类型的交换机** @return*/@Beanpublic DirectExchange testDirectExchange() {// durable:是否持久化,true,持久化交换机。// autoDelete:是否自动删除，交换机先有队列或者其他交换机绑定的时候，然后当该交换机没有队列或其他交换机绑定的时候，会自动删除。// arguments：交换机设置的参数，比如设置交换机的备用交换机（Alternate Exchange），当消息不能被路由到该交换机绑定的队列上时，会自动路由到备用交换机return new DirectExchange(EXCHANGE);}/*** 绑定交换机和队列** @return*/@Beanpublic 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 {/*@RabbitHandlerpublic void process(Object data, Channel channel, Message message) throws IOException {log.info("消费者接受到的消息是：{},消息体为：{}", data, message);//由于配置设置了手动应答，所以这里要进行一个手动应答。注意：如果设置了自动应答，这里又进行手动应答，会出现double ack，那么程序会报错。channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}*/@RabbitHandlerpublic void process(Object message, Channel channel,@Headers Map<String,Object> map) {System.out.println(message);//这里只是模拟业务，其实还有很多可能，比如验证用户的银行卡号已经过期等等，都可以发出nackif (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