MQ保证消息的可靠性传输

- 日理万妓 2023-07-24 02:57 13阅读 0赞

**关注微信公众号“****虾米聊吧****”，后续持续放送“****技术架构和资料****”干货！！**

**前景概要：**

我们在使用mq的时候肯定是希望数据不能多也不能少的，不能多即需要考虑mq的重复消费（幂等性）问题。不能少即不能丢失数据。

很明显如果如果你的mq中存放的是非常核心、重要的数据，比如订单数据、积分流水等，

这些数据是必须不能丢的，一旦丢失则会造成数据的一致性问题，可能会给公司带来极大损失。

**丢失数据分析：**

丢失数据一般来说分为2种情况，一种是数据已经发送到了mq，但是mq自身把数据弄丢了，第二种是我们在消费的时候把数据弄丢了，接下来简单从rabbitmq和kafka来分析一下。

![format_png][]

**tips**

根据之前的使用场景来看，一般rabbitmq都会存放一些比较核心业务的数据，理论上是不能弄丢的；

kafka一般用于大数据领域，承载较多的可能是一些日志数据。

**（1）RabbitMq**

**1）生产者丢失数据**

**关键点：rabbitmq开启confirm确认机制后，rabbitmq通过异步回调的形式告诉生产者是否已接收到。**

生产者将数据发送到rabbitmq的时候，可能数据就在半路给丢了，因为网络问题或者机宕机问题等都有可能。

一般来说，要确保发送给rabbitmq的消息不能丢失，可以开启confirm模式，在生产者端设置开启confirm模式之后，每次写入mq的消息都会分配一个唯一的id，如果写入了rabbitmq中，rabbitmq则会回传一个ack确认消息，告诉你说这个消息已经成功写入了。如果rabbitmq没能处理这个消息，会回调你一个nack接口，告诉你这个消息接收失败，你可以重试。而且你可以结合这个机制自己在内存里维护每个消息id的状态，如果超过一定时间还没接收到这个消息的回调，那么你可以重发。

**2）rabbitmq自身丢失了数据**

**关键点：开启持久化机制**

这个就是说rabbitmq自己弄丢了数据，所以我们必须开启rabbitmq的持久化机制，当消息写入之后会持久化到磁盘，即使是rabbitmq自己挂了，恢复之后会自动读取之前存储的数据，一般数据不会丢。除非极其罕见的是，rabbitmq还没持久化，自己就挂了，可能导致少量数据会丢失的，但是这个概率较小。

**设置持久化有两个步骤：**

第一个是创建queue的时候将其设置为持久化的，这样就可以保证rabbitmq持久化queue的元数据，但是不会持久化queue里的数据；

第二个是发送消息的时候将消息的deliveryMode设置为2，就是将消息设置为持久化，此时rabbitmq就会将消息持久化到磁盘上去。必须要同时设置这两个持久化才行，rabbitmq即使挂了，当再次重启的时候，也会从磁盘上恢复queue，恢复这个queue里的数据。

而且持久化可以跟生产者那边的confirm机制配合起来，只有消息被持久化到磁盘之后，才会通知生产者ack了，所以哪怕是在持久化到磁盘之前，rabbitmq挂了，数据丢了，生产者收不到ack，你也是可以自己重发的。

但是即使开启了rabbitmq持久化机制，也有一种可能仍然导致数据丢失，就是这个消息写到了rabbitmq中，但是还没来得及持久化到磁盘上，结果不巧，此时rabbitmq挂了，就会导致内存里的一点点数据会丢失。

**3）消费端丢失了数据**

**关键点：关闭自动ack，开启手动ack**

这个很容易理解，主要是因为消费者那一端在消费的时候，刚消费到数据还没完成处理过程，结果消费者服务的进程挂了，但是rabbitmq会认为你已经消费了，此时这数据就丢了。

这个时候就需要采用rabbitmq提供的ack机制，简单来说，就是你需要关闭rabbitmq自动ack，可以通过一个api来调用就行，然后每次你自己代码里确保处理完成的时候，然后程序里再执行ack。这样的话，ack的主动权就掌握在我们自己手里了，即使数据还没处理完，也不会执行ack，这样的话rabbitmq就不会认为这条消息已经被消费了，所以消息是不会丢的。

**（2）Kafka**

**1） 消费端弄丢了数据**

**关键点：关闭自动提交offset，手动提交offset**

唯一可能导致消费者弄丢数据的情况，就是消费者消费到了这个消息，然后消费者那边自动提交了offset，让kafka以为你已经消费了这个消息，但其实你还没处理完程序的逻辑就挂了，此时这条消息就丢了。

我们都知道kafka默认会自动提交offset，因此只要关闭自动提交offset，在处理完之后自己手动提交offset，就可以保证数据不会丢。但是此时可能会出现另外一个问题就是重复消费，比如说你刚处理完代码逻辑，还没来得及提交offset，结果程序就挂了，此时肯定会重复消费一次，因此这个需要自身在消费者那端保证幂等性。

**2）kafka自己弄丢了数据**

较常见的一个场景就是kafka某个broker宕机，然后重新选举partition的leader时候，如果此时其他的follower分片刚好还有些数据没有同步，结果这个时候leader挂了，然后选举某个follower成leader之后，它就会少了一部分数据。

**所以此时一般是要求起码设置如下4个参数：**

给这个topic设置replication.factor参数：这个值必须大于1，要求每个partition必须有至少2个副本；

在kafka服务端设置min.insync.replicas参数：这个值必须大于1，这个是要求一个leader至少感知到有至少一个follower还跟自己保持联系，没掉队，这样才能确保leader挂了还有一个follower吧；

在producer端设置acks=all：这个是要求每条数据，必须是写入所有replica副本之后，才能认为是写成功了；

在producer端设置retries=MAX（可以设置一个比较大的值，重试次数）：这个是要求一旦写入失败，进行重试；

**3）生产者会不会弄丢数据**

如果按照上述的思路设置了ack=all，一定不会丢，要求是，你的leader接收到消息，所有的follower都同步到了消息之后，才认为本次写成功了。如果没满足这个条件，生产者会自动不断的重试，重试无限次。

<table style="width:453px;"> 
 <tbody> 
 <tr> 
 <td style="width:450px;"> 关注微信公众号“虾米聊吧”，后续持续放送“技术架构和资料”干货！！ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;<img alt="" src="https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9ublkwNGdUaWFEVmljVFNYZTBpYlJqa3VZMDdrOTJFYVlJcmVnbVJZMVZMaWFscDNqblZLUVFoMTc3ekdBZFhCdzVGbWdpY1RGcnF2a2pGck9pYjRpY0FOVkJiSmcvNjQw?x-oss-process=image/format,png"> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;扫码关注获取更多技术架构知识干货哟~ </td> 
 </tr> 
 </tbody> 
</table>

[format_png]: /images/20230528/b4b3203b65ae45a4a5a5a0d294df3f22.png