说明:在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.175
port: 5672
username: root
password: 123456
virtual-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 {
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
// 获取RabbitTemplate对象
RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
// 设置ReturnCallback
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
/**
* 回执信息
* @param message 信息对象
* @param replyCode 回执码
* @param replyText 回执内容
* @param exchange 交换机
* @param routingKey 路由键值
*/
@Override
public 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(发送者确定)代码
@Test
public void sendExceptionMessage() {
// 路由键值
String routingKey = "exception";
// 消息
String message = "This is a exception message";
// 给消息设置一个唯一ID
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
// 编写confirmCallBack回调函数
correlationData.getFuture().addCallback(new SuccessCallback<CorrelationData.Confirm>() {
@Override
public void onSuccess(CorrelationData.Confirm confirm) {
if (confirm.isAck()) {
// 消息发送交换机成功
log.debug("消息送达至交换机成功");
} else {
// 消息发送交换机失败,打印消息
log.error("消息未能送达至交换机,ID{},原因{}", correlationData.getId(), confirm.getReason());
}
}
}, new FailureCallback() {
// 消息发送交换机异常
@Override
public 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
*/
@Bean
public DirectExchange directExchange(){
// 三个参数分别是:交换机名、是否持久化、没有队列与之绑定时是否自动删除
return new DirectExchange("durable.direct",true,false);
}
/**
* 队列持久化
* @return
*/
@Bean
public Queue durableQueue(){
return QueueBuilder.durable("durable.queue").build();
}
/**
* 交换机与队列绑定
* @return
*/
@Bean
public Binding binding(){
return BindingBuilder.bind(durableQueue()).to(directExchange()).with("durable");
}
}
发送一个持久化的消息
/**
* 发送持久化消息
*/
@Test
public 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
可编写代码测试,下面是生产者代码,发送消息
/**
* 发送普通消息
*/
@Test
public 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 有状态 如果业务中包含事务 改为false
stateless: 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
*/
@Bean
public DirectExchange errorExchange(){
return new DirectExchange("error.direct");
}
/**
* 创建一个队列,用于存储消费失败的消息
* @return
*/
@Bean
public Queue errorQueue(){
return new Queue("error.queue");
}
/**
* 绑定
* @return
*/
@Bean
public Binding errorBinding(){
return BindingBuilder.bind(errorQueue()).to(errorExchange()).with("error");
}
/**
* 路由,当消费失败时,把消费失败的消息路由到此队列中,路由key为"error"
* @param rabbitTemplate
* @return
*/
@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
return new RepublishMessageRecoverer(rabbitTemplate,"error.direct","error");
}
}
可以看到,消息消费失败后并没有被丢失,而是路由到错误队列中存储了起来。因为错误队列没有设置RabbitListener,所以可以存储消息,等带代码问题被排查出来后,可以再针对该队列设置监听方法,消费这部分错误的消息。
另外,值得一提的是,消费者这边的控制台会报一个警告,提示路由密钥错误。我们可以理解,在RabbitMQ底层,会把消费失败了的消息,统一路由到一个地方去,而我们这种手动把消费失败的消息路由到自定义的队列中的方式,打破了这种“默认的规则”,所以报了一个这样的警告。这种警告是在可控范围内的。
文章来源:https://www.toymoban.com/news/detail-601451.html
总结
RabbitMQ发送消息,为了确保消息的可靠性,保证消息能被交换机、队列收到,消息能被正常消费,而不会因消费失败而丢失,提供了对应的一系列方法,并且最后还提供了两种消费失败重试方法,优化了消费过程,非常Nice。文章来源地址https://www.toymoban.com/news/detail-601451.html
到了这里,关于RabbitMQ消息可靠性问题及解决的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!