Kafka核心技术与实战 11 无消息丢失

喜欢ヅ旅行 2022-10-15 09:57 205阅读 0赞

**Kafka 只对“已提交”的消息（committed message）做有限度的持久化保证。**

### 生产者程序丢失数据 ###

目前 Kafka Producer 是异步发送消息的，也就是说如果你调用的是 producer.send(msg) 这个 API，那么它通常会立即返回，但此时你不能认为消息发送已成功完成。

用这个方式，可能会有哪些因素导致消息没有发送成功呢？其实原因有很多，例如网络抖动，导致消息压根就没有发送到 Broker 端；或者消息本身不合格导致 Broker 拒绝接收（比如消息太大了，超过了 Broker 的承受能力）等。

**解决：**

**Producer 永远要使用带有回调通知的发送 API，也就是说不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)**。不要小瞧这里的 callback（回调），它能准确地告诉你消息是否真的提交成功了。一旦出现消息提交失败的情况，你就可以有针对性地进行处理。

如果是因为那些瞬时错误，那么仅仅让 Producer 重试就可以了；如果是消息不合格造成的，那么可以调整消息格式后再次发送。总之，处理发送失败的责任在 Producer 端而非 Broker 端。

### 消费者程序丢失数据 ###

展示了 Consumer 端的位移数据

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FidTkzNTAwOTA2Ng_size_16_color_FFFFFF_t_70][]

**维持先消费消息（阅读），再更新位移（书签）的顺序**

**如果是多线程异步处理消费消息，Consumer 程序不要开启自动提交位移，而是要应用程序手动提交位移**。

### 总结 ###

1.  不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)。记住，一定要使用带有回调通知的 send 方法。
2.  设置 acks = all。acks 是 Producer 的一个参数，代表了你对“已提交”消息的定义。如果设置成 all，则表明所有副本 Broker 都要接收到消息，该消息才算是“已提交”。这是最高等级的“已提交”定义。
3.  设置 retries 为一个较大的值。这里的 retries 同样是 Producer 的参数，对应前面提到的 Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了 retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失。
4.  设置 unclean.leader.election.enable = false。这是 Broker 端的参数，它控制的是哪些 Broker 有资格竞选分区的 Leader。如果一个 Broker 落后原先的 Leader 太多，那么它一旦成为新的 Leader，必然会造成消息的丢失。故一般都要将该参数设置成 false，即不允许这种情况的发生。
5.  设置 replication.factor >= 3。这也是 Broker 端的参数。其实这里想表述的是，最好将消息多保存几份，毕竟目前防止消息丢失的主要机制就是冗余。
6.  设置 min.insync.replicas > 1。这依然是 Broker 端参数，控制的是消息至少要被写入到多少个副本才算是“已提交”。设置成大于 1 可以提升消息持久性。在实际环境中千万不要使用默认值 1。
7.  确保 replication.factor > min.insync.replicas。如果两者相等，那么只要有一个副本挂机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要在不降低可用性的基础上完成。推荐设置成 replication.factor = min.insync.replicas + 1。
8.  确保消息消费完成再提交。Consumer 端有个参数 enable.auto.commit，最好把它设置成 false，并采用手动提交位移的方式。就像前面说的，这对于单 Consumer 多线程处理的场景而言是至关重要的。

### 常见问题 ###

*  replication.refactor是副本replica总数， min.insync.replicas是要求确保至少有多少个replica副本写入后才算是提交成功，这个参数是个硬指标；acks=all是个动态指标，确保当前能正常工作的replica副本都写入后才算是提交成功。举个例子：比如，此时副本总数3，即replication.refactor = 3，设置min.insync.replicas=2，acks=all，那如果所有副本都正常工作，消息要都写入三个副本，才算提交成功，此时这个min.insync.replicas=2下限值不起作用。如果其中一个副本因为某些原因挂了，此时acks=all的动态约束就是写入两个副本即可，触达了min.insync.replicas=2这个下限约束。如果三个副本挂了两个，此时ack=all的约束就变成了1个副本，但是因为有min.insync.replicas=2这个下限约束，写入就会不成功。

acks = 0/1/all 是指 接收/落盘/ISR同步，三个步骤，而不是指副本数量，你不能设置acks = 5的。

*  ISR In-Sync Replicas，这是一个副本集合，里面的所有副本都是和Leader副本保持同步的
 *  producer发送消息后，broker成功写入消息了，但是ack因为网络问题没有到达producer，生产者可能会重试发送这条消息。  
    这种问题如何避免重复消费呢 使用幂等producer 幂等，就是对同一条消息，无论producer发送多少次，都是一样的效果。一旦写入一次，第N次重复写入，都不会在broker产生新的message
 *  关于offset，有必要外部持久化记录吗？ 出问题后查出来继续消费。你说的更新offset是怎么处理? 是直接调用它的api更新吗？

你可以选择将offset保存在外部持久化设备中，不过更常见的做法是使用consumer API保存在Kafka里面。常见的API包括commitSync和commitAsync

*  acks=all，只是保证消息被存储到所有broker的pagecache中就认为消息成功，如果再此时间点，所有broker全部异常停机，那么pagecache中的消息就不会被异步拉取到磁盘，当重启之后，消息就丢失了。这种情况如何考虑呢？
    
    是的，你说的这种情况是会出现消息丢失，所以Kafka保证不丢消息是有条件的，即ISR中至少有一个副本同步了该消息且存活。不可能无条件地保证消息不丢失。
 *  broker上成功保存该消息，是指持久化到磁盘，还是只是写到page cache中？

写到page cache中

*  异步发送消息，如果retry，是否会造成消息乱序？

是可能的，所以有max.in.flight.requests.per.connection这个参数。

参考：

*  https://time.geekbang.org/column/article/102931

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FidTkzNTAwOTA2Ng_size_16_color_FFFFFF_t_70]: /images/20221014/1948c6bd19bd4c82b427176ef1180e5c.png