Kafka這名字在Java后端圈子里基本屬于“躲不開”的那一類技術(shù)。不管是搞微服務(wù)、大數(shù)據(jù)管道還是簡單的異步解耦消息隊列總得選一款而Kafka往往是選項里最“硬核”的那個。很多人第一反應(yīng)是“分布式、高吞吐、源碼復(fù)雜”還沒上手就先被勸退也有同學(xué)把Kafka當(dāng)成普通MQ用結(jié)果生產(chǎn)環(huán)境頻頻踩坑才回頭補課。這篇文章不打算從源碼講起也不粘貼一堆文檔參數(shù)湊字數(shù)。我從一名Java開發(fā)者的視角出發(fā)聊清楚三件事Kafka到底是靠什么做到的“快”、我們在項目里怎么用Java客戶端正確讀寫消息以及日常運維/開發(fā)中最容易踩的坑和排查思路。內(nèi)容會帶著我實際項目中的參數(shù)選擇、異常處理和排障實錄力求讓看完的人能直接回到工位上用起來。1. 先搞清楚Kafka是干嘛的——本質(zhì)是“日志”不是“隊列”很多Java開發(fā)者第一次接觸Kafka都會下意識把它和RabbitMQ、ActiveMQ放在一起比較。這種對比能幫助理解但也會帶來誤導(dǎo)Kafka的底層抽象和傳統(tǒng)消息隊列有著根本區(qū)別。1.1 從“削峰填谷”說起Kafka解決的痛點場景很老套某個接口突然被流量打滿數(shù)據(jù)庫連接池耗盡應(yīng)用直接雪崩。加機器數(shù)據(jù)庫成了瓶頸擴展數(shù)據(jù)庫不是寫幾個腳本就能完成的。不加機器流量高峰扛不住。消息隊列在這里面扮演的角色是“蓄水池”請求先打到隊列由消費者按自己的節(jié)奏慢慢處理。Kafka能承擔(dān)這個角色而且承擔(dān)得比大多數(shù)消息中間件更“寬”——單機每秒幾十萬條消息的吞吐量還能扛住長時間的數(shù)據(jù)堆積。用生活里的話說RabbitMQ像一家定制化服務(wù)的小店每個客戶單獨招呼Kafka像一條大型流水線不在乎單個消息的“特殊待遇”只關(guān)心整體處理速度。1.2 核心抽象Topic、Partition、Offset與Consumer GroupKafka最核心的模型我用一張“抽屜”來類比Topic相當(dāng)于一個分類標(biāo)簽所有同類型消息放進同一個Topic。Partition是Topic內(nèi)部的分區(qū)相當(dāng)于把抽屜分成了多個小格子。消息真正存儲的最小單位是分區(qū)。Offset是消息在分區(qū)內(nèi)的序號相當(dāng)于抽屜里每件物品的“位置編號”。Consumer Group是一組消費者協(xié)作消費同一個Topic的機制相當(dāng)于多個人分工處理同一批物品。這套模型里最關(guān)鍵的一點是Topic只是邏輯概念分區(qū)才是物理概念。生產(chǎn)者寫入消息時會根據(jù)key或者輪詢策略決定消息落到哪個分區(qū)消費者讀取時也是在分區(qū)維度上拉取數(shù)據(jù)。為什么要這么設(shè)計直接回答為了并行度和擴展性。一個分區(qū)的消息是有序的但Kafka不承諾Topic維度跨分區(qū)的有序性。如果你只用一個分組消費分區(qū)數(shù)決定了你能啟動多少個消費者實例并行干活如果分區(qū)數(shù)是3即使你啟動了10個消費者也最多只有3個消費者在干活另外7個處于空閑狀態(tài)。這個特性既是Kafka吞吐量的來源也是很多人使用時迷茫的起點——我在后面會專門展開。1.3 Kafka為什么快順序?qū)懕P、零拷貝與批量拉取Kafka高吞吐的三大殺器值得每個Java開發(fā)者認真理解一遍因為理解了它們你才能在設(shè)計業(yè)務(wù)時充分利用Kafka的特性。第一順序?qū)懘疟P。傳統(tǒng)數(shù)據(jù)庫隨機讀寫磁盤磁頭要來回移動尋址耗時嚴(yán)重。Kafka基于文件追加寫的方式每個分區(qū)在磁盤上對應(yīng)一組segment文件消息永遠追加在文件尾部寫入過程接近順序IO速度可以接近內(nèi)存。這里有個容易誤解的點Kafka不是把數(shù)據(jù)放在內(nèi)存里就完事它把數(shù)據(jù)持久化到磁盤但依靠順序?qū)懽尨疟P寫入速度跑出內(nèi)存級體驗。第二零拷貝。消費者讀取消息時傳統(tǒng)流程是磁盤→內(nèi)核態(tài)緩沖區(qū)→用戶態(tài)→應(yīng)用程序→socket發(fā)送數(shù)據(jù)被拷貝了多次。Kafka使用sendfile系統(tǒng)調(diào)用讓數(shù)據(jù)從磁盤直接進入socket緩沖區(qū)繞開了用戶態(tài)拷貝大大降低了IO消耗和上下文切換開銷。第三批量與壓縮。生產(chǎn)者不是一條一條發(fā)送消息的而是把多條消息攢成一個批次batch再發(fā)送消費者拉取時也不是一次一條而是按最大拉取字節(jié)數(shù)批量返回。配合壓縮算法比如lz4、zstdKafka在網(wǎng)絡(luò)傳輸上的效率遠高于逐條發(fā)送。這三條疊加Kafka才扛得住大數(shù)據(jù)生態(tài)里那個著名的口號——“一個Topic承載全公司所有日志”。理解了原理你后面配置參數(shù)時才明白為什么要調(diào)batch.size為什么要關(guān)注頁緩存的利用率。2. 從Java開發(fā)者視角看Kafka——客戶端實踐與參數(shù)選型服務(wù)端原理了解個大概就夠了Java開發(fā)者日常接觸最多的還是客戶端。Kafka官方Java客戶端kafka-clients設(shè)計得相對簡潔但參數(shù)極多每一項都直接影響性能和數(shù)據(jù)可靠性。我挑幾個實操價值最高的配置仔細講。2.1 工程接入Maven依賴與基礎(chǔ)環(huán)境準(zhǔn)備我建議新項目優(yōu)先選用服務(wù)端版本和客戶端版本一致或接近的組合。比如服務(wù)端是2.8.x客戶端就用2.8.x或者3.x早期版本避免協(xié)議兼容性問題。Maven依賴只需要一個dependency groupIdorg.apache.kafka/groupId artifactIdkafka-clients/artifactId version3.4.0/version /dependency不要直接引入kafka全量依賴那會帶出一大堆scala依賴和broker端實現(xiàn)客戶端用不到還會拖大構(gòu)建體積。本機開發(fā)環(huán)境最簡單的方式是用Docker起一個單節(jié)點實例docker run -d --name kafka-dev -p 9092:9092 \ -e KAFKA_ADVERTISED_LISTENERSPLAINTEXT://localhost:9092 \ -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR1 \ -e KAFKA_AUTO_CREATE_TOPICS_ENABLEtrue \ apache/kafka:3.4.0注意上面設(shè)置了分區(qū)副本因子為1因為單節(jié)點多個副本沒有意義而且默認副本因子為3會導(dǎo)致分區(qū)一直處于ISR不滿狀態(tài)。2.2 生產(chǎn)者核心配置從“最快”到“最穩(wěn)”的取舍生產(chǎn)者最核心的三個參數(shù)放在任何生產(chǎn)環(huán)境都必須考慮acksall retries3 enable.idempotencetrue很多人看到acksall會擔(dān)心性能下降確實有影響但換來的是消息不丟失的強保證。acksall表示分區(qū)leader收到消息且ISR中所有副本都同步完成后才返回確認代價是單條消息延遲升高但吞吐量在批量發(fā)送模式下并沒有想象中那么差。我在項目中觀測過acksall配合linger.ms5、batch.size32KB千兆網(wǎng)絡(luò)下單條平均延遲約增加2~4毫秒吞吐量幾乎不變。關(guān)于冪等生產(chǎn)者enable.idempotencetrue很多人忽略它。這個參數(shù)讓producer為每條消息生成序列號broker端用序列號去重保證即使網(wǎng)絡(luò)重試也不會產(chǎn)生重復(fù)消息。Kafka從3.0開始默認開啟冪等但如果你的代碼沒顯式設(shè)置舊版本客戶端還是默認關(guān)閉的。建議所有新項目在生產(chǎn)者配置里加上這行它是性價比極高的一項數(shù)據(jù)保障。剩下的生產(chǎn)者參數(shù)按項目規(guī)模調(diào)整buffer.memory生產(chǎn)者本地緩沖區(qū)大小默認32MB。高并發(fā)場景建議調(diào)大否則瞬時流量積壓會導(dǎo)致RecordTooLargeException或TimeoutException。max.block.ms緩沖區(qū)滿了之后send操作阻塞的最長時間默認60秒太短會導(dǎo)致頻繁報錯太長會拖垮上游接口。compression.type建議lz4或zstdlz4是速度和壓縮比折中zstd壓縮率更高但CPU開銷稍大。日志類數(shù)據(jù)首選zstd業(yè)務(wù)消息用lz4即可。2.3 生產(chǎn)者代碼模板異步發(fā)送與回調(diào)處理Java生產(chǎn)者API的經(jīng)典寫法如下Properties props new Properties(); props.put(bootstrap.servers, 192.168.1.10:9092,192.168.1.11:9092); props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer); props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer); props.put(acks, all); props.put(retries, 3); props.put(linger.ms, 5); props.put(batch.size, 32768); props.put(enable.idempotence, true); KafkaProducerString, String producer new KafkaProducer(props); producer.send(new ProducerRecord(user_events, userId, eventJson), (metadata, exception) - { if (exception null) { log.info(send success, offset{}, metadata.offset()); } else { log.error(send failed, exception); // 這里建議把消息落入本地文件或者Redis做補償 } });這段代碼里最該注意的一點是send()方法是異步的。很多人圖省事直接調(diào)用producer.send(record)不關(guān)心返回結(jié)果當(dāng)broker端異?；蚓W(wǎng)絡(luò)故障時消息其實已經(jīng)寫失敗了但你的業(yè)務(wù)無感知最后數(shù)據(jù)對不上賬又無從排查。生產(chǎn)環(huán)境務(wù)必把回調(diào)函數(shù)寫完整或者至少在發(fā)送失敗時打印日志。多條記錄批量發(fā)送的封裝可以用producer.send循環(huán)也可以自己維護一個List然后循環(huán)發(fā)送后者對性能和可維護性更友好。我曾經(jīng)在一個訂單系統(tǒng)里直接每條消息單獨調(diào)用send()TPS只有幾百時毫無問題后來活動流量激增到每秒五千條生產(chǎn)者遲遲不返回接口堆積崩潰。排查后確認問題不在broker而是本地的buffer.memory被打滿、發(fā)送回調(diào)沒有及時釋放加批量發(fā)送策略后才穩(wěn)定下來。2.4 消費者核心配置消費者組與提交策略消費者端的參數(shù)比生產(chǎn)者更容易踩坑。核心配置如下group.iduser-event-consumer-group enable.auto.commitfalse auto.offset.resetearliest max.poll.records500 max.poll.interval.ms300000 session.timeout.ms10000這里要展開的細節(jié)比較多我分別說明。enable.auto.commit到底是true還是false我強烈建議生產(chǎn)環(huán)境設(shè)置為false用手動提交。自動提交默認是每5秒提交一次消費位點。如果你拉取了一批消息處理耗時超過5秒消費者在提交位點時這批消息可能還沒處理完——這時候進程崩潰重啟后會從已提交的位點繼續(xù)消費那些沒處理完的消息就丟了。即使你的業(yè)務(wù)對少量數(shù)據(jù)丟失不敏感手動提交也能讓你對消費進度“心里有數(shù)”不會被自動提交的滯后時間迷惑。手動提交的兩種方式// 同步提交 —— 提交失敗會立即拋異常便于感知 consumer.commitSync(); // 異步提交 —— 提交失敗不阻斷消費需要回調(diào)處理 consumer.commitAsync((offsets, exception) - { if (exception ! null) { log.error(commit failed, exception); } });我在項目里的習(xí)慣是先commitAsync處理正常情況如果異常退出再加一次commitSync兜底。因為commitSync在rebalance時會阻塞線程沒必要每次都用它拖慢消費速度。auto.offset.reset這個參數(shù)決定了消費者從頭讀還是從最新讀。earliest如果沒有提交位點從分區(qū)最早的消息開始讀。latest如果沒有提交位點從最新的消息開始讀。最容易出問題的是“消費者組第一次啟動”的場景。新consumer group啟動時沒有歷史位點如果業(yè)務(wù)要求處理歷史數(shù)據(jù)用earliest如果只關(guān)心后續(xù)新數(shù)據(jù)用latest。但是只要這個消費組之前已經(jīng)提交過位點auto.offset.reset就不再起作用消費者只會從上次提交的位點繼續(xù)消費。很多人以為改了latest就能從頭消費其實不會生效必須手動重置位點。重置消費位點的命令行kafka-consumer-groups.sh --bootstrap-server localhost:9092 \ --group user-event-consumer-group \ --topic user_events \ --reset-offsets \ --to-earliest \ --execute注意執(zhí)行這個命令前消費者組必須處于停止?fàn)顟B(tài)否則重置會被跳過或者報錯。2.5 消費者代碼模板從poll到處理的完整閉環(huán)Properties props new Properties(); props.put(bootstrap.servers, localhost:9092); props.put(group.id, user-event-service); props.put(key.deserializer, org.apache.kafka.common.serialization.StringDeserializer); props.put(value.deserializer, org.apache.kafka.common.serialization.StringDeserializer); props.put(enable.auto.commit, false); props.put(auto.offset.reset, earliest); props.put(max.poll.records, 500); KafkaConsumerString, String consumer new KafkaConsumer(props); consumer.subscribe(Collections.singletonList(user_events)); try { while (running) { // 注意poll必須循環(huán)調(diào)用不能省略很多初學(xué)者只poll一次就退出了 ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecordString, String record : records) { handleMessage(record.value()); } // 處理完畢后提交 consumer.commitAsync(); } } catch (WakeupException e) { // 關(guān)閉時會觸發(fā)wakeup不用處理 } finally { consumer.commitSync(); consumer.close(); }這段代碼中有個細節(jié)需要注意poll(Duration)的語義是“最多阻塞這么久等待消息到達”而不是“超過這個時間就拋異常”。只要你循環(huán)調(diào)用poll消費者就會持續(xù)工作。如果業(yè)務(wù)消息來得少比如每秒只有幾條poll會頻繁返回空列表這完全正常不要去降低阻塞時間來提升“靈敏度”純粹是無效操作。處理消息的handleMessage方法里還需要考慮一個問題如果某條消息處理失敗到底要不要提交offset兩個常見做法直接拋出異常讓消費線程退出/停止這樣該批記錄的offset沒有提交后續(xù)rebalance時會再次消費到失敗記錄實現(xiàn)“重試”。但要注意如果消息本身是臟數(shù)據(jù)每次都失敗就會形成消費卡死。捕獲異常并記錄日志然后繼續(xù)消費。這種方式適合“消息丟了影響不大”的場景但要注意數(shù)據(jù)完整性。更穩(wěn)妥的方案是結(jié)合死信隊列處理失敗的消息先判斷重試次數(shù)達到閾值后寫入一個專門存放死信的Topic人工介入修復(fù)。這個方案需要自己寫一點重試狀態(tài)邏輯但對于核心鏈路的數(shù)據(jù)完整性非常有效。3. 實際項目中的Topic設(shè)計與參數(shù)調(diào)優(yōu)到了真實項目里Kafka的難點往往不在API使用而在于“設(shè)計”?！霸趺炊ǚ謪^(qū)數(shù)”“怎么設(shè)計消息結(jié)構(gòu)”“怎么選擇消息可靠性級別”每一樣都影響后續(xù)的系統(tǒng)穩(wěn)定性和排障難度。3.1 分區(qū)數(shù)量的設(shè)定別拍腦袋定個數(shù)分區(qū)數(shù)是Kafka并行度的上限這決定了消費端能啟動的最大并發(fā)消費者數(shù)量。設(shè)計分區(qū)數(shù)先要回答一個問題你的下游處理能力能消化多少并發(fā)我見過團隊一上來就建48個分區(qū)的Topic理由是“以后業(yè)務(wù)量會很大多分區(qū)保險”。結(jié)果消費者只有兩臺機器各2個實例總共4個消費者大部分分區(qū)處于無人消費狀態(tài)。分流不均、rebalance耗時增加、個別分區(qū)數(shù)據(jù)傾斜運維復(fù)雜度遠超實際收益。分區(qū)數(shù)參考公式基于目標(biāo)吞吐量計算目標(biāo)吞吐量每秒需要處理的峰值消息數(shù) 5000條/秒 單個消費者實際處理能力1000條/秒含業(yè)務(wù)邏輯 所需消費者實例數(shù) 5000 / 1000 5個 分區(qū)數(shù) 消費者實例數(shù) 或 略大于該值預(yù)留彈性和rebalance空間一般建議分區(qū)數(shù)取消費者實例數(shù)的1.5~2倍既能充分利用消費能力又能在某個消費者宕機時有足夠分區(qū)分配給其他消費者。比如需要5個消費者分區(qū)數(shù)設(shè)為10左右是合理區(qū)間。另外一個容易忽略的問題Kafka基本不支持減少分區(qū)數(shù)只能增加。增加分區(qū)會導(dǎo)致key-based路由下同一key的消息分布到新的分區(qū)打亂原有順序。所以設(shè)計分區(qū)數(shù)時寧多勿少同時也要避免無腦擴大。3.2 消息體設(shè)計的加分項schema、key與時間戳Kafka的消息格式自由但建議在項目一開始就定好規(guī)范寫入統(tǒng)一信封格式包含eventId、eventType、timestamp、payload。用eventId做去重依據(jù)比用payload內(nèi)容優(yōu)雅得多。合理設(shè)置key如果業(yè)務(wù)要求同一個用戶的消息按時間順序處理key就用userId如果不關(guān)心順序甚至可以不傳key讓Kafka輪詢分區(qū)分擔(dān)負載。傳key后同一個key一定進同一個分區(qū)但也可能造成某些熱點key所在分區(qū)數(shù)據(jù)膨脹。慎重使用Avro/Protobuf還是JSONJSON最方便但占用空間大Protobuf性能最好但引入schema管理和代碼生成成本。我的建議是團隊規(guī)模小、階段早先用JSON快速迭代等業(yè)務(wù)穩(wěn)定、消息量大到需要優(yōu)化時再遷移到Protobuf或Avro。3.3 消息可靠性級別三種組合看清代價Kafka的可靠性除了生產(chǎn)者端acks還涉及broker端的副本配置以及消費者端的提交策略。我把組合整理成一張表可靠性級別生產(chǎn)者設(shè)置消費者設(shè)置可能存在的問題適用場景最高不丟不重acksall,enable.idempotencetrue手動提交冪等消費延遲升高消費端需處理重復(fù)訂單、支付、積分變動中盡量不丟允許不重復(fù)acksall自動提交極端情況下消費端可能重復(fù)或丟失一般業(yè)務(wù)通知低高吞吐優(yōu)先acks0自動提交消息可能大量丟失日志采集、監(jiān)控指標(biāo)實際做項目時很多同學(xué)把“最高可靠”設(shè)為默認但代價是吞吐量下降和消費端復(fù)雜度上升。建議按消息業(yè)務(wù)價值分級核心交易鏈路走最高可靠普通用戶行為日志走最低可靠兩者各有歸屬不浪費性能也不犧牲關(guān)鍵數(shù)據(jù)。從我自己的經(jīng)歷來看Kafka大部分可靠性問題出在“配置組合不一致”上生產(chǎn)者用了acksall消費者卻用了自動提交加短poll數(shù)據(jù)鏈路依然是脆弱的?？煽啃缘耐暾湕l是由三端生產(chǎn)者、broker、消費者共同組成的只看單點配置沒有意義。4. 常見問題與排查技巧實錄Kafka開發(fā)里最考驗人的不是寫代碼而是線上出問題后能不能快速定位。我把反復(fù)踩過的坑整理成一份“排障速查表”每一條都對應(yīng)真實的排查過程。4.1 消費者組頻繁Rebalance消費時停時續(xù)這是消費者端最經(jīng)典的問題?，F(xiàn)象是日志里頻繁出現(xiàn)“Rebalance started/Completed”消費速率忽高忽低有時甚至幾分鐘不消費。原因一單條消息處理時間過長超過max.poll.interval.ms默認5分鐘。消費者poll后開始處理處理時間超過閾值broker認為消費者死亡觸發(fā)rebalance把分區(qū)分配給其他消費者。等原消費者處理完再poll時又會觸發(fā)新一輪rebalance。原因二session.timeout.ms設(shè)置太短而消費者的GC停頓或網(wǎng)絡(luò)抖動超過該時間broker誤判會話失效。排查建議分兩步查看日志里rebalance前是否有處理耗時長的方法調(diào)用。我遇到過某個業(yè)務(wù)在消息處理器里同步調(diào)了一次外部HTTP接口接口平均耗時2秒但偶發(fā)超時達到30秒而當(dāng)時max.poll.interval.ms只有60秒高峰時段連續(xù)觸發(fā)rebalance協(xié)調(diào)開銷巨大。檢查消費者實例的JVM GC日志。Full GC超過幾秒在session.timeout.ms10000的設(shè)置下很容易被判死。調(diào)整方向調(diào)大max.poll.interval.ms和session.timeout.ms給業(yè)務(wù)處理留足時間。在消息處理邏輯里把耗時操作異步化或者拆成多個步驟放在線程池里執(zhí)行縮減poll間隔內(nèi)的工作量。減少單次poll回來的消息數(shù)量比如max.poll.records100可以有效降低單批處理時間。4.2 消息發(fā)送成功但消費端永遠收不到有一個排查場景讓我印象很深服務(wù)A發(fā)送消息到Topic控制臺確認消息數(shù)在上漲但消費端就是打印不到日志。查了消費者日志沒有異常查了offset顯示消費位點沒動查了消費者組狀態(tài)顯示“Stable”。最后發(fā)現(xiàn)消費端訂閱的Topic名稱寫錯了——消息發(fā)到了user_events消費者訂閱的是user-event連字符和下劃線之差數(shù)據(jù)各走各路。因為Kafka允許自動創(chuàng)建不存在的Topic開發(fā)環(huán)境開啟auto.create.topics.enable消費者訂閱時自動建了一個空Topic所以既不報錯也不消費。排查順序建議固定下來先看消費端有沒有加入正確的group用命令確認kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group user-event-consumer-group輸出會顯示LAG字段如果LAG一直為0說明消費端在正常消費問題在下游邏輯如果LAG持續(xù)增長且活躍消費者為0說明消費者根本沒poll到數(shù)據(jù)。確認Topic列表和分區(qū)leader狀態(tài)kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic user_events確認客戶端配置的bootstrap.servers能連通所有broker不只是某一個節(jié)點。這種問題最坑的地方在于一切“看起來正?！币驗镵afka的容錯設(shè)計讓異常靜默化了。務(wù)必在消費者啟動日志里加上“subscribed topic xxx”的輸出方便排查時一眼看出訂閱對沒對。4.3 消息堆積嚴(yán)重分區(qū)數(shù)據(jù)傾斜消息堆積在三種場景下出現(xiàn)消費者處理速度跟不上生產(chǎn)速度。分區(qū)數(shù)遠多于消費者數(shù)導(dǎo)致部分分區(qū)長期無人消費。某個分區(qū)的數(shù)據(jù)量特別大比如同一key的消息特別多把消費者線程堵在該分區(qū)上。前面兩種好解決加消費者實例、調(diào)大max.poll.records、優(yōu)化處理邏輯。第三種比較隱蔽本質(zhì)是key設(shè)計的問題。我曾見過一個行為日志Topic全部用userId做key結(jié)果是某個頭部用戶的消息占了一個分區(qū)總消息量的50%以上其他分區(qū)空閑。由于同key消息必須有序處理消費者只能串行處理該分區(qū)其他線程再快也幫不上忙。解決思路熱點key加后綴打散到多個分區(qū)比如userId _ (hash % 10)。代價是同一個用戶的消息可能亂序降級為“最終可達”要在業(yè)務(wù)上評估能否接受。如果必須保證嚴(yán)格順序只能接受熱點分區(qū)慢的事實并把該分區(qū)的消費鏈路單獨優(yōu)化比如減少下游IO操作。日常如何發(fā)現(xiàn)數(shù)據(jù)傾斜除了業(yè)務(wù)日志可以借助Kafka自帶的消費組命令看每個分區(qū)的LAG差異kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group xxx輸出中同一個分區(qū)如果LAG達到幾十萬而其他分區(qū)LAG為零就要考慮熱點key的問題了。4.4 消費重復(fù)數(shù)據(jù)怎么從根本上去重Kafka官方語義是“至少一次”At Least Once也就是說重復(fù)消費是設(shè)計上允許發(fā)生的。生產(chǎn)端重試會重復(fù)消費端rebalance會重復(fù)。要保證業(yè)務(wù)數(shù)據(jù)的最終一致性必須自己搞定冪等。第一層防線利用業(yè)務(wù)消息內(nèi)的唯一ID做去重。比如訂單消息自帶orderId消費時先查Redis里有沒有這個orderId的處理標(biāo)記有則跳過沒有則處理并寫入標(biāo)記。這個方法簡單有效絕大多數(shù)場景夠用。第二層防線數(shù)據(jù)庫寫入用唯一約束。比如消費邏輯是往訂單表插入記錄直接在表上建orderId唯一索引重復(fù)消費時數(shù)據(jù)庫會拒絕重復(fù)插入日志報錯后業(yè)務(wù)判定為“已處理”。這種方式比Redis標(biāo)記更可靠因為Redis本身也可能丟數(shù)據(jù)。第三層防線消費邏輯天然冪等。比如“累加積分”“更新最后登錄時間”這類操作重復(fù)執(zhí)行多次和執(zhí)行一次結(jié)果一致。設(shè)計新系統(tǒng)時盡量把消費邏輯設(shè)計成這種形式從根上消滅重復(fù)問題。我在不同項目里都實踐過上述方案最省心的還是數(shù)據(jù)庫唯一索引。但要注意如果消費邏輯涉及多個表的更新唯一索引只能保證單表不重復(fù)跨表事務(wù)還是要用分布式事務(wù)或事務(wù)性消息方案這屬于另一個深水區(qū)了本文不展開。4.5 Broker端“磁盤寫滿”和“分區(qū)副本失效”最后提兩個運維側(cè)的常見問題因為Java開發(fā)者排查消息發(fā)送超時時很容易忽略broker的狀態(tài)。磁盤寫滿Kafka不會主動拒絕寫入請求它會持續(xù)報NotEnoughReplicasException或DiskUsageException但業(yè)務(wù)側(cè)表現(xiàn)出來的是生產(chǎn)者的send回調(diào)異常。排查時看broker日志里是否有“Disk is full”相關(guān)消息。預(yù)留足夠的磁盤空間、配置好清理策略是關(guān)鍵。Kafka的日志清理策略有兩種delete按過期時間刪除舊數(shù)據(jù)compact按key保留最新值。業(yè)務(wù)消息一般用delete配置log.retention.hours即可。副本失效某個broker宕機后它管轄的分區(qū)副本全部變?yōu)榉峭綘顟B(tài)。如果所有副本都失效分區(qū)就不可用。日常檢查/describe輸出看Isr列是否包含所有副本。處理方式取決于集群的容災(zāi)需求。如果允許自動恢復(fù)確保unclean.leader.election.enablefalse避免“臟副本”被選為leader導(dǎo)致消息丟失。如果該參數(shù)被改成true數(shù)據(jù)丟失的風(fēng)險會顯著升高但分區(qū)可用性提高了。在業(yè)務(wù)允許臨時不可用但絕不允許丟消息的場景下務(wù)必保持false。5. 基于實際項目經(jīng)驗的幾條終極大坑提醒寫到這里Kafka客戶端使用、參數(shù)配置、問題排查的主體內(nèi)容基本覆蓋了。最后再分享幾條只有實際跑過線上系統(tǒng)才會體會到的經(jīng)驗每一句話都對應(yīng)過我踩過的坑。第一條不要為了“解耦”而強行引入Kafka。有些系統(tǒng)只有兩個服務(wù)A調(diào)用B同步調(diào)用完全能滿足卻為了“跟上架構(gòu)潮流”加入Kafka。結(jié)果消息鏈路多出一環(huán)排障復(fù)雜度上升延遲多了幾十毫秒分布式一致性問題也隨之而來。Kafka更像是“高吞吐量場景下的基礎(chǔ)設(shè)施”小流量系統(tǒng)用HTTP同步調(diào)用反而更簡單可靠。第二條Consumer的冪等不是Kafka的職責(zé)是業(yè)務(wù)自己的職責(zé)。無數(shù)團隊在消費端接入Kafka后被重復(fù)消息打得焦頭爛額才意識到Kafka官方從未承諾過不重復(fù)。提前在系統(tǒng)設(shè)計階段把冪等方案想清楚遠比事后補丁優(yōu)雅。第三條監(jiān)控一定要前置。Kafka的排障大多數(shù)情況下靠數(shù)據(jù)說話沒有監(jiān)控全靠日志撈非常痛苦。至少要在應(yīng)用容器里集成Metrics采集關(guān)注消費延遲、生產(chǎn)者發(fā)送失敗率、請求隊列平均時間這幾個指標(biāo)。我在項目上線穩(wěn)定運行兩個月后深有體會監(jiān)控指標(biāo)比任何代碼都能更快告訴你系統(tǒng)什么時候開始“不舒服”。第四條升級版本要謹慎。我曾在一個項目中從Kafka 2.7升級到3.2單測階段一切正常上預(yù)發(fā)后發(fā)現(xiàn)一個消費組的位點全部跳到latest。排查后確認是舊版客戶端提交的位點在新版服務(wù)端讀取時出現(xiàn)了協(xié)議不兼容問題。升級前先讀官方遷移文檔尤其關(guān)注“client compatibility”章節(jié)最好先在預(yù)發(fā)環(huán)境跑滿一天再灰度生產(chǎn)。Kafka的深度遠不止這篇筆記講完的內(nèi)容還有Kafka Streams流處理、Exactly-Once語義、Schema Registry等更進階的話題。但作為一名Java開發(fā)者先把客戶端用穩(wěn)、把可靠性參數(shù)理解透、把常見排障方法掌握住已經(jīng)能支撐90%的日常工作了。之后再有新的實戰(zhàn)經(jīng)驗我會繼續(xù)在這個話題下補充。