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