聊聊 Kafka：Kafka 消息丟失的場景以及最佳實踐

四、最佳實踐

4.1 生產(chǎn)端

• 不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)。帶有回調(diào)通知的 send 方法可以針對發(fā)送失敗的消息進行重試處理。

• 設(shè)置 acks = all。代表了你對“已提交”消息的定義。如果設(shè)置成 all，則表明所有副本 Broker 都要接收到消息，該消息才算是“已提交”。這是最高等級的“已提交”定義。

  

• 設(shè)置 retries = 3，當(dāng)出現(xiàn)網(wǎng)絡(luò)的瞬時抖動時，消息發(fā)送可能會失敗，此時配置了 retries > 0 的 Producer 能夠自動重試消息發(fā)送，避免消息丟失。

  

  
  如果重試達到設(shè)定的次數(shù)，那么生產(chǎn)者就會放棄重試并返回異常。不過并不是所有的異常都是可以通過重試來解決的，比如消息太大，超過max.request.size參數(shù)配置的值時，這種方式就不可行了。

• 設(shè)置 retry.backoff.ms = 300，合理估算重試的時間間隔，可以避免無效的頻繁重試。
  

  

  
  它用來設(shè)定兩次重試之間的時間間隔，避免無效的頻繁重試。在配置 retries 和 retry.backoff.ms之前，最好先估算一下可能的異常恢復(fù)時間，這樣可以設(shè)定總的重試時間大于這個異常恢復(fù)時間，以此來避免生產(chǎn)者過早地放棄重試。

4.2 Broker 端

• 設(shè)置 unclean.leader.election.enable = false。它控制的是哪些 Broker 有資格競選分區(qū)的 Leader。如果一個 Broker 落后原先的 Leader 太多，那么它一旦成為新的 Leader，必然會造成消息的丟失。故一般都要將該參數(shù)設(shè)置成 false，即不允許這種情況的發(fā)生。

  
  

  

• 設(shè)置 replication.factor >= 3。其實這里想表述的是，最好將消息多保存幾份，畢竟目前防止消息丟失的主要機制就是冗余。

  
  

  

• 設(shè)置 min.insync.replicas > 1。這控制的是消息至少要被寫入到多少個副本才算是“已提交”。設(shè)置成大于 1 可以提升消息持久性。在實際環(huán)境中千萬不要使用默認值 1。

  
  

  

• 確保 replication.factor > min.insync.replicas。如果兩者相等，那么只要有一個副本掛機，整個分區(qū)就無法正常工作了。我們不僅要改善消息的持久性，防止數(shù)據(jù)丟失，還要在不降低可用性的基礎(chǔ)上完成。推薦設(shè)置成 replication.factor = min.insync.replicas + 1。

4.3 消費端

• 確保消息消費完成再提交。最好把它設(shè)置成 enable.auto.commit = false，并采用手動提交位移的方式。這對于單 Consumer 多線程處理的場景而言是至關(guān)重要的。
  

  

  
  雖然采用手動提交位移的方式可以解決消費端消息丟失的場景，但同時會存在重復(fù)消費問題，關(guān)于重復(fù)消費問題我們下一篇再講。

• 像我們上面說的那個線上問題，即使你設(shè)置了手動提交，消費異常了同時也提交了位移，還是會存在消息丟失。

  Kafka 沒有重試機制不支持消息重試，也沒有死信隊列，因此使用 Kafka 做消息隊列時，需要自己
  實現(xiàn)消息重試的功能。這里我先說下大致的思路，后續(xù)有時間再分享代碼出來：

• 創(chuàng)建一個 Topic 作為重試 Topic，用于接收等待重試的消息。

• 普通 Topic 消費者設(shè)置待重試消息的下一個重試 Topic。

• 從重試 Topic 獲取待重試消息存儲到 Redis 的 ZSet 中，并以下一次消費時間排序。

• 定時任務(wù)從 Redis 獲取到達消費時間的消息，并把消息發(fā)送到對應(yīng)的 Topic。

• 同一個消息重試次數(shù)過多則不再重試。