大家好，我是君哥。

我們使用 Kafka 的時候，怎樣能保證不丟失消息呢？今天來聊一聊這個話題。

首先我們看一下 Kafka 的架構圖，

場景一：異步發送

Producer 異步發送是丟失消息比較多的場景，Kafka 異步發送的代碼如下：

ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("the-topic", key, value);
RecordMetadata metadata = producer.send(record).get();

Producer 發送消息后，不用等待發送結果，就可以繼續執行后面的邏輯。如果發送失敗，就會丟失消息。

Kafka 提供了回調方法，可以同步等待發送結果，這樣降低了發送效率，但可以對發送失敗的場景進行處理，比如重新發送。

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());
                    }
                });

場景二：配置 acks=0

從文章開頭的架構圖中可以看到，Broker Leader 節點收到消息后，會同步給 Follower 節點。

在 Producer 端有一個 acks 配置，說明如下 ：

• acks=0：Producer 發送消息后不等待 Broker 的響應；
• acks=1：Producer 發送消息后，Leader 節點寫入消息成功后給 Producer 發送響應；
• acks=all/-1：Producer 發送消息后，需要 ISR 列表中所有 Broker 節點都寫入消息成功才會給 Producer 發送響應。

注意：acks=all/-1 是最高安全級別，可以配合 min.insync.replicas 參數使用，當 acks=all/-1 時，min.insync.replicas 表示 ISR 列表中最小寫入消息成功的副本數。

如下圖，cks=all/-1，當 min.insync.replicas=2 時。

如果 ISR 列表中有【Broker0、Broker1】，即使 Broker2 寫入消息失敗，也會給 Producer 返回成功。

如果 ISR 列表中只有【Broker0】，則無論如何都不會給 Producer 返回成功。

如果 ISR 列表中有【Broker0、Broker1、Broker2】，則 3 個 Broker 都寫成功才會給 Producer 返回成功。

場景三：發送端重試

如果配置 retries=0，Producer 發送消息失敗后是不會進行重試的，要保證消息不丟失，可以增加 retries 的配置值，避免因為網絡抖動而造成的發送失敗。

場景四：Follower 落后太多

Kafka Broker 有一個參數：unclean.leader.election.enable，這個參數值說明如下：

• true：允許 ISR 列表之外的節點參與競選 Leader；
• false：不允許 ISR 列表之外的節點參與競選 Leader。

如果設置為 true，也是會丟失消息的，看下圖：

如果 Leader 和 Follower1 都掛了，這時就要考慮是否讓 Follower2 參加競選，把 unclean.leader.election.enable 參數值設置為 true，則 Follower2 也可以競選 Leader，并且作為唯一存活節點成功競選為 Leader，但是它并沒有同步到偏移量為 3、4、5 的消息，

而之前的 Leader 上線后，成為了 Follower，因為 Follower 的 LEO（Log End Offset）不能大于 Leader，所以之前偏移量為 3、4、5 的消息就被丟棄了。如下圖：

所以，要保證消息不丟失，unclean.leader.election.enable 這個參數值要設置為 false。

場景五：Broker 宕機

為了提升性能，Kafka 使用 Page Cache，先將消息寫入 Page Cache，采用了異步刷盤機制去把消息保存到磁盤。如果刷盤之前，Broker Leader 節點宕機了，并且沒有 Follower 節點可以切換成 Leader，則 Leader 重啟后這部分未刷盤的消息就會丟失。

這種場景下多設置副本數是一個好的選擇，通常的做法是設置 replication.factor >= 3，這樣每個 Partition 就會有 3 個以上 Broker 副本來保存消息，同時宕機的概率很低。

同時可以配合場景二中的參數 min.insync.replicas > 1（不建議使用默認值 1），表示消息至少要被成功寫入到 2 個 Broker 副本才算是發送成功。

注意：參數配置要保證 replication.factor > min.insync.replicas，通常設置成 replication.factor = min.insync.replicas + 1。如果這 2 個參數設置成相等，則只要有一個 Broker 節點宕機，Broker 就無法給 Producer 返回發送成功，系統可用性降低。

場景六：并發消費

如果消費端采用多線程并發消費，很容易因為并發更新 Offset 導致消費失敗。看下圖：

線程 1 拉取 3 條消息把 Offset 更新成 3，線程 2 把 Offset 更新成 6，線程 3 把 Offset 更新成 9。這時如果線程 2 消費失敗了，想要重新消費，但是 Offset 已經更新到了 9，不能拉取到 Offset 9 以前的消息了。

所以，消費者并發消費很可能會造成消息丟失，如果對消息丟失很敏感，最好使用單線程來進行消費。

如果采用多線程，可以把 enable.auto.commit 設置為 false，這樣相當于每次消費完后手動更新 Offset。不過這又會帶來重復消費問題，比如上面的例子，如果線程 2 消費失敗了，則手動把 Offset 更新成 3，線程 3 消費成功后，再次拉取，還會拉取到  6、7、8 這三條數據。因此消費端需要做好冪等處理。

總結

本文介紹了 Kafka 丟失消息的六個場景，使用 Kafka 時需要根據實際情況制定解決方案，希望本文介紹的場能夠對你有所幫助。