RabbitMQ--消息丢失--如何保证消息不丢失--原因/解决方案

小灰灰 2022-09-14 00:21 319阅读 0赞

原文网址：[RabbitMQ--消息丢失--如何保证消息不丢失--原因/解决方案\_IT利刃出鞘的博客-CSDN博客][RabbitMQ--_--_--_IT_-CSDN]

# 简介 #

**说明**

对于消息队列（MQ）来说，消息丢失/消息重复/消费顺序/消息堆积是比较常见的问题，都属于消息异常，这几个问题比较重要面试中也会经常问到。

本文针对RabbitMQ，在生产者、MQ服务器、消费者三方分析出现问题的原因以及解决的方案。

**官方文档**

[Consumer Acknowledgements and Publisher Confirms — RabbitMQ][Consumer Acknowledgements and Publisher Confirms _ RabbitMQ]

**消息丢失的场景**

首先明确一条消息的传送流程：生产者->MQ->消费者

所以这三个节点都可能丢失数据：

1.  Producer端
    
    1.  发送消息过程中出现网络问题：producer以为发送成功，但RabbitMQ server没有收到；
2.  RabbitMQ server 端
    
    1.  接收到消息后由于服务器宕机或重启等原因（消息默认存在内存中）导致消息丢失；
3.  Consumer端
    
    1.  Consumer端接收到消息后自动返回ack，但后边处理消息出错，没有完成消息的处理；

# **生产者丢失消息** #

## **消息丢失场景** ##

生产者将数据发送到RabbitMQ的时候，可能因为网络问题导致数据没到达RabbitMQ Server。

## **方案1：发送方确认机制（推荐，最常用）** ##

**其他网址**

《RabbitMQ实战指南》=> 4.8.2 发送方确认机制

**详解**

生产者将信道设置成confirm（确认）模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，RabbitMQ就会发送一个确认（Basic.Ack)给生产者（包含消息的唯一ID)，这就使得生产者知晓消息己经正确到达了目的地了。如果消息和队列是可持久化的，那么确认消息会在消息写入磁盘之后发出。

RabbitMQ回传给生产者的确认消息中的deliveryTag包含了确认消息的序号，此外RabbitMQ也可以设置channel.basicAck方法中的multiple参数，表示不到这个序号之前的所有消息都已经得到了处理。

如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack(Basic.Nack)命令。

事务机制在一条消息发送之后会使发送端阻塞，以等待RabbitMQ的回应，之后才能继续发送下一条消息。相比之下，**发送方确认机制最大的好处在于它是异步的**，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用程序便可以通过回调方法来处理该确认消息。（发送完一个消息之后就可以发送下一个消息）。

**引入新问题**：

问题：如果RabbitMQ服务端正常接收到了，把ack信息发送给生产者，结果这时网断了，怎么办？

解决方案：在内存里维护每个消息id的状态以及其发送的时间，然后启动一个定时线程去检查它，若超过一定时间还没接收到这个消息的回调，那么就重发。此时，消费者就要处理幂等问题（多次接收到同一条消息）。

## **解决方案2：本地消息表+定时任务（不推荐，因为依赖太多）** ##

见：[分布式事务--消息表+MQ\_IT利刃出鞘的博客-CSDN博客][--_MQ_IT_-CSDN]  
=>本地消息表+MQ（无事务）  
=> 实例流程

## **解决方案3：事务（不推荐，因为性能差）** ##

可以选择用RabbitMQ提供的事务功能，在生产者发送数据之前开启RabbitMQ事务（channel.txSelect），然后发送消息，如果消息没有成功被RabbitMQ接收到，那么生产者会收到异常报错，此时就可以回滚事务（channel.txRollback），然后重试发送消息；如果收到了消息，那么可以提交事务（channel.txCommit）。但是问题是，开始RabbitMQ事务机制，基本上吞吐量会下来，因为太耗性能。

# **Broker丢失消息** #

## **消息丢失场景** ##

**RabbitMQ服务端**接收到消息后由于服务器宕机或重启等原因（消息默认存在内存中）导致消息丢失；

## **解决方案：开启Broker持久化** ##

**结论**

为防止RabbitMQ服务端弄丢数据，要开启RabbitMQ的持久化，就是消息写入之后会持久化到磁盘，哪怕是RabbitMQ自己挂了，恢复之后会自动读取之前存储的数据，一般数据不会丢。

**详解**

设置持久化有两个步骤，第一个是创建queue的时候将其设置为持久化的，这样就可以保证RabbitMQ持久化queue的元数据，但是不会持久化queue里的数据；第二个是发送消息的时候将消息的deliveryMode设置为2，就是将消息设置为持久化的，此时RabbitMQ就会将消息持久化到磁盘上去。必须要同时设置这两个持久化才行，RabbitMQ哪怕是挂了，再次重启，也会从磁盘上重启恢复queue，恢复这个queue里的数据。

极其罕见的是，RabbitMQ还没持久化，自己就挂了，导致数据丢失，但是这个概率较小。当然，也可以解决，解决方法如下：

RabbitMQ开启持久化，生产者开启confirm机制。只有消息被持久化到磁盘之后，才会通知生产者ack了，所以哪怕是在持久化到磁盘之前，RabbitMQ挂了，数据丢了，生产者收不到ack，你也是可以自己重发的。

# **消费者丢失消息** #

## **消息丢失场景** ##

**情景1**：RabbitMQ服务端向消费者发送完消息之后，网络断了，消息并没有到达消费者（RabbitMQ服务端的消息此时已删除）。

**情景2：**Consumer端接收到消息后自动返回ack，但后边处理消息出错，没有完成消息的处理。

## **解决方案** ##

**结论**

设置返回确认的模式为手动，并在处理完消息后手动去提交确认。(做法：消费者在订阅队列时，指定autoAck为false)

见《RabbitMQ实战指南》=> 3.5 消费端的确认与拒绝

**详解**

当设置返回确认的模式为手动（autoAck参数置为false）,对于RabbitMQ服务端而言，队列中的消息分成了两个部分：一部分是等待投递给消费者的消息；一部分是已经投递给消费者，但是还没有收到消费者确认信号的消息。如果RabbitMQ—直没有收到消费者的确认信号，并且消费此消息的消费者己经断开连接，则RabbitMQ会安排该消息重新进入队列，等待投递给下一个消费者，当然也有可能还是原来的那个消费者。

RabbitMQ不会为未确认的消息设置过期时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否已经断开，这么设计的原因是RabbitMQ允许消费者消费一条消息的时间可以很久很久。

# 其他网址 #

[【消息队列】RabbitMQ如何处理消息丢失 - 个人文章 - SegmentFault 思否][RabbitMQ_ - _ - SegmentFault]  
[RabbitMQ防止消息丢失\_大数据\_杨航的专栏-CSDN博客][RabbitMQ_-CSDN]

[RabbitMQ--_--_--_IT_-CSDN]: https://knife.blog.csdn.net/article/details/120558617
[Consumer Acknowledgements and Publisher Confirms _ RabbitMQ]: https://www.rabbitmq.com/confirms.html
[--_MQ_IT_-CSDN]: https://blog.csdn.net/feiying0canglang/article/details/106783654
[RabbitMQ_ - _ - SegmentFault]: https://segmentfault.com/a/1190000019125512
[RabbitMQ_-CSDN]: https://blog.csdn.net/hanghangaidoudou/article/details/76686032