当前位置：首页 > news >正文

RabbitMQ消息可靠性问题及解决

news 2025/8/15 20:07:37

说明：在RabbitMQ消息传递过程中，有以下问题：

消息没发到交换机
消息没发到队列
MQ宕机，消息在队列中丢失
消息者接收到消息后，未能正常消费（程序报错），此时消息已在队列中移除

针对以上问题，提供以下解决方案：

消息确认：确认消息是否发送到交换机、队列；
消息持久化：持久化消息，以防MQ宕机造成消息丢失；
消费者消息确认：确认消费者已正确消费消息，才把消息从队列中删除；

在这里插入图片描述

消息确认

可以使用Rabbit MQ提供的publisher confirm机制来避免消息发送到MQ过程丢失。具体实现是，publisher-confirm（发送者确定）、publisher-return（发送者回执），前者判断消息到交换机、后者判断交换机到队列

publisher-confirm（发送者确定）

消息成功投递到交换机，返回ack；
消息未投递到交换机，返回nack；

publisher-return（发送者回执）

消息投递到交换机，但没有到队列，返回ack，即失败原因；

在生产者端添加配置

spring:rabbitmq:# rabbitMQ相关配置host: 118.178.228.175port: 5672username: rootpassword: 123456virtual-host: /# 开启生产者确认，correlated为异步，simple为同步publisher-confirm-type: correlated# 开启publish-return功能，基于callback机制publisher-returns: true# 开启消息路由失败的策略，true是调用returnCallback方法，false是丢弃消息template:mandatory: true

publisher-return（发送者回执）代码

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;/*** 发送者回执实现*/
@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取RabbitTemplate对象RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);// 设置ReturnCallbackrabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {/*** 回执信息* @param message 信息对象* @param replyCode 回执码* @param replyText 回执内容* @param exchange 交换机* @param routingKey 路由键值*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.info("消息发送队列失败=====replyCode{}，replyText{}，exchange{}，routingKey{}，message{}",replyCode,replyText,exchange,routingKey,message);}});}
}

publisher-confirm（发送者确定）代码

    @Testpublic void sendExceptionMessage() {// 路由键值String routingKey = "exception";// 消息String message = "This is a exception message";// 给消息设置一个唯一IDCorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());// 编写confirmCallBack回调函数correlationData.getFuture().addCallback(new SuccessCallback<CorrelationData.Confirm>() {@Overridepublic void onSuccess(CorrelationData.Confirm confirm) {if (confirm.isAck()) {// 消息发送交换机成功log.debug("消息送达至交换机成功");} else {// 消息发送交换机失败，打印消息log.error("消息未能送达至交换机，ID{}，原因{}", correlationData.getId(), confirm.getReason());}}}, new FailureCallback() {// 消息发送交换机异常@Overridepublic void onFailure(Throwable ex) {log.error("消息发送交换机异常，ID:{}，原因{}", correlationData.getId(), ex.getMessage());}});rabbitTemplate.convertAndSend("amq.direct", routingKey, message, correlationData);}

测试，设置一个不存在的routingKey，被发送者确认（publisher-confirm）捕获到；

// 路由键值
String routingKey = "null";

在这里插入图片描述

设置一个不存在的路由，被发送者回执（publisher-return）捕获到；

rabbitTemplate.convertAndSend("null", routingKey, message, correlationData);

在这里插入图片描述

消息持久化

消息持久化，是指把消息保存到磁盘中，在RabbitMQ宕机或者关机时，重启后，消息仍可以保存下来。消息依赖于交换机、队列，因此持久化消息，同时也需要持久化交换机、队列。

创建一个持久化的交换机、队列

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 消息持久化*/
@Configuration
public class DurableConfig {/*** 交换机持久化* @return*/@Beanpublic DirectExchange directExchange(){// 三个参数分别是：交换机名、是否持久化、没有队列与之绑定时是否自动删除return new DirectExchange("durable.direct",true,false);}/*** 队列持久化* @return*/@Beanpublic Queue durableQueue(){return QueueBuilder.durable("durable.queue").build();}/*** 交换机与队列绑定* @return*/@Beanpublic Binding binding(){return BindingBuilder.bind(durableQueue()).to(directExchange()).with("durable");}}

发送一个持久化的消息

    /*** 发送持久化消息*/@Testpublic void sendDurableMessage() {String routingKey = "durable";CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());Message message = MessageBuilder.withBody("This is a durable message".getBytes(StandardCharsets.UTF_8))// 设置该消息未持久化消息.setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();rabbitTemplate.convertAndSend("durable.direct", routingKey, message, correlationData);}

打开RabbitMQ管理平台，可以看到"delivery_mode: 2"，表示该消息是持久化消息

在这里插入图片描述

（源码：MessageDeliveryMode类）
在这里插入图片描述

实际上，交换机、队列默认就是持久化的（durable: true），所以不用特意设置；

在这里插入图片描述

消费者消息确认

介绍

消费者消息确认，是为了确保消费者已经消费了消息，才让MQ把该消息删除；

可通过在消费者的配置文件中增加下面这行配置实现，备选项有以下三个：

none：关闭ack，表示不做处理，消息发给消费者之后就立即被删除；
auto：自动ack，表示由Spring检测代码是否出现异常，出现异常则保留消息，没有异常则删除消息；
manual：手动ack，可根据业务手动编写代码，返回ack；

spring:rabbitmq:listener:simple:# 设置消息确认模式acknowledge-mode: none

测试：none

可编写代码测试，下面是生产者代码，发送消息

    /*** 发送普通消息*/@Testpublic void sendNoneMessage() {String directName = "none.direct";String routingKey = "none";String message = "This is a test message";rabbitTemplate.convertAndSend(directName, routingKey, message);}

消费者代码有问题，未能正常消费消息

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "none.queue"),exchange = @Exchange(name = "none.direct",type = ExchangeTypes.DIRECT),key = {"none"}))public void getNoneMessage(String normalMessage){System.out.println(1/0);System.out.println("normalMessage = " + normalMessage);}

测试结果，程序报错，消息也没能保留下来

在这里插入图片描述

测试：auto

更改设置为：auto，重试

在这里插入图片描述

但是消息未被删除

在这里插入图片描述

这种情况，在实际开发中是不能允许，可以通过更改消费失败的重试机制解决。

消费失败重试机制

方法一：设置retry

因为消息被消费失败，消息会一直循环重试，无限循环，导致mq的消息处理飙升，带来不必要的压力，这种情况可以通过在消费者端添加以下配置，限制失败重试的条件来解决：

spring:rabbitmq:listener:simple:retry:# 开启消费者失败重试enabled: true# 初次失败等待时长为1秒initial-interval: 1000# 失败的等待时长倍数，即后一次等待的时间是前一次等待时间的多少倍multiplier: 1# 最多重试次数max-attempts: 3# true 无状态 false 有状态 如果业务中包含事务 改为falsestateless: true

开启后，控制台可以发现，信息不回一直循环打印，而是打印数条后停止，日志信息中有提示“Retry Policy Exhausted”（重试策略已用尽）

在这里插入图片描述
这种通过配置的方式，并不会重试数次后仍保留消息，而是重试数次仍失败，随即丢弃消息，消息丢失，这在实际开发中也是不能被允许的。

方法二：路由存储消息

因此，可以通过下面这个方法，把消费失败的消息，通过交换机路由到另外的队列中存储起来，等业务代码被修复，再路由回来消费。

在这里插入图片描述

代码如下

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 错误消息队列*/
@Configuration
public class ErrorMessageQueueConfig {/*** 创建一个交换机，用于路由消费失败的消息* @return*/@Beanpublic DirectExchange errorExchange(){return new DirectExchange("error.direct");}/*** 创建一个队列，用于存储消费失败的消息* @return*/@Beanpublic Queue errorQueue(){return new Queue("error.queue");}/*** 绑定* @return*/@Beanpublic Binding errorBinding(){return BindingBuilder.bind(errorQueue()).to(errorExchange()).with("error");}/*** 路由，当消费失败时，把消费失败的消息路由到此队列中，路由key为"error"* @param rabbitTemplate* @return*/@Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate,"error.direct","error");}
}