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

消息丟失

• 生產(chǎn)者
  • acks = 0
    • 不需要等待任何Broker確認(rèn)收到消息的回復(fù)就可以繼續(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ā)消息
      • 只要保證有一個副本存活就不會丟消息，一般使用在金融場景，當(dāng)然如果配置副本只有一個也可能會丟消息跟acks=1情況類似
• 消費(fèi)者
  • 如果消費(fèi)者配置的是自動提交，恰好此時消費(fèi)服務(wù)掛了，沒有處理完的所有數(shù)據(jù)，這樣就導(dǎo)致了數(shù)據(jù)丟失，下次也消費(fèi)不到了

重復(fù)消費(fèi)

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

消費(fèi)亂序

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

消息積壓

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

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

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

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

延遲消息

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

消息回溯

• 如果覺得某段時間對已消費(fèi)消息的結(jié)果存在質(zhì)疑，比如代碼存在問題，當(dāng)修復(fù)之后，可以指定offset將過去的消息重新消費(fèi)一次

消息傳遞保障

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

Kafka的事務(wù)

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

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

如何保證消息不丟失

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

消息積壓如何處理

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

如何保證消息順序

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