当前位置:   article > 正文

RabbitMQ(四)之消息可靠性_mq消息可靠性

mq消息可靠性

一、消息可靠性分析

以支付服务为案例进行分析

问题一、在扣减余额和更新支付状态之后,支付服务会发送消息通知交易服务更新状态,如果在支付服务与MQ之间出现了网络故障,消息无法到达MQ,交易服务无法收到消息。即发送者弄丢了消息。

问题二、消息投递到了MQ,但在MQ将消息传递给交易服务之前,MQ挂掉了,消息也会丢失。即MQ弄丢了消息。

问题三、消息投递到了交易服务,但在处理的过程中抛出异常,也会丢失消息。即消费者弄丢了消息。

消息的可靠性:消息被发送后,至少要被消费一次。

二、发送者的可靠性 

发送者的可靠性问题通过两种方式解决:生产者重连生产者确认

2.1、生产者重连

有的时候由于网络波动,可能会出现客户端连接MQ失败的情况。通过配置我们可以开启连接失败后的重连机制。

注意:当网络不稳定的时候,利用重试机制可以有效提高消息发送的成功率。不过Spring AMQP提供的重试机制是阻塞式的重试,也就是说多次重试等待的过程中,当前线程是被阻塞的,后续业务代码不会执行,会影响业务性能。如果对于业务性能有要求,建议禁用重试机制。如果一定要使用,请合理配置等待时长和重试次数,当然也可以考虑使用异步线程来执行发送消息的代码。 

  1. spring:
  2. rabbitmq:
  3. connection-timeout: 1s #设置MQ的连接超时时间
  4. template:
  5. retry:
  6. enabled: true #开启超时重试机制
  7. initial-interval: 1000ms #失败后的初始等待时间
  8. multiplier: 1 # 失败后下次的等待时长倍数,下次等待时长 = initial-interval * multiplier
  9. max-attempts: 3 #最大重试次数

 2.2、生产者确认

 RabbitMQ提供了Publisher Confirm和Publisher Return两种确认机制。开启确机制认后,在MQ成功收到消息后会返回确认消息给生产者。返回的结果有以下几种情况:

1、消息投递到了交换机,但是路由失败(routingKey不对,找不到对应的MQ)。此时会通过Publisher Return返回路由异常原因,然后返回ACK,告知投递成功。

2、临时消息投递到了MQ,并且入队成功,返回ACK,告知投递成功。

3、持久消息投递到了MQ,并且入队完成持久化(保存到磁盘),返回ACK ,告知投递成功。

4、其它情况都会返回NACK,告知投递失败。

 

生产者确认需要额外的网络和系统资源开销,尽量不要使用。

如果一定要使用,无需开启Publisher-Return机制,因为一般路由失败是自己业务问题。
对于nack消息可以有限次数重试,依然失败则记录异常消息。

 2.3、生产者确认的代码实现

 第一步、添加yml配置

  1. spring:
  2. rabbitmq:
  3. #开启publisher confirm机制,并设置confirm类型:
  4. #1、none:关闭confirm机制;2、simple:同步阻塞等待MQ的回执消息;3、correlated:MQ异步回调方式返回回执消息
  5. publisher-confirm-type: correlated
  6. publisher-returns: true # 开后publisher return机制

第二步、配置Publisher-Return机制,每个RabbitTemplate只能配置一个ReturnCallback,因此需要在项目启动过程中配置

  1. @Slf4j
  2. @Configuration
  3. public class MqConfirmConfig implements ApplicationContextAware {
  4. @Override
  5. public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
  6. RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
  7. //配置回调
  8. rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
  9. @Override
  10. public void returnedMessage(ReturnedMessage returnedMessage) {
  11. log.debug("收到消息的return callback, exchange:{},key:{},msg:{},code:{},text:{}",
  12. returnedMessage.getExchange(),returnedMessage.getRoutingKey(),returnedMessage.getMessage(),
  13. returnedMessage.getReplyCode(),returnedMessage.getReplyText());
  14. }
  15. });
  16. }
  17. }

第三步、Publisher Confirm机制,在业务代码中主动获取异步结果

  1. @Test
  2. void testConfirmCallback() throws InterruptedException {
  3. //1、创建cd
  4. CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
  5. //2、添加ConfirmCallback
  6. cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
  7. @Override
  8. public void onFailure(Throwable ex) {
  9. log.error("消息回调失败",ex);
  10. }
  11. @Override
  12. public void onSuccess(CorrelationData.Confirm result) {
  13. log.debug("收到confirm callback 回执");
  14. if (result.isAck()){
  15. //消息发送成功
  16. log.debug("消息发送成功,收到ack");
  17. }else {
  18. //消息发送失败
  19. log.error("消息发送失败,收到nack,原因:" ,result.getReason());
  20. //重发消息逻辑代码
  21. }
  22. }
  23. });
  24. rabbitTemplate.convertAndSend("hmall.direct","red","hello",cd);
  25. Thread.sleep(2000);
  26. }

三、MQ的可靠性

在默认情况下,RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。这样会导致两个问题:

1、一旦MQ宕机,内存中的消息会丢失。

2、内存空间有限,当消费者故障或处理过慢时,会导致消息积压,引发MQ阻塞。

 解决方法:数据持久化(3.6版本之前);Lazy Queue(3.6版本之后)。

3.1、数据持久化

RabbitMQ实现数据持久化包括3个方面:

交换机持久化(Durable属性,Spring默认设置为Durable);

队列持久化(Durable属性,Spring默认设置为Durable);

消息持久化(发送消息时设置delivery_mode=2persisent,Spring发送的消息默认是持久化的);

  1. @Test
  2. void testPageOut(){
  3. Message message = MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF_8))
  4. .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();
  5. for (int i = 0; i < 1000000; i++) {
  6. rabbitTemplate.convertAndSend("simple.queue",message);
  7. }
  8. }

发送消息的情况

非持久化,会出现paged out,会阻塞IO,性能下降

持久化消息,不会出现阻塞情况,性能下降的比较小

 

 3.2、Lazy Queue

从RabbitMQ的3.6.0版本开始,就增加了Lazy Queue的概念,也就是惰性队列。

惰性队列的特征如下:
1、接收到消息后直接存入磁盘而非内存(内存中只保留最近的消息,默认2048条)。

2、消费者要消费消息时才会从磁盘中读取并加载到内存。
3、支持数百万条的消息存储。
4、在3.12版本后,所有队列都是Lazy Queue模式,无法更改 。

(3.12之前)需要设置一个队列为惰性队列,只需要在声明队列时,指定x-queue-mode属性为lazy即可:

  1. //声明Bean
  2. @Bean
  3. public Queue lazyQueue(){
  4. return QueueBuilder.durable("lazy.queue").lazy().build(); //创建惰性队列
  5. }
  6. //基于注解创建
  7. @RabbitListener(queuesToDeclare = @Queue(
  8. name = "lazy.queue",
  9. durable = "true",
  10. arguments = @Argument(name = "x-queue-mode",value = "lazy")
  11. ))
  12. public void listenLazyQueue(String msg){
  13. log.info("接收到lazy.queue的消息:{}",msg);
  14. }

发送非持久化消息,不会发生阻塞IO,消息直接到Paged Out,不用经过Memery到Paged Out的过程。

RabbitMQ如何保证消息的可靠性:
首先通过配置可以让交换机、队列、以及发送的消息都持久化。这样队列中的消息会持久化到磁盘,MQ重启消息依然存在。
RabbitMQ在3.6版本引入了LazyQueue,并且在3.12版本后会称为队列的默认模式,LazyQueue会将所有消息都持久化。
开启持久化和生产者确认时,RabbitMQ只有在消息持久化完成后才会给生产者返回ACK回执。

四、消费者的可靠性 

 保证消费者的可靠性主要有三种手段:消费者确认机制;消费失败处理;业务幂等性

4.1、消费者确认机制

为了确认消费者是否成功处理消息,RabbitMQ提供了消费者确认机制(Consumer Acknowledgement)。当消费者处理消息结束后,应该向RabbitMQ发送一个回执,告知RabbitMQ自己消息处理状态。

回执有三种可选值:
ack:成功处理消息,RabbitMQ从队列中删除该消息。
nack:消息处理失败,RabbitMQ需要再次投递消息  。
reject:消息处理失败并拒绝该消息,RabbitMQ从队列中删除该消息(一般是消息的参数不正确) 。

Spring  AMQP已经实现了消息确认功能。并允许我们通过配置文件选择ACK处理方式,有三种方式

none:不处理。即消息投递给消费者后立刻ack,消息会立刻从MQ删除。非常不安全,不建议使用

manual:手动模式。需要自己在业务代码中调用api,发送ack或reject,存在业务入侵,但更灵活

auto:自动模式。Spring AMQP利用AOP对我们的消息处理逻辑做了环绕增强,当业务正常执行时则自动返回ack。当业务出现异常时,根据异常判断返回不同结果:
如果是业务异常,会自动返回nack
如果是消息处理或校验异常,自动返回reject

第一步、开启消费者确认其机制

  1. spring:
  2. rabbitmq:
  3. listener:
  4. simple:
  5. prefetch: 1 #每次只能获取一条消息,处理完才能获取下一条消息
  6. acknowledge-mode: auto #none:关闭ack;manual:手动ack;auto:自动ack

 第二步、消费者业务模拟异常

  1. @RabbitListener(queues = "simple.queue")
  2. public void listSimpleQueue(String msg) {
  3. System.out.println("消费者收到了simple.queue的消息:【" + msg + "】");
  4. throw new RuntimeException("测试异常");
  5. //throw new MessageConversionException("消息转换异常");
  6. }

第三步、发送消息

  1. @Test
  2. void testSendMessage2Queue() {
  3. //队列名称
  4. String queueName = "simple.queue";
  5. //消息
  6. String msg = "hello, amqp!";
  7. //发送消息
  8. rabbitTemplate.convertAndSend(queueName, msg);
  9. }

 auto模式下,消息未被处理会保留消息,尝试重新投递给消费者。像模拟的异常情况下,消息会一直被投递。

4.2、消息失败处理策略

失败重试机制

当消费者出现异常后,消息会不断requeue(重新入队)到队列,再重新发送给消费者,然后再次异常,再次requeue无限循环,导致mg的消息处理飙升,带来不必要的压力。

我们可以利用spring的retry机制,在消费者出现异常时利用本地重试,而不是无限制的requeue到MQ队列

 

  1. spring:
  2. rabbitmq:
  3. listener:
  4. simple:
  5. prefetch: 1 #每次只能获取一条消息,处理完才能获取下一条消息
  6. acknowledge-mode: auto #none:关闭ack;manual:手动ack;auto:自动ack
  7. retry:
  8. enabled: true #开启消费者失败重试
  9. initial-interval: 1000ms #初始的失败等待时长为1
  10. multiplier: 1 #下次失败的等待时长倍数,下次等待时长 = multiplier * last-interval
  11. max-attempts: 3 #最大重试次数
  12. stateless: true #true无状态;false有状态。如果业务中包含事务,这里改为false

失败消息处理策略

在开启重试模式后,重试次数耗尽,如果消息依然失败,则需要有MessageRecoverer接口来处理,它包含三种不同的实现:
RejectAndDontRequeueRecoverer: 重试耗尽后,直接reject,丢弃消息。默认就是这种方式。

ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队。RepublishMessageRecoverer: 重试耗尽后,将失败消息投递到指定的交换机。

RepublishMessageRecoverer原理 

 

第一步、首先,定义接收失败消息的交换机、队列及其绑定关系。

第二步、然后,定义RepublishMessageRecoverer。

  1. @Slf4j
  2. @Configuration
  3. //@ConditionalOnProperty(prefix = "spring.rabbitmq.listener.simple.retry.enable",havingValue = "true") //开启了消费者失败重试机制才会生效
  4. public class ErrorConfiguration {
  5. @Bean
  6. public DirectExchange errorExchange(){
  7. return new DirectExchange("error.direct");
  8. }
  9. @Bean
  10. public Queue errorQueue(){
  11. return new Queue("error.queue");
  12. }
  13. @Bean
  14. public Binding errorBinding(Queue errorQueue,DirectExchange errorExchange){
  15. return BindingBuilder.bind(errorQueue).to(errorExchange).with("error");
  16. }
  17. @Bean
  18. public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){
  19. log.debug("加载RepublishMessageRecoverer");
  20. return new RepublishMessageRecoverer(rabbitTemplate,"error.direct","error");
  21. }
  22. }

消费者如何保证消息一定被消费?
1、开启消费者确认机制为auto,由spring确认消息处理成功后返回ack,异常时返回nack。如果一直处理异常会一直重试,所以可以
2、开启消费者失败重试机制,并设置MessageRecoverer,多次重试失败后将消息投递到异常交换机,交由人工处理

4.3、业务幂等性

 幂等是一个数学概念,用函数表达式来描述是这样的: f(x)= f(f(X)。在程序开发中,则是指同一个业务,执行一次或多次对业务状态的影响是一致的。

方案一,是给每个消息都设置一个唯一id,利用id区分是否是重复消息
1、每一条消息都生成一个唯一的id,与消息一起投递给消费者。
2、消费者接收到消息后处理自己的业务,业务处理成功后将消息ID保存到数据库。(对业务有侵入)
3、如果下次又收到相同消息,去数据库查询判断是否存在,存在则为重复消息放弃处理。

消息发送方的启动类定义,生成唯一id,和消息一起投递

  1. @Bean
  2. public MessageConverter jacksonMessageConvertor() {
  3. //1.定义消息转换器
  4. Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
  5. //2.配置自动创建消息id,用于识别不同消息,也可以在业务中基于ID判断是否是重复消息
  6. jjmc.setCreateMessageIds(true);
  7. return jjmc;
  8. }

方案二,是结合业务逻辑,基于业务本身做判断。以我们的业务为例:我们要在支付后修改订单状态为已支付,应该在修改订单状态前先查询订单状态,判断状态是否是未支付。只有未支付订单才需要修改,其它状态不做处理。

 方案三、使用Token令牌,生成一个token存储在redis中,请求的时候携带这个token一起请求(Token 最好将其放到 Headers 中),后端需要对这个Token作为 Key在redis中进行校验,如果 Key存在就执行删除命令,然后正常执行后面的业务逻辑。如果不存在对应的 Key 就返回重复执行的错误信息,这样来保证幂等操作。

五、在支付服务和交易服务之间如何保证订单状态的一致性

首先,支付服务会正在用户支付成功以后利用MQ消息通知交易服务完成订单状态同步。
其次,为了保证MQ消息的可靠性,可以采用生产者确认机制、消费者确认、消费者失败重试等策略,确保消息投递和处理的可靠性。同时也开启了MQ的持久化,避免因服务宕机导致消息丢失。
最后,我们还在交易服务更新订单状态时做了业务幂等判断,避免因消息重复消费导致订单状态异常。

六、如果交易服务消息处理失败,有没有什么兜底方案?

可以使用RepublishMessageRecoverer机制,进行人工介入。
我们可以在交易服务设置定时任务,定期查询订单支付状态。这样即便MQ通知失败,还可以利用定时任务作为兜底方案,确保订单支付状态的最终一致性。

本文内容由网友自发贡献,转载请注明出处:https://www.wpsshop.cn/w/正经夜光杯/article/detail/805552
推荐阅读
相关标签
  

闽ICP备14008679号