0
0

RabbitMQ：如何保证消息的可靠性？

  • RabbitMQ基础
  • RabbitMQ支持的消息模型
  • SpringBoot集成RabbitMQ

一、发送者的可靠性

消息从发送者发送消息，到消费者处理消息，需要经过的流程是这样的：

消息从生产者到消费者的每一步都有可能导致消息丢失：

  • 发送消息时丢失：- 生产者发送消息时连接 MQ- 失败生产者发送消息到达 MQ 后未找到 Exchange- 生产者发送消息到达 MQ 的 Exchange 后，未找到合适的 Queue- 消息到达 MQ 后，处理消息的进程发生异常。
  • MQ导致丢失：- 消息到达 MQ，保存到队列后，尚未消费就突然宕机。
  • 消费者处理消费时丢失：- 消息接收后尚未处理突然宕机- 消息接收后处理过程中抛出异常

综上 我们要解决消息丢失问题，保证 

MQ

的可靠性，就必须从

3

个方面入手：

  1. 确保生产者一定把消息发送到 MQ
  2. 确保 MQ 不会将消息弄丢
  3. 确保消费者一定要处理消息
1.1、生产者重试机制

第一种情况：生产者发送消息时，出现了网络故障，导致与

MQ

的连接中断。为了解决这个问题，

SpringAMQP

提供消息发送时重试机制。修改 

publisher

模块的配置文件，添加内容如下：

spring:rabbitmq:host: 192.168.137.120
    port:5672virtual-host: /
    username: admin
    password:123456connection-timeout: 1s  # 设置MQ的连接超时时间template:retry:enabled:true# 开启超时重试机制initial-interval: 1000s # 失败后的初始等待时间multiplier:1# 失败后下次等待时长倍数max-attempts:3# 最大重试次数

停止

rabbitMQ

服务

docker stop mq

测试发送一条消息，查看结果是否有重试机制出现

注意：当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过

SpringAMQP

提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的。如果对于业务性能有要求，建议禁用重试机制。如果一定要使用，请合理配置等待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码。

1.2、生产者确认机制

一般情况下，只要生产者与

MQ

之间的网路连接顺畅，基本不会出现发送消息丢失的情况，因此大多数情况下我们无需考虑这种问题。不过，在少数情况下，也会出现消息发送到

MQ

之后丢失的现象，比如：

  • MQ 内部处理消息的进程发生了异常
  • 生产者发送消息到达 MQ 后未找到Exchange
  • 生产者发送消息到达 MQ 的 Exchange 后，未找到合适的Queue，因此无法路由。

针对上述情况，

RabbitMQ

提供了生产者消息确认机制，包括

Publisher Confirm

和

Publisher Return

两种。在开启确认机制的情况下，当生产者发送消息给

MQ

后，

MQ

会根据消息处理的情况返回不同的回执。

总结如下：

  • 当消息投递到 MQ，但是路由失败时，通过 Publisher Return 返回异常信息，同时返回 ACK 的确认信息，代表投递成功。
  • 临时消息投递到了 MQ，并且入队成功，返回 ACK，告知投递成功。
  • 持久消息投递到了MQ，并且入队完成持久化，返回 ACK ，告知投递成功。
  • 其它情况都会返回 NACK，告知投递失败。

其中

ack

和 

nack

属于 

Publisher Confirm

机制，

ack

是投递成功；

nack

是投递失败。而

return

则属于 

Publisher Return

机制。
默认两种机制都是关闭状态，需要通过配置文件来开启。

1.3、生产者确认机制实现
1.3.1、生产者开启确认机制

在 

publisher

模块的 

application.yaml

中添加配置：

spring:rabbitmq:publisher-confirm-type: correlated # 开启publisher confirm 机制，并设置confirm 类型publisher-returns:true# 开启publisher return 机制

这里

publisher-confirm-type

有三种模式可选：

  • none：关闭 confirm 机制。
  • simple： 同步阻塞等待 MQ 的回执。
  • correlated：****MQ异步回调返回回执。

一般选择 correlated 回调机制。

1.3.2、定义
ReturnCallback

每个

RabbitTemplate

只能配置一个

ReturnCallback

，因此我们可以在配置类中统一设置。我们在

publisher

模块定义一个配置类，内容如下：

packagecom.itheima.publisher.config;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.core.ReturnedMessage;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.BeansException;importorg.springframework.context.ApplicationContext;importorg.springframework.context.ApplicationContextAware;importorg.springframework.context.annotation.Configuration;/**
 * @author db
 * @version 1.0
 * @description MqConfig
 * @since 2023/12/7
 */@Configuration@Slf4jpublicclassMqConfigimplementsApplicationContextAware{@OverridepublicvoidsetApplicationContext(ApplicationContext applicationContext)throwsBeansException{RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        rabbitTemplate.setReturnsCallback(newRabbitTemplate.ReturnsCallback(){@OverridepublicvoidreturnedMessage(ReturnedMessage returnedMessage){
                log.debug("触发return callback,");
                log.debug("exchange: {}", returnedMessage.getExchange());
                log.debug("routingKey: {}", returnedMessage.getRoutingKey());
                log.debug("message: {}", returnedMessage.getMessage());
                log.debug("replyCode: {}", returnedMessage.getReplyCode());
                log.debug("replyText: {}", returnedMessage.getReplyText());}});}}
1.3.3、定义
ConfirmCallback

由于每个消息发送时的处理逻辑不一定相同，因此

ConfirmCallback

需要在每次发消息时定义。具体来说，是在调用

RabbitTemplate

中的

convertAndSend

方法时，多传递一个参数

CorrelationData

这里的

CorrelationData

中包含两个核心的东西：

  • id： 消息的唯一标示，MQ 对不同的消息的回执以此做判断，避免混淆。
  • SettableListenableFuture： 回执结果的 Future 对象。

新建一个测试，向系统自带的交换机发送消息，并且添加

ConfirmCallback：

@TestpublicvoidtestPublisherConfirm()throwsInterruptedException{// 创建CorrelationDataCorrelationData correlationData =newCorrelationData();// 给future添加confirmCallback
        correlationData.getFuture().addCallback(newListenableFutureCallback<CorrelationData.Confirm>(){@OverridepublicvoidonFailure(Throwable ex){// 2.1.Future发生异常时的处理逻辑，基本不会触发
                log.error("send message fail", ex);}@OverridepublicvoidonSuccess(CorrelationData.Confirm result){// 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容if(result.isAck()){// result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执
                    log.debug("发送消息成功，收到 ack!");}else{// result.getReason()，String类型，返回nack时的异常描述
                    log.error("发送消息失败，收到 nack, reason : {}", result.getReason());}}});// 发送消息
        rabbitTemplate.convertAndSend("logs","info","hello",correlationData);Thread.sleep(2000);}

测试结果：

改变交换机，使用错误的交换机，由于传递的

RoutingKey

是错误的，路由失败后，触发了

return callback，

同时也收到了

ack

。当我们修改为正确的

RoutingKey

以后，就不会触发

return callback

了，只收到

ack

。而如果连交换机都是错误的，则只会收到

nack

。

注意：开启生产者确认比较消耗MQ性能，一般不建议开启。

二、

MQ

的可靠性

在默认情况下，

RabbitMQ

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

  • 一旦MQ宕机，内存中的消息会丢失。
  • 内存空间有限，当消费者故障或处理过慢时，会导致消息积压，引发MQ阻塞。
2.1、数据持久化

为了提升性能，默认情况下

MQ

的数据都是在内存存储的临时数据，重启后就会消失。为了保证数据的可靠性，必须配置数据持久化，包括：

  • 交换机持久化
  • 队列持久化
  • 消息持久化

交换机的持久化：添加交换机时，配置交换机的

Durability

参数：

队列的持久化，添加队列时，配置交换机的

Durability

参数：

队列的持久化，控制台发送消息的时候，可以添加很多参数，而消息的持久化要配置一个

properties：

说明：

在开启持久化机制以后，如果同时还开启了生产者确认，那么

MQ

会在消息持久化以后才发送

ACK

回执，进一步确保消息的可靠性。不过出于性能考虑，为了减少

IO

次数，发送到

MQ

的消息并不是逐条持久化到数据库的，而是每隔一段时间批量持久化。一般间隔在

100

毫秒左右，这就会导致

ACK

有一定的延迟，因此建议生产者确认全部采用异步方式。

2.2、LazyQueue

在默认情况下，

RabbitMQ

会将接收到的信息保存在内存中以降低消息收发的延迟。但在某些特殊情况下，这会导致消息积压，比如：

  • 消费者宕机或出现网络故障
  • 消息发送量激增，超过了消费者处理速度
  • 消费者处理业务发生阻塞

一旦出现消息堆积问题，

RabbitMQ

的内存占用就会越来越高，直到触发内存预警上限。此时

RabbitMQ

会将内存消息刷到磁盘上，这个行为成为

PageOut

，

PageOut

会耗费一段时间，并且会阻塞队列进程。因此在这个过程中

RabbitMQ

不会再处理新的消息，生产者的所有请求都会被阻塞为了解决这个问题，从

RabbitMQ

的3.6.0版本开始，就增加了

Lazy Queues

的模式，也就是惰性队列。惰性队列的特征如下：

  • 接收到消息后直接存入磁盘而非内存
  • 消费者要消费消息时才会从磁盘中读取并加载到内存（也就是懒加载）
  • 支持数百万条的消息存储

控制台配置

Lazy

模式

在添加队列的时候，添加

x-queue-mod=lazy

参数即可设置队列为

Lazy

模式!

代码配置

Lazy

模式

利用

SpringAMQP

声明队列的时候，添加

x-queue-mod=lazy

参数也可设置队列为

Lazy

模式：

@BeanpublicQueuelazyQueue(){returnQueueBuilder.durable("lazy.queue").lazy()// 开启Lazy模式.build();}

也可以基于注解来声明队列并设置为

Lazy

模式：

@RabbitListener(bindings =@QueueBinding(value =@Queue,// 创建临时队列
            key ={"user.#"},//路由key
            exchange =@Exchange(name="topics",type="topic"),// 绑定交换机
            arguments =@Argument(name="x-queue-mod",value ="lazy")))publicvoidlistenLazyQueueMessage(String msg)throwsInterruptedException{System.out.println("消费者2接收到消息：【"+ msg +"】");}

三、消费者的可靠性

3.1、消费者确认机制

为了确认消费者是否成功处理消息，

RabbitMQ

提供了消费者确认机制

（Consumer Acknowledgement）

。即：当消费者处理消息结束后，应该向

RabbitMQ

发送一个回执，告知

RabbitMQ

自己消息处理状态。回执有三种可选值：

  • ack： 成功处理消息，RabbitMQ 从队列中删除该消息。
  • nack： 消息处理失败，RabbitMQ 需要再次投递消息。
  • reject： 消息处理失败并拒绝该消息，RabbitMQ 从队列中删除该消息。

一般

reject

方式用的较少，除非是消息格式有问题，那就是开发问题了。因此大多数情况下我们需要将消息处理的代码通过

try catch

机制捕获，消息处理成功时返回

ack

，处理失败时返回

nack

。由于消息回执的处理代码比较统一，因此

SpringAMQP

帮我们实现了消息确认。并允许我们通过配置文件设置

ACK

处理方式，有三种模式：

  • none：不处理。即消息投递给消费者后立刻 ack，消息会立刻从 MQ 删除。非常不安全，不建议使用。
  • manual：手动模式。需要自己在业务代码中调用 api ，发送 ack 或 reject，存在业务入侵，但更灵活。
  • auto：自动模式。SpringAMQP 利用 AOP 对我们的消息处理逻辑做了环绕增强，当业务正常执行时则自动返回 ack。当业务出现异常时，根据异常判断返回不同结果： - 如果是业务异常，会自动返回 nack；- 如果是消息处理或校验异常，自动返回 reject;

修改

SpringAMQP

的

ACK

处理方式：

spring:rabbitmq:host: 192.168.137.120
    port:5672username: admin
    password:123456virtual-host: /
    connection-timeout:600000listener:simple:acknowledge-mode: none  # 不做处理

修改

consumer

服务的

SpringRabbitListener

类中的方法，模拟一个消息处理的异常：

@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {
    log.info("spring 消费者接收到消息：【" + msg + "】");
    if (true) {
        throw new MessageConversionException("故意的");
    }
    log.info("消息处理完成");
}

测试可以发现：当消息处理发生异常时，消息依然被

RabbitMQ

删除了。确认机制修改为

auto

：

spring:rabbitmq:host: 192.168.137.120
    port:5672username: admin
    password:123456virtual-host: /
    connection-timeout:600000listener:simple:acknowledge-mode: auto  # 自动ack

在异常位置打断点，再次发送消息，程序卡在断点时，可以发现此时消息状态为 

unacked

（未确定状态）放行以后，由于抛出的是消息转换异常，因此 

Spring

会自动返回 

reject

，所以消息依然会被删除：将异常改为 

RuntimeException

类型：

@RabbitListener(queues = "object.queue")
public void listenObjectMessage(Map<String,Object> msg) throws InterruptedException {
    System.out.println("消费者接收到消息：【" + msg + "】");
    throw new RuntimeException("故意的");
}

在异常位置打断点，然后再次发送消息测试，程序卡在断点时，可以发现此时消息状态为 

unacked

（未确定状态）：放行以后，由于抛出的是业务异常，所以 

Spring

返回 

ack

，最终消息恢复至 

Ready

状态，并且没有被 

RabbitMQ

删除：当我们把配置改为 

auto

时，消息处理失败后，会回到 

RabbitMQ

，并重新投递到消费者。

3.2、失败重试机制

当消费者出现异常后，消息会不断

requeue

（重入队）到队列，再重新发送给消费者。如果消费者再次执行依然出错，消息会再次

requeue

到队列，再次投递，直到消息处理成功为止。极端情况就是消费者一直无法执行成功，那么消息

requeue

就会无限循环，导致

MQ

的消息处理飙升，带来不必要的压力

失败重试机制：在消费者出现异常时利用本地重试，而不是无限制的

requeue

到

MQ

队列。

修改

consumer

服务的

application.yml

文件，添加内容：

spring:rabbitmq:host: 192.168.137.120
    port:5672username: admin
    password:123456virtual-host: /
    connection-timeout:600000listener:simple:acknowledge-mode: auto  # 不做处理retry:enabled:true# 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier:1# 失败的等待时长倍数，下次等待时长 = multiplier * last-intervalmax-attempts:3# 最大重试次数stateless:true# true无状态；false有状态。如果业务中包含事务，这里改为false

重启 

consumer

服务，重复之前的测试。可以发现：

  • 消费者在失败后消息没有重新回到 MQ 无限重新投递，而是在本地重试了3次
  • 本地重试3次以后，抛出了 AmqpRejectAndDontRequeueException 异常。查看 RabbitMQ 控制台，发现消息被删除了，说明最后 SpringAMQP 返回的是 reject

结论：

  1. 开启本地重试时，消息处理过程中抛出异常，不会 requeue 到队列，而是在消费者本地重试。
  2. 重试达到最大次数后，Spring 会返回 reject，消息会被丢弃。
3.3、失败处理策略

在之前的测试中，本地测试达到最大重试次数后，消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下，显然不太合适了。因此 

Spring

允许我们自定义重试次数耗尽后的消息处理策略，这个策略是由

MessageRecovery

接口来定义的，它有3个不同实现：

  • RejectAndDontRequeueRecoverer： 重试耗尽后，直接 reject，丢弃消息。默认就是这种方式。
  • ImmediateRequeueMessageRecoverer： 重试耗尽后，返回nack，消息重新入队。
  • RepublishMessageRecoverer： 重试耗尽后，将失败消息投递到指定的交换机。

比较优雅的一种处理方案是 RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工集中处理。

1）在 

consumer

服务中定义处理失败消息的交换机和队列。

2）定义一个

RepublishMessageRecoverer

，关联队列和交换机。

package com.itheima.consumer.config;

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;

@Configuration
@ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled", havingValue = "true")
public class ErrorMessageConfig {
    @Bean
    public DirectExchange errorMessageExchange(){
        return new DirectExchange("error.direct");
    }
    @Bean
    public Queue errorQueue(){
        return new Queue("error.queue", true);
    }
    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
        return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
    }

    @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }
}

消费者如何保证消息一定被消费？

开启消费者确认机制为

auto

，由

Spring

确认消息处理成功后返回

ack

，异常时返回

nack

开启消费者失败重试机制，并设置

MessageRecoverer

多次重试失败后将消息投递给异常交换机，交由人工处理。

代码地址：

https://gitee.com/duan138/practice-code/tree/master/mq-demo

四、总结

要想保证消息的可靠性，就要先知道什么时候消息会丢失？会有三方面分别是：消息到

MQ

的过程丢失、

MQ

自己丢失、

MQ

到消费过程中丢失。文章中提到从三方面去保证消息的可靠性。分别是生产者到 

RabbitMQ

使用事务机制或者

Confirm

机制；

RabbitMQ

自身持久化、集群等；

RabbitMQ

到消费者采用

basicAck

机制、死信队列、消息补偿机制。

参考：https://www.bilibili.com/video/BV1mN4y1Z7t9

改变你能改变的，接受你不能改变的，关注公众号：程序员康康，一起成长，共同进步。

标签： rabbitmq java
本文转载自: https://blog.csdn.net/abc_138/article/details/140647295
版权归原作者 cxykk1217 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ：如何保证消息的可靠性？”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

【Python】【持续项目】Python-安全项目搜集

泷羽sec学习day5-安全见闻6

随身 WiFi 通过 USB 连接路由器共享网络 扩展网络覆盖范围

如何将 Kubernetes 中的两个 Nginx Ingress 合并成一个：操作步骤与注意事项

单元测试使用JUnit 5

【大数据学习 | HBASE高级】hive操作hbase

MySQL Workbench 8.0 CE 中文汉化教程

文章导航