注意：Kafka 的這六個(gè)場(chǎng)景會(huì)丟失消息！
大家好，我是君哥。
我們使用 Kafka 的時(shí)候，怎樣能保證不丟失消息呢？今天來(lái)聊一聊這個(gè)話(huà)題。
首先我們看一下 Kafka 的架構(gòu)圖，
場(chǎng)景一：異步發(fā)送
Producer 異步發(fā)送是丟失消息比較多的場(chǎng)景，Kafka 異步發(fā)送的代碼如下：
ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("the-topic", key, value);
RecordMetadata metadata = producer.send(record).get();Producer 發(fā)送消息后，不用等待發(fā)送結(jié)果，就可以繼續(xù)執(zhí)行后面的邏輯。如果發(fā)送失敗，就會(huì)丟失消息。
Kafka 提供了回調(diào)方法，可以同步等待發(fā)送結(jié)果，這樣降低了發(fā)送效率，但可以對(duì)發(fā)送失敗的場(chǎng)景進(jìn)行處理，比如重新發(fā)送。
ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("the-topic", key, value);
producer.send(record,
(Callback) (metadata, e) -> {
if(e != null) {
e.printStackTrace();
} else {
System.out.println("The offset of the record we just sent is: " + metadata.offset());
}
});場(chǎng)景二：配置 acks=0
從文章開(kāi)頭的架構(gòu)圖中可以看到，Broker Leader 節(jié)點(diǎn)收到消息后，會(huì)同步給 Follower 節(jié)點(diǎn)。
在 Producer 端有一個(gè) acks 配置，說(shuō)明如下 ：
- acks=0：Producer 發(fā)送消息后不等待 Broker 的響應(yīng)；
- acks=1：Producer 發(fā)送消息后，Leader 節(jié)點(diǎn)寫(xiě)入消息成功后給 Producer 發(fā)送響應(yīng)；
- acks=all/-1：Producer 發(fā)送消息后，需要 ISR 列表中所有 Broker 節(jié)點(diǎn)都寫(xiě)入消息成功才會(huì)給 Producer 發(fā)送響應(yīng)。
注意：acks=all/-1 是最高安全級(jí)別，可以配合 min.insync.replicas 參數(shù)使用，當(dāng) acks=all/-1 時(shí)，min.insync.replicas 表示 ISR 列表中最小寫(xiě)入消息成功的副本數(shù)。
如下圖，cks=all/-1，當(dāng) min.insync.replicas=2 時(shí)。
如果 ISR 列表中有【Broker0、Broker1】，即使 Broker2 寫(xiě)入消息失敗，也會(huì)給 Producer 返回成功。
如果 ISR 列表中只有【Broker0】，則無(wú)論如何都不會(huì)給 Producer 返回成功。
如果 ISR 列表中有【Broker0、Broker1、Broker2】，則 3 個(gè) Broker 都寫(xiě)成功才會(huì)給 Producer 返回成功。
場(chǎng)景三：發(fā)送端重試
如果配置 retries=0，Producer 發(fā)送消息失敗后是不會(huì)進(jìn)行重試的，要保證消息不丟失，可以增加 retries 的配置值，避免因?yàn)榫W(wǎng)絡(luò)抖動(dòng)而造成的發(fā)送失敗。
場(chǎng)景四：Follower 落后太多
Kafka Broker 有一個(gè)參數(shù)：unclean.leader.election.enable，這個(gè)參數(shù)值說(shuō)明如下：
- true：允許 ISR 列表之外的節(jié)點(diǎn)參與競(jìng)選 Leader；
- false：不允許 ISR 列表之外的節(jié)點(diǎn)參與競(jìng)選 Leader。
如果設(shè)置為 true，也是會(huì)丟失消息的，看下圖：
如果 Leader 和 Follower1 都掛了，這時(shí)就要考慮是否讓 Follower2 參加競(jìng)選，把 unclean.leader.election.enable 參數(shù)值設(shè)置為 true，則 Follower2 也可以競(jìng)選 Leader，并且作為唯一存活節(jié)點(diǎn)成功競(jìng)選為 Leader，但是它并沒(méi)有同步到偏移量為 3、4、5 的消息，
而之前的 Leader 上線(xiàn)后，成為了 Follower，因?yàn)?Follower 的 LEO（Log End Offset）不能大于 Leader，所以之前偏移量為 3、4、5 的消息就被丟棄了。如下圖：
所以，要保證消息不丟失，unclean.leader.election.enable 這個(gè)參數(shù)值要設(shè)置為 false。
場(chǎng)景五：Broker 宕機(jī)
為了提升性能，Kafka 使用 Page Cache，先將消息寫(xiě)入 Page Cache，采用了異步刷盤(pán)機(jī)制去把消息保存到磁盤(pán)。如果刷盤(pán)之前，Broker Leader 節(jié)點(diǎn)宕機(jī)了，并且沒(méi)有 Follower 節(jié)點(diǎn)可以切換成 Leader，則 Leader 重啟后這部分未刷盤(pán)的消息就會(huì)丟失。
這種場(chǎng)景下多設(shè)置副本數(shù)是一個(gè)好的選擇，通常的做法是設(shè)置 replication.factor >= 3，這樣每個(gè) Partition 就會(huì)有 3 個(gè)以上 Broker 副本來(lái)保存消息，同時(shí)宕機(jī)的概率很低。
同時(shí)可以配合場(chǎng)景二中的參數(shù) min.insync.replicas > 1（不建議使用默認(rèn)值 1），表示消息至少要被成功寫(xiě)入到 2 個(gè) Broker 副本才算是發(fā)送成功。
注意：參數(shù)配置要保證 replication.factor > min.insync.replicas，通常設(shè)置成 replication.factor = min.insync.replicas + 1。如果這 2 個(gè)參數(shù)設(shè)置成相等，則只要有一個(gè) Broker 節(jié)點(diǎn)宕機(jī)，Broker 就無(wú)法給 Producer 返回發(fā)送成功，系統(tǒng)可用性降低。
場(chǎng)景六：并發(fā)消費(fèi)
如果消費(fèi)端采用多線(xiàn)程并發(fā)消費(fèi)，很容易因?yàn)椴l(fā)更新 Offset 導(dǎo)致消費(fèi)失敗。看下圖：
線(xiàn)程 1 拉取 3 條消息把 Offset 更新成 3，線(xiàn)程 2 把 Offset 更新成 6，線(xiàn)程 3 把 Offset 更新成 9。這時(shí)如果線(xiàn)程 2 消費(fèi)失敗了，想要重新消費(fèi)，但是 Offset 已經(jīng)更新到了 9，不能拉取到 Offset 9 以前的消息了。
所以，消費(fèi)者并發(fā)消費(fèi)很可能會(huì)造成消息丟失，如果對(duì)消息丟失很敏感，最好使用單線(xiàn)程來(lái)進(jìn)行消費(fèi)。
如果采用多線(xiàn)程，可以把 enable.auto.commit 設(shè)置為 false，這樣相當(dāng)于每次消費(fèi)完后手動(dòng)更新 Offset。不過(guò)這又會(huì)帶來(lái)重復(fù)消費(fèi)問(wèn)題，比如上面的例子，如果線(xiàn)程 2 消費(fèi)失敗了，則手動(dòng)把 Offset 更新成 3，線(xiàn)程 3 消費(fèi)成功后，再次拉取，還會(huì)拉取到 6、7、8 這三條數(shù)據(jù)。因此消費(fèi)端需要做好冪等處理。
總結(jié)
本文介紹了 Kafka 丟失消息的六個(gè)場(chǎng)景，使用 Kafka 時(shí)需要根據(jù)實(shí)際情況制定解決方案，希望本文介紹的場(chǎng)能夠?qū)δ阌兴鶐椭?/p>
分享到微信 
分享到微博
