Kafka生產(chǎn)常見問題分析與總結

消息丟失

• 生產(chǎn)者
  • acks = 0
    • 不需要等待任何Broker確認收到消息的回復就可以繼續(xù)發(fā)消息
      • 性能最高，但是最容易丟消息，對于數(shù)據(jù)丟失不敏感的場景可以使用，如大數(shù)據(jù)統(tǒng)計報表
  • acks = 1
    • 只要等待Broker中的leader成功寫入數(shù)據(jù)成功就可以繼續(xù)發(fā)消息
      • 如果follower沒有成功備份數(shù)據(jù)而此時leader剛好掛了，就會丟消息
  • acks = -1 或 all
    • 等待Broker中的leader、follower都寫入成功才可以繼續(xù)發(fā)消息
      • 只要保證有一個副本存活就不會丟消息，一般使用在金融場景，當然如果配置副本只有一個也可能會丟消息跟acks=1情況類似
• 消費者
  • 如果消費者配置的是自動提交，恰好此時消費服務掛了，沒有處理完的所有數(shù)據(jù)，這樣就導致了數(shù)據(jù)丟失，下次也消費不到了

重復消費

• 生產(chǎn)者
  • 發(fā)送消息如果配置了重試機制，比如網(wǎng)絡抖動時間過長導致發(fā)送端發(fā)送超時，實際Broker可能已經(jīng)接受到消息，但發(fā)送方會重發(fā)消息
• 消費者
  • 如果消費消息配置了自動提交，剛拉取了一批處理了一部分，但是尚未提交，服務掛了，下次重啟時又會拉取到相同的一批數(shù)據(jù)重復處理，一般情況下消費端會進行冪等性處理

消費亂序

• 如果發(fā)送端配置重試機制，Kafka不會等之前那條消息完全發(fā)送成功才去發(fā)送下一條消息，這樣可能會出現(xiàn)發(fā)送了1、2、3三條消息，第1條消息超時了，后面2條消息發(fā)送成功，然后再重試發(fā)送第1條消息，這時Broker端分區(qū)存入的消息順序為2、3、1，所以是否需要配置重試機制得根據(jù)業(yè)務去定，當然也可以用同步發(fā)送的模式去發(fā)送并且acks≠0，這樣也能保證消息從發(fā)送到消費是全鏈路有序的
  • 發(fā)送端的消息發(fā)送模式
    • 發(fā)后即忘
      • 不關心消息是否成功到達，對返回結果不做任何判斷處理，這種方式注重吞吐量，但是無法保證消息的可靠性
    • 同步
      • 消息發(fā)送出去之后，關心消費端是否成功接受，只有成功了才能繼續(xù)下一條
    • 異步
      • 在發(fā)送消息的同時通過指定的回調函數(shù)去進行消費端的響應處理
• 注意: Kafka保證全鏈路消息順序消費需要從生產(chǎn)端開始
  • 方案一: 將所有有序消息發(fā)送到同一個分區(qū)，然后使用一個消費者去消費，但是這種性能較低
  • 方案二: 可以在消費者端接受到消息后將需要保證順序消費的幾條消息發(fā)送到內存隊列(可以整多個)，一個內存隊列安排一個線程去順序處理

消息積壓

• 線上有時因為生產(chǎn)端發(fā)送消息速度過快或者消費端消費過慢，可能會導致Broker積壓大量未消費的消息

  • 一般情況下可以通過增加當前topic的分區(qū)將消息拆分到更多的分區(qū)中去，同時增加對應的消費者去進行消費
    • 消費者數(shù) = 分區(qū)數(shù)
  • 如果積壓了百萬級消息需要進行緊急處理，可以修改消費端程序，將其收到的消息快速轉發(fā)到其他topic(可以設置多個分區(qū))，然后再啟動多個消費者去同時消費新增topic多個分區(qū)下消息

• 由于消息數(shù)據(jù)格式變動或消費端程序存在問題，導致消費端消費失敗，可能會導致Broker積壓大量未消費的消息

  • 可以將這些消費失敗的消息轉發(fā)到其他隊列(類似死信隊列)，后面再慢慢分析死信隊列中的消息去進行問題處理

延遲消息

• 延時隊列存儲的對象是延時消息，所謂的延時消息就是發(fā)送出去之后，消費端需要等待某個特定的時間才能進行獲取到該消息進行消費
  • 應用場景
    • 超時訂單
    • 訂單完成多長時間后通知進行評價
  • 實現(xiàn)思路
    • 發(fā)送延時消息先把消息按照不同的延遲時間段發(fā)送到指定的topic中，然后通過定時任務進行輪詢消費這些topic，查看消息是否到期，如果時間到了就把這些消息發(fā)送到具體業(yè)務的topic中
      • 注意
        • 如果用定時任務執(zhí)行，對項目性能也是一種考驗，而且會有一定的延遲，如果要保證時間偏差在2min左右，這樣會導致執(zhí)行過于頻繁，所以比建議使用Kafka去實現(xiàn)延遲消息，建議使用RocketMQ、RabbtMQ

消息回溯

• 如果覺得某段時間對已消費消息的結果存在質疑，比如代碼存在問題，當修復之后，可以指定offset將過去的消息重新消費一次

消息傳遞保障

• at most once(最多收到一次)
  • 生產(chǎn)端使用 acks = 0
• at least lonce(至少收到一次)
  • 生產(chǎn)端使用 acks = -1 || all
• exactly once(收到一次)
  • at least once 加上消費端增加冪等性處理，也可以使用Kafka生產(chǎn)者的冪等性來實現(xiàn)
    • Kafka生產(chǎn)者的冪等性
      • 因為生產(chǎn)端重試導致消息重復發(fā)送，Kafka的冪等性可以保證重復發(fā)送的消息只接受一次，只需要在生產(chǎn)端參數(shù)開啟即可

Kafka的事務

• Kafka的事務不同于RocketMQ，RocketMQ是保障本地事務(比如數(shù)據(jù)庫)與MQ消息發(fā)送的事務一致性，Kafka的事務主要保障一次發(fā)送多條消息的事務一致性(要么同時成功，要么同時失敗)，一般在Kafka流式計算場景較多

生產(chǎn)問題場景匯總

如何保證消息不丟失

• 生產(chǎn)端發(fā)送消息到Broker不丟失
  • 生產(chǎn)端配置
    • acks = 0
      • 不需要等待任何Broker確認收到消息的回復就可以繼續(xù)發(fā)消息
      • 性能最高，但是最容易丟消息，對于數(shù)據(jù)丟失不敏感的場景可以使用，如大數(shù)據(jù)統(tǒng)計報表
    • acks = 1
      • 只要等待Broker中的leader成功寫入數(shù)據(jù)成功就可以繼續(xù)發(fā)消息
        • 如果follower沒有成功備份數(shù)據(jù)而此時leader剛好掛了，就會丟消息
    • acks = -1 或 all
      • 等待Broker中的leader、follower都寫入成功才可以繼續(xù)發(fā)消息
        • 只要保證有一個副本存活就不會丟消息，一般使用在金融場景，當然如果配置副本只有一個也可能會丟消息跟acks = 1情況類似
  • 對于生產(chǎn)端只要使用acks = 1 || all 即可，生產(chǎn)端發(fā)送消息后可以拿到Broker的反饋去進行判斷是否發(fā)送成功，再根據(jù)是否需要重發(fā)
• Broker端保存消息不丟失
  • 合理優(yōu)化刷盤頻率，防止服務異常崩潰造成消息未刷盤
    • Kafka的消息是先寫入操作系統(tǒng)的頁緩存中，然后再刷盤寫入硬盤，頁緩存中的消息斷電即丟失，Kafka不支持寫一條刷一次盤的同步機制，只能通過調整刷盤頻率提升消息安全，另外需要配置多備份因子，避免單點消息丟失，配置好備份因子之后，Kafka會給每個分區(qū)分配多個備份分區(qū)，這些分區(qū)會盡量平均分配到多個Broker上，當出現(xiàn)故障時也能進行選舉，繼續(xù)向外提供服務
• 消費端防止異步處理丟失消息
  • 消費者端由于有消息重試機制，正常情況下不會丟消息，每次消費處理一批消息，需要在處理完之后給Brocker進行應答，提交當前消息offset，Broker進行應答后，會推進本地日志的offset記錄，如果Broker沒有接到應答，Broker會重新向一個消費者組的消費者推送消息，最終保證消息不丟失，消費端采用手動提交offset的方式，相比自動提交更容易掌握提交offset的時機
  • 消費端唯一要注意的是，不能進行異步處理業(yè)務邏輯，因為如果業(yè)務邏輯異步進行，而消費者已經(jīng)同步提交了offset，如果業(yè)務邏輯出現(xiàn)異常失敗了，此時Broker已經(jīng)收到的消費者應答，后續(xù)不會再重新推送消息，造成業(yè)務層面的消息丟失

消息積壓如何處理

• 業(yè)務運行正常的情況下
  • 如果只是因為消費端處理消息過慢造成積壓，可以增加對應topic的分區(qū)數(shù)，將消息拆分到更多的分區(qū)中，然后增加同比例的消費者數(shù)，另外再發(fā)送消息的時候，盡量要保證各個分區(qū)之間的數(shù)據(jù)分布均衡，可以調整生產(chǎn)端的分區(qū)策略，讓后續(xù)更多的消息分配到新增的分區(qū)里，或新開一個topic，配置更多的分區(qū)以及對應的消費者數(shù)，然后啟動一批消費者(充當搬運工)，將消息從舊topic轉發(fā)到新topic中去
  • 分區(qū)數(shù) = 消費者數(shù)
• 業(yè)務運行異常的情況下
  • 如果是因為消費端業(yè)務問題導致積壓，影響了程序正常運行，比如消費者序列化失敗、業(yè)務處理異常，可以采用一種降級的方案，先啟動一個消費者將topic下的消息轉發(fā)到其它隊列里(類似于死信隊列)，然后后續(xù)再進行分析以及問題處理

如何保證消息順序

• 如何保證生產(chǎn)端發(fā)送到分區(qū)消息有序
  • 第一種
    • 一個topic配置一個分區(qū)，這樣犧牲吞吐量保證全局有序
  • 第二種
    • 通過定制生產(chǎn)端的分區(qū)器，將消息分配到同一個分區(qū)
      • 可以滿足一些要求局部有序的場景，比如訂單相關的多條消息但是不要求所有消息有序，就可以通過自定義分區(qū)器處理
• 分區(qū)中的消息有序后，如何保證消費端消費順序有序
  • 基于分區(qū)中消息的局部有序性，由于Kafka消費端拉取消息都是并行拉取多個批次的消息進行處理，所以無法保證串行消費，如果非要實現(xiàn)此功能，可以將消息按照業(yè)務獨立性收集到對應的內存隊列中，進行特定的排序進行處理
    • 對于RocketMQ中提供了順序消息，實現(xiàn)原理是先鎖定一個MesageQueue(類似分區(qū))，消費完這個隊列之后再鎖定下一個隊列進行消費