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

【RabbitMQ】RabbitMQ保证消息不丢失的N种策略的思想总结

news 来源：原创 2025/5/24 11:19:01

在这里插入图片描述

文章目录

- 生产者端（消息发布端）保证机制
- RabbitMQ服务器端保证机制
- 消费者端（消息接收端）保证机制
- 除了MQ自带的机制，还能做的操作
- 持久化的原理
- ACK思想

更多相关内容可查看

消息从发送，到消费者接收，会经理多个过程 , 其中的每一步都可能导致消息丢失

生产者端（消息发布端）保证机制

消息确认机制（Confirm）
- RabbitMQ提供了消息确认模式。生产者将消息发送给RabbitMQ后，RabbitMQ会返回一个确认回执给生产者。生产者可以通过设置channel.confirmSelect()开启确认模式，然后使用异步或者同步的方式等待确认。
- 在异步确认方式下，生产者可以通过注册一个回调函数来处理确认消息。例如：
```
channel.addConfirmListener(new ConfirmListener() {@Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {// 消息成功被接收，记录日志或者更新状态}@Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {// 消息未被接收，进行重发或者记录错误}
});
```
- 同步确认方式则是通过channel.waitForConfirms()方法来等待确认，这种方式会阻塞生产者线程，直到收到所有已发送消息的确认回执或者超时。
消息持久化
- 生产者在发送消息时，可以将消息设置为持久化。在AMQP协议中，消息有两个属性用于控制消息的持久化：deliveryMode。当deliveryMode = 2时，表示消息是持久化消息。
- 例如，在Java客户端中，可以这样设置：
```
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().deliveryMode(2).build();
channel.basicPublish("", "queue_name", properties, message.getBytes());
```
- 这样可以确保消息在RabbitMQ服务器端持久化存储，即使RabbitMQ服务器意外重启，消息也不会丢失。

RabbitMQ服务器端保证机制

队列持久化
- 除了消息持久化，队列也需要进行持久化设置。在定义队列时，需要将队列设置为持久化队列。例如，在RabbitMQ管理界面或者通过代码（以Java为例）进行如下设置：
```
boolean durable = true;
channel.queueDeclare("queue_name", durable, false, false, null);
```
- 这样当RabbitMQ服务器重启时，持久化队列及其内部的持久化消息都会被恢复。
集群与高可用配置
- RabbitMQ可以配置为集群模式。在集群环境下，消息会在多个节点之间进行复制和同步。如果一个节点出现故障，其他节点可以继续提供服务，保证消息不会丢失。
- 例如，可以使用RabbitMQ的镜像队列（Mirrored Queues）功能。镜像队列会在多个节点上保存相同内容的队列，当主节点上的队列发生变化（如收到新消息或者消息被消费）时，变化会被同步到镜像节点。这样，即使主节点出现故障，镜像节点也可以接替工作，确保消息的可用性。

消费者端（消息接收端）保证机制

手动确认机制（Acknowledge）
- 消费者可以通过手动确认机制来确保消息被正确处理后才从队列中移除。在大多数RabbitMQ客户端中，都支持手动确认模式。
- 例如，在Java客户端中，消费者可以通过channel.basicConsume()方法设置autoAck = false来开启手动确认模式。然后在成功处理消息后，通过channel.basicAck()方法来确认消息，告诉RabbitMQ可以将该消息从队列中移除。
- 如果消费者在处理消息过程中出现异常，没有确认消息，RabbitMQ会认为该消息没有被正确处理，会将消息重新放回队列，等待下一次被消费，从而保证消息不会因为消费者端的异常而丢失。
消费者预取计数（Qos - Quality of Service）
- 消费者可以设置预取计数（basicQos），限制一次从队列中获取的消息数量。这样可以防止消费者因为接收过多消息而无法及时处理，导致消息在消费者端丢失。
- 例如，在Java客户端中，可以这样设置预取计数：
```
int prefetchCount = 1;
channel.basicQos(prefetchCount);
```
- 这表示消费者一次最多从队列中获取1条消息，只有在成功处理并确认这条消息后，才会获取下一条消息，确保消费者能够有条不紊地处理消息，减少消息丢失的风险。

除了MQ自带的机制，还能做的操作

通过RabbitMQ本身所提供的机制基本上已经可以保证消息不丢失 , 但是因为一些特殊的原因还是会发送消息丢失问题 , 例如 : 回调丢失 , 系统宕机, 磁盘损坏等 , 这种概率很小 , 但是如果想规避这些问题 , 进一步提高消息发送的成功率, 也可以通过程序自己进行控制

设计一个消息状态表 , 主要包含 : 消息id , 消息内容 , 交换机 , 消息路由key , 发送时间, 签收状态等字段 , 发送方业务执行完毕之后 , 向消息状态表保存一条消息记录, 消息状态为未签收 , 之后再向MQ发送消息 , 消费方接收消息消费完毕之后 , 向发送方发送一条签收消息 , 发送方接收到签收消息之后 , 修改消息状态表中的消息状态为已签收 ! 之后通过定时任务扫描消息状态表中这些未签收的消息 , 重新发送消息, 直到成功为止 , 对于已经完成消费的消息定时清理即可 !

持久化的原理

消息持久化在生产者端的原理
- 消息属性设置：在生产者端，消息持久化主要通过设置消息的属性来实现。在AMQP（高级消息队列协议）中，消息有一个deliveryMode属性。当deliveryMode = 2时，就表示该消息是持久化消息。例如，在Java使用RabbitMQ客户端发送消息时，会构建一个AMQP.BasicProperties对象来设置消息的各种属性，其中就包括deliveryMode。
- 消息存储过程：当生产者发送持久化消息时，RabbitMQ客户端会将消息标记为需要持久化。这个标记信息会随着消息一起发送给RabbitMQ服务器。RabbitMQ服务器收到消息后，会根据这个标记将消息存储到磁盘上的存储介质中，而不是仅仅存储在内存中。具体存储位置和方式取决于RabbitMQ的存储配置，通常会存储在一个消息存储文件（如在默认的基于文件存储的配置下，消息存储在msg_store_persistent文件中）中，这个文件会定期进行刷盘操作，确保消息能够真正持久地保存到磁盘。
消息持久化在RabbitMQ服务器端（队列持久化）的原理
- 队列定义与存储：队列持久化是消息持久化的重要组成部分。在RabbitMQ中，队列是消息存储的容器。当创建一个持久化队列时，RabbitMQ会将队列的定义信息（如队列名称、队列属性、绑定关系等）存储到磁盘上的队列元数据文件中。例如，在Linux系统下，这些元数据可能存储在/var/lib/rabbitmq/mnesia/rabbit@localhost目录下的相关文件中。
- 消息存储与恢复：对于持久化队列中的消息，当消息被发送到该队列时，RabbitMQ会将消息存储到与该队列关联的存储区域。这个存储区域可以是基于文件系统的，也可以是基于数据库（如使用插件支持）的存储方式。在基于文件系统的存储中，消息会按照一定的规则写入到消息存储文件中。当RabbitMQ服务器重启时，它会首先读取队列的元数据文件，恢复队列的定义，然后根据队列的定义和存储的消息文件，将消息重新加载到内存中的队列里，从而保证消息不会丢失，并且队列的结构和消息的顺序等都能恢复到服务器重启前的状态。
消息持久化的整体流程原理总结
- 生产者将带有持久化标记（deliveryMode = 2）的消息发送给RabbitMQ服务器。RabbitMQ服务器收到消息后，先检查消息所属的队列是否为持久化队列。如果是，就将消息存储到磁盘上的消息存储区域，同时将队列的相关信息（包括消息的索引等）存储到磁盘上的队列元数据区域。当消费者从队列中获取消息时，RabbitMQ会从磁盘存储区域读取消息并发送给消费者。如果RabbitMQ服务器因为各种原因（如断电、软件故障等）重启，它会根据磁盘上存储的队列元数据和消息内容，重新构建队列并将消息加载回队列，使得消息可以继续被消费者获取，从而实现了消息的持久化，保证消息在整个生命周期内不会因为服务器故障等原因而丢失。

ACK思想

ACK（Acknowledge）的基本概念
- 在消息队列（MQ）中，ACK是一种确认机制，用于消费者向消息队列告知消息已经被成功接收和处理。当消费者从消息队列中获取到一条消息后，它需要在处理完该消息后发送一个ACK信号给消息队列，以表示消息已经被正确处理。
手动ACK和自动ACK的区别
- 自动ACK：在某些MQ的默认配置中，可能采用自动ACK模式。在这种模式下，消费者一旦接收到消息，消息队列就会立即认为该消息已经被成功处理，然后将消息从队列中移除。这种方式的优点是简单快捷，适用于一些对消息处理可靠性要求不高的场景。例如，在简单的日志收集系统中，如果偶尔丢失一条日志消息对系统影响不大，就可以使用自动ACK。但是，这种方式存在风险，因为如果消费者在接收到消息后，还没来得及处理就出现故障（如程序崩溃、网络问题等），那么消息就会丢失。
- 手动ACK：手动ACK模式则要求消费者在成功处理消息后，通过代码显式地向消息队列发送ACK信号。例如，在RabbitMQ中，使用Java客户端时，消费者可以通过channel.basicAck()方法来发送ACK。只有在收到ACK信号后，消息队列才会将消息从队列中移除。这种方式可以确保消息在真正被处理后才从队列中移除，大大提高了消息处理的可靠性。如果消费者在处理消息过程中出现异常，没有发送ACK，消息队列会认为该消息没有被正确处理，会将消息重新放回队列，等待下一次被消费。
ACK在消息队列中的重要性
- 保证消息不丢失：通过手动ACK机制，消息队列可以确保消息不会因为消费者端的故障而丢失。例如，在一个电商订单处理系统中，订单消息通过消息队列传递给库存管理系统。如果库存管理系统在处理订单消息时出现故障，没有发送ACK，那么订单消息会被重新放回队列，等待库存管理系统恢复后再次处理，这样就保证了订单消息不会丢失，从而保证了整个业务流程的准确性。
- 实现消息的精确消费：ACK机制有助于消息队列精确地控制消息的消费顺序和状态。消息队列可以根据ACK信号来判断哪些消息已经被成功处理，哪些消息还需要重新分配给消费者进行处理。这对于一些对消息处理顺序和完整性要求较高的场景非常重要，比如金融交易系统中的交易消息处理。