kafka是如何保证消息不丢失的？生产端/消费端

ゞ浴缸里的玫瑰 2021-09-25 03:44 399阅读 0赞

## **前提条件** ##

任何消息组件不丢数据都是在特定场景下一定条件的，`kafka`要保证消息不丢，有两个核心条件。

第一，必须是`已提交的消息`，即`committed message`。`kafka`对于`committed message`的定义是，生产者提交消息到`broker`，并等到多个`broker`确认并返回给生产者已提交的确认信息。而这`多个broker`是由我们自己来定义的，可以选择只要有一个`broker`成功保存该消息就算是已提交，也可以是令所有`broker`都成功保存该消息才算是已提交。不论哪种情况，`kafka`只对已提交的消息做持久化保证。

第二，也就是最基本的条件，虽然`kafka`集群是分布式的，但也必须保证有足够`broker`正常工作，才能对消息做持久化做保证。也就是说 `kafka`不丢消息是有前提条件的，假如你的消息保存在 N 个`kafka broker`上，那么这个前提条件就是这 N 个`broker`中至少有 1 个存活。只要这个条件成立，`kafka`就能保证你的这条消息永远不会丢失。

## **如何保证消息不丢** ##

一条消息从产生，到发送到`kafka`保存，到被取出消费，会有多个场景和流程阶段，可能会出现丢失情况，我们聊一下`kafka`通过哪些手段来保障消息不丢。

### **生产端** ###

`Producer`端可能会丢失消息。目前`Kafka Producer`是异步发送消息的，也就是说如果你调用的是`producer.send(msg)`这个`API`，那么它通常会立即返回，但此时你不保证消息发送已成功完成。可能会出现：网络抖动，导致消息压根就没有发送到`Broker`端；或者消息本身不合规导致`Broker`拒绝接收（比如消息太大了，超过了`Broker`的限制）。

实际上，使用`producer.send(msg, callback)`接口就能避免这个问题，根据回调，一旦出现消息提交失败的情况，就可以有针对性地进行处理。如果是因为那些瞬时错误，`Producer`重试就可以了；如果是消息不合规造成的，那么调整消息格式后再次发送。总之，处理发送失败的责任在`Producer`端而非`Broker`端。当然，如果此时`broker`宕机，那就另当别论，需要及时处理`broker`异常问题。

### **消费端** ###

`Consumer`端丢数据的情况，稍微复杂点。`Consumer`有个”位移“(`offset`)的概念，表示`Consumer`当前消费到`topic`分区的哪个位置。如图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70][]

`kafka`通过先消费消息，后更新`offset`，来保证消息不丢失。但是这样可能会出现消息重复的情况，具体如何保证`only-once`，后续再单独分享。

当我们`consumer`端开启多线程异步去消费时，情况又会变得复杂一些。此时`consumer`自动地向前更新`offset`，假如其中某个线程运行失败了，它负责的消息没有被成功处理，但位移已经被更新了，因此这条消息对于`consumer`而言实际上是丢失了。这里的关键就在自动提交`offset`，如何真正地确认消息是否真的被消费，再进行更新`offset`。

这个问题的解决起来也简单：如果是多线程异步处理消费消息，`consumer`不要开启自动提交`offset`，`consumer`端程序自己来处理`offset`的提交更新。提醒你一下，单个`consumer`程序使用多线程来消费消息说起来容易，写成代码还是有点麻烦的，因为你很难正确地处理`offset`的更新，也就是说避免无消费消息丢失很简单，但极易出现消息被消费了多次的情况。

## **实践配置** ##

最后分享下`kafka`无消息丢失配置：

1.  `producer`端使用`producer.send(msg, callback)`带有回调的`send`方法。
2.  设置`acks = all`。`acks`是`Producer`的一个参数，代表“已提交”消息的定义。如果设置成`all`，则表明所有`Broker`都要接收到消息，该消息才算是“已提交”。
3.  设置`retries`为一个较大的值。同样是`Producer`的参数。当出现网络抖动时，消息发送可能会失败，此时配置了`retries`的`Producer`能够自动重试发送消息，尽量避免消息丢失。
4.  设置`unclean.leader.election.enable = false`。这是`Broker`端的参数，在`kafka`版本迭代中社区也多次反复修改过他的默认值，之前比较具有争议。它控制哪些`Broker`有资格竞选分区的`Leader`。如果一个`Broker`落后原先的`Leader`太多，那么它一旦成为新的`Leader`，将会导致消息丢失。故一般都要将该参数设置成`false`。
5.  设置`replication.factor >= 3`。这也是`Broker`端的参数。保存多份消息冗余，不多解释了。
6.  设置`min.insync.replicas > 1`。`Broker`端参数，控制消息至少要被写入到多少个副本才算是“已提交”。设置成大于 1 可以提升消息持久性。在生产环境中不要使用默认值 1。确保`replication.factor > min.insync.replicas`。如果两者相等，那么只要有一个副本离线，整个分区就无法正常工作了。推荐设置成`replication.factor = min.insync.replicas + 1`。
7.  确保消息消费完成再提交。`Consumer`端有个参数`enable.auto.commit`，最好设置成`false`，并自己来处理`offset`的提交更新。

参考：[https://cloud.tencent.com/developer/article/1589157][https_cloud.tencent.com_developer_article_1589157]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70]: /images/20210923/2b176d1a81e1432f9be7edeac20258ec.png
[https_cloud.tencent.com_developer_article_1589157]: https://cloud.tencent.com/developer/article/1589157