1. 從一條消息的旅程說(shuō)起Kafka Sender 線(xiàn)程到底在忙什么很多人第一次接觸 Kafka 生產(chǎn)者的時(shí)候腦子里裝的都是producer.send()這一行代碼覺(jué)得消息發(fā)出去就完事了。但真正在生產(chǎn)環(huán)境里跑過(guò)一段時(shí)間的人都知道send()只是把消息丟進(jìn)了一個(gè)緩沖區(qū)真正把數(shù)據(jù)推送到 broker 的是背后那個(gè)默默干活的Sender 線(xiàn)程。這個(gè)線(xiàn)程是 Kafka 生產(chǎn)者客戶(hù)端里唯一負(fù)責(zé)網(wǎng)絡(luò) I/O 的線(xiàn)程它的設(shè)計(jì)直接決定了生產(chǎn)者的吞吐量、延遲和可靠性。我見(jiàn)過(guò)不少團(tuán)隊(duì)在壓測(cè)時(shí)發(fā)現(xiàn)吞吐上不去第一反應(yīng)是加分區(qū)、加 broker結(jié)果折騰半天發(fā)現(xiàn)瓶頸其實(shí)在客戶(hù)端這邊——Sender 線(xiàn)程的配置沒(méi)調(diào)好或者對(duì)它的工作方式理解有偏差。所以這篇文章我想把 Sender 線(xiàn)程從里到外拆一遍講清楚它是怎么攢批、怎么選節(jié)點(diǎn)、怎么發(fā)請(qǐng)求、怎么處理響應(yīng)的以及在實(shí)際調(diào)優(yōu)中哪些參數(shù)真正起作用。這篇文章適合已經(jīng)用過(guò) Kafka 生產(chǎn)者、但對(duì)內(nèi)部機(jī)制還比較模糊的開(kāi)發(fā)者也適合正在做生產(chǎn)者性能調(diào)優(yōu)的運(yùn)維和架構(gòu)同學(xué)。我會(huì)盡量用大白話(huà)把機(jī)制講透同時(shí)給出可以直接抄的配置和排查方法。核心關(guān)鍵詞就一個(gè)Kafka Sender 線(xiàn)程如何發(fā)送數(shù)據(jù)。搞懂它你就能明白為什么你的生產(chǎn)者有時(shí)候快有時(shí)候慢為什么消息會(huì)丟為什么延遲忽高忽低。在展開(kāi)之前先建立一個(gè)整體認(rèn)知Kafka 生產(chǎn)者客戶(hù)端里有兩個(gè)核心線(xiàn)程角色。一個(gè)是用戶(hù)線(xiàn)程也就是你調(diào)用send()的那個(gè)線(xiàn)程它負(fù)責(zé)序列化、分區(qū)選擇、把消息追加到緩沖區(qū)。另一個(gè)就是Sender 線(xiàn)程它是一個(gè)獨(dú)立的后臺(tái)線(xiàn)程負(fù)責(zé)從緩沖區(qū)里把消息撈出來(lái)、按 broker 分組、組裝成請(qǐng)求、通過(guò)NetworkClient發(fā)出去、然后處理響應(yīng)。兩者之間通過(guò)一個(gè)叫RecordAccumulator的緩沖區(qū)解耦用戶(hù)線(xiàn)程只管往里塞Sender 線(xiàn)程只管往外發(fā)。這種生產(chǎn)者-消費(fèi)者模型是整個(gè)高吞吐的基石。2. Sender 線(xiàn)程的整體設(shè)計(jì)與核心思路拆解2.1 為什么要把網(wǎng)絡(luò) I/O 單獨(dú)抽一個(gè)線(xiàn)程要理解 Sender 線(xiàn)程的設(shè)計(jì)先得回答一個(gè)問(wèn)題為什么不讓用戶(hù)線(xiàn)程直接發(fā)答案其實(shí)很樸素——批量。如果每個(gè)用戶(hù)線(xiàn)程調(diào)用send()就立刻發(fā)一個(gè)網(wǎng)絡(luò)請(qǐng)求那網(wǎng)絡(luò)往返的開(kāi)銷(xiāo)會(huì)把吞吐量壓得極低。一條消息一次 RTT就算 RTT 只有 1ms一秒也就千把條這在 Kafka 的場(chǎng)景里完全不夠看。所以 Kafka 的做法是用戶(hù)線(xiàn)程把消息塞進(jìn)緩沖區(qū)就立刻返回Sender 線(xiàn)程在后臺(tái)攢一批消息湊夠一定大小或者等夠一定時(shí)間再一次性發(fā)出去。這樣一次網(wǎng)絡(luò)請(qǐng)求能帶上成百上千條消息攤薄了網(wǎng)絡(luò)開(kāi)銷(xiāo)。這就是所謂的批量發(fā)送batching也是 Sender 線(xiàn)程存在的根本理由。另一個(gè)原因是線(xiàn)程安全與復(fù)雜度隔離。網(wǎng)絡(luò) I/O 涉及連接管理、重試、超時(shí)、響應(yīng)解析等一堆狀態(tài)如果讓每個(gè)用戶(hù)線(xiàn)程都去碰這些鎖競(jìng)爭(zhēng)和狀態(tài)同步會(huì)非常復(fù)雜。把網(wǎng)絡(luò) I/O 收斂到單一線(xiàn)程用戶(hù)線(xiàn)程只需要操作緩沖區(qū)整個(gè)模型就清晰多了。當(dāng)然代價(jià)是 Sender 線(xiàn)程成了單點(diǎn)它的處理能力就是生產(chǎn)者的上限這也是為什么調(diào)優(yōu)時(shí)總盯著它。2.2 RecordAccumulator用戶(hù)線(xiàn)程和 Sender 線(xiàn)程的交接區(qū)RecordAccumulator是理解 Sender 線(xiàn)程的關(guān)鍵。你可以把它想象成一個(gè)按分區(qū)組織的隊(duì)列集合。每個(gè)分區(qū)對(duì)應(yīng)一個(gè)DequeProducerBatch也就是一個(gè)雙端隊(duì)列隊(duì)列里裝的是一個(gè)個(gè)批次batch。用戶(hù)線(xiàn)程追加消息時(shí)會(huì)找到對(duì)應(yīng)分區(qū)的隊(duì)列看看隊(duì)尾那個(gè)批次還有沒(méi)有空間有就塞進(jìn)去沒(méi)有就新建一個(gè)批次再塞。這里有個(gè)細(xì)節(jié)值得說(shuō)批次的大小由batch.size控制默認(rèn) 16KB。注意這個(gè) 16KB 是每個(gè)批次的上限不是每個(gè)分區(qū)的上限。一個(gè)分區(qū)可以堆積很多個(gè)批次直到觸達(dá)buffer.memory這個(gè)總內(nèi)存上限。當(dāng)緩沖區(qū)滿(mǎn)了用戶(hù)線(xiàn)程的send()就會(huì)阻塞阻塞時(shí)間由max.block.ms控制默認(rèn) 60 秒超時(shí)就拋異常。Sender 線(xiàn)程干活的時(shí)候就是從這些隊(duì)列的頭部取批次。為什么是頭部因?yàn)橐ＷC順序。同一個(gè)分區(qū)內(nèi)的消息必須按順序發(fā)送所以只能從最早入隊(duì)的批次開(kāi)始取。取的時(shí)候它會(huì)檢查這個(gè)批次是否就緒——要么批次已經(jīng)滿(mǎn)了要么等待時(shí)間超過(guò)了linger.ms。只有就緒的批次才會(huì)被取走發(fā)送沒(méi)就緒的就繼續(xù)等。2.3 攢批策略batch.size 和 linger.ms 的博弈Sender 線(xiàn)程什么時(shí)候把批次發(fā)出去取決于兩個(gè)參數(shù)的配合。batch.size決定批次多大算滿(mǎn)linger.ms決定最多等多久。這兩個(gè)參數(shù)是一對(duì)典型的吞吐與延遲的權(quán)衡。如果linger.ms0默認(rèn)值意味著批次一創(chuàng)建就立刻標(biāo)記為可發(fā)送Sender 線(xiàn)程下一輪循環(huán)就會(huì)把它取走。這種情況下攢批效果很差基本是來(lái)一條發(fā)一條延遲最低但吞吐最差。如果把linger.ms設(shè)成 5ms 或 10msSender 線(xiàn)程會(huì)等這段時(shí)間讓更多消息進(jìn)入同一個(gè)批次吞吐能提升好幾倍代價(jià)是每條消息多了幾毫秒的延遲。我個(gè)人的經(jīng)驗(yàn)是對(duì)延遲敏感的在線(xiàn)業(yè)務(wù)linger.ms 設(shè) 0 到 5ms對(duì)吞吐敏感的日志、埋點(diǎn)類(lèi)業(yè)務(wù)設(shè) 10 到 100ms。但要注意linger.ms不是硬等待如果批次在等待期間就滿(mǎn)了會(huì)立刻發(fā)送不用等滿(mǎn) linger 時(shí)間。所以設(shè)大一點(diǎn)通常沒(méi)壞處只是極端低峰期會(huì)有額外延遲。2.4 一個(gè)容易忽略的點(diǎn)批次是按分區(qū)獨(dú)立的新手常有的誤解是Sender 線(xiàn)程一次發(fā)一批消息給一個(gè) broker。實(shí)際上更準(zhǔn)確的說(shuō)法是Sender 線(xiàn)程一次會(huì)把發(fā)往同一個(gè) broker 的多個(gè)分區(qū)的多個(gè)批次合并成一個(gè)請(qǐng)求發(fā)出去。因?yàn)橐粋€(gè) broker 可能承載多個(gè)分區(qū)作為這些分區(qū)的 leader把這些分區(qū)的數(shù)據(jù)打包進(jìn)一個(gè)請(qǐng)求能進(jìn)一步減少網(wǎng)絡(luò)往返。這就引出了 Sender 線(xiàn)程的核心工作流先從所有就緒的批次里按broker 節(jié)點(diǎn)分組把發(fā)往同一個(gè)節(jié)點(diǎn)的批次聚在一起然后為每個(gè)節(jié)點(diǎn)構(gòu)造一個(gè)ProduceRequest再通過(guò)NetworkClient發(fā)出去。所以 Sender 線(xiàn)程的一輪循環(huán)本質(zhì)上是收集就緒批次 → 按節(jié)點(diǎn)分組 → 構(gòu)造請(qǐng)求 → 發(fā)送 → 處理響應(yīng)這樣一個(gè)閉環(huán)。3. Sender 線(xiàn)程發(fā)送數(shù)據(jù)的完整流程拆解3.1 主循環(huán) run() 的骨架Sender 線(xiàn)程實(shí)現(xiàn)了Runnable它的run()方法是一個(gè)while循環(huán)只要生產(chǎn)者沒(méi)關(guān)閉就一直轉(zhuǎn)。每一輪循環(huán)大致做這么幾件事調(diào)用accumulator.ready()找出所有就緒的批次得到一批ReadyCheckResult里面包含就緒的節(jié)點(diǎn)集合、是否還有未就緒的批次、以及是否需要更新元數(shù)據(jù)。如果發(fā)現(xiàn)某些節(jié)點(diǎn)沒(méi)有可用的 leader 信息就觸發(fā)元數(shù)據(jù)更新請(qǐng)求。調(diào)用accumulator.drain()把就緒批次從隊(duì)列里取出來(lái)按節(jié)點(diǎn)分組同時(shí)做一次過(guò)期批次的清理。為每個(gè)節(jié)點(diǎn)構(gòu)造ProduceRequest通過(guò)NetworkClient.send()發(fā)出去。調(diào)用client.poll()處理網(wǎng)絡(luò) I/O包括讀取響應(yīng)、處理超時(shí)、執(zhí)行回調(diào)。處理發(fā)送失敗需要重試的批次把它們重新放回隊(duì)列。這個(gè)循環(huán)的節(jié)奏由NetworkClient.poll()的阻塞時(shí)間控制通常不會(huì)空轉(zhuǎn)燒 CPU因?yàn)?poll 在沒(méi)有事件時(shí)會(huì)阻塞一小段時(shí)間由request.timeout.ms等參數(shù)間接影響。3.2 ready() 階段誰(shuí)有資格被發(fā)送ready()是每輪循環(huán)的第一步它的任務(wù)是篩選。它會(huì)遍歷所有分區(qū)的批次隊(duì)列對(duì)每個(gè)隊(duì)列頭部的批次做判斷這個(gè)批次是否已經(jīng)滿(mǎn)了或者它的創(chuàng)建時(shí)間加上linger.ms是否已經(jīng)早于當(dāng)前時(shí)間只要滿(mǎn)足其一這個(gè)批次就算就緒。同時(shí)它還會(huì)檢查這個(gè)批次所屬分區(qū)的 leader 節(jié)點(diǎn)是否已知。如果 leader 未知說(shuō)明元數(shù)據(jù)過(guò)期了需要標(biāo)記要更新元數(shù)據(jù)這個(gè)批次暫時(shí)發(fā)不了。這一步很關(guān)鍵因?yàn)?Kafka 集群的分區(qū) leader 可能因?yàn)?broker 故障而發(fā)生遷移如果客戶(hù)端還往舊 leader 發(fā)就會(huì)失敗。ready()返回的結(jié)果里有個(gè)readyNodes集合就是這一輪有數(shù)據(jù)要發(fā)的節(jié)點(diǎn)。還有個(gè)nextReadyCheckDelayMs告訴循環(huán)下一輪大概什么時(shí)候再檢查避免無(wú)謂的空轉(zhuǎn)。3.3 drain() 階段把批次撈出來(lái)并分組drain()是真正把數(shù)據(jù)從緩沖區(qū)搬走的動(dòng)作。它接收ready()給出的節(jié)點(diǎn)集合然后對(duì)每個(gè)節(jié)點(diǎn)遍歷它作為 leader 的所有分區(qū)從每個(gè)分區(qū)的隊(duì)列頭部取出就緒的批次放進(jìn)一個(gè)MapInteger, ListProducerBatch里key 是節(jié)點(diǎn) idvalue 是發(fā)往這個(gè)節(jié)點(diǎn)的批次列表。這里有個(gè)重要的順序保證同一個(gè)分區(qū)的批次是按入隊(duì)順序取出的所以發(fā)送順序和寫(xiě)入順序一致。但不同分區(qū)之間沒(méi)有順序保證這是 Kafka 分區(qū)模型的固有特性。drain()還會(huì)順手做一件事清理過(guò)期批次。如果某個(gè)批次在隊(duì)列里待的時(shí)間超過(guò)了delivery.timeout.ms默認(rèn) 120 秒它會(huì)被直接標(biāo)記為失敗觸發(fā)回調(diào)不再發(fā)送。這個(gè)機(jī)制防止了消息無(wú)限期地卡在緩沖區(qū)里。3.4 構(gòu)造請(qǐng)求ProduceRequest 里裝了什么拿到按節(jié)點(diǎn)分組的批次后Sender 線(xiàn)程會(huì)為每個(gè)節(jié)點(diǎn)構(gòu)造一個(gè)ProduceRequest。這個(gè)請(qǐng)求里包含acks來(lái)自acks配置決定需要多少個(gè)副本確認(rèn)。timeout來(lái)自request.timeout.msbroker 端等待 acks 的超時(shí)。每個(gè)分區(qū)的批次數(shù)據(jù)包括分區(qū)號(hào)、批次里的消息集合、以及一些壓縮相關(guān)的元信息。注意批次在進(jìn)入緩沖區(qū)之前就已經(jīng)被壓縮了如果配置了compression.type所以請(qǐng)求里帶的是壓縮后的數(shù)據(jù)這能顯著減少網(wǎng)絡(luò)傳輸量。壓縮是在用戶(hù)線(xiàn)程追加消息時(shí)按批次做的不是 Sender 線(xiàn)程做的這點(diǎn)別搞混。請(qǐng)求構(gòu)造好后通過(guò)NetworkClient.send()交給網(wǎng)絡(luò)層。NetworkClient會(huì)維護(hù)到每個(gè) broker 的連接如果連接還沒(méi)建立會(huì)先發(fā)起連接請(qǐng)求進(jìn)入排隊(duì)等待。3.5 處理響應(yīng)成功、失敗與重試client.poll()會(huì)把 broker 返回的響應(yīng)讀進(jìn)來(lái)交給對(duì)應(yīng)的回調(diào)處理。響應(yīng)分幾種情況成功批次里的所有消息都寫(xiě)入成功觸發(fā)每個(gè)消息的回調(diào)釋放批次占用的內(nèi)存?？芍卦嚨氖”热?leader 切換、網(wǎng)絡(luò)抖動(dòng)導(dǎo)致的NotLeaderForPartitionException、NetworkException等。這些批次會(huì)被重新放回對(duì)應(yīng)分區(qū)的隊(duì)列頭部等待下一輪重發(fā)。重試次數(shù)由retries控制默認(rèn)Integer.MAX_VALUE但實(shí)際受delivery.timeout.ms約束。不可重試的失敗比如消息太大超過(guò)max.request.size、序列化錯(cuò)誤等直接標(biāo)記失敗觸發(fā)回調(diào)。重試這里有個(gè)坑如果開(kāi)啟了enable.idempotencetrue冪等生產(chǎn)者重試是安全的broker 會(huì)去重。如果沒(méi)開(kāi)冪等重試可能導(dǎo)致消息重復(fù)。所以生產(chǎn)環(huán)境我強(qiáng)烈建議開(kāi)冪等代價(jià)很小收益很大。3.6 內(nèi)存管理批次什么時(shí)候被釋放批次發(fā)送成功或最終失敗后它占用的內(nèi)存要還給BufferPool。這個(gè)內(nèi)存池是RecordAccumulator管理的默認(rèn)總大小由buffer.memory控制默認(rèn) 32MB。批次釋放后等待在send()上阻塞的用戶(hù)線(xiàn)程就能被喚醒繼續(xù)往里寫(xiě)。這里有個(gè)容易被忽視的點(diǎn)如果 Sender 線(xiàn)程發(fā)送速度跟不上用戶(hù)線(xiàn)程寫(xiě)入速度緩沖區(qū)會(huì)滿(mǎn)用戶(hù)線(xiàn)程就會(huì)阻塞。表現(xiàn)出來(lái)就是send()調(diào)用變慢甚至超時(shí)拋TimeoutException: Failed to allocate memory within the configured max blocking time。遇到這個(gè)錯(cuò)誤要么加大buffer.memory要么優(yōu)化 Sender 線(xiàn)程的發(fā)送效率比如調(diào)大batch.size、linger.ms或者增加分區(qū)數(shù)讓并發(fā)度更高。4. 關(guān)鍵參數(shù)與實(shí)操配置詳解4.1 影響 Sender 線(xiàn)程的核心參數(shù)一覽下面這張表是我在實(shí)際調(diào)優(yōu)中總結(jié)的、和 Sender 線(xiàn)程關(guān)系最密切的參數(shù)。理解它們之間的聯(lián)動(dòng)比單獨(dú)記某個(gè)值重要得多。參數(shù)默認(rèn)值作用調(diào)優(yōu)方向batch.size16KB單個(gè)批次大小上限吞吐優(yōu)先可調(diào)到 64KB~128KBlinger.ms0批次等待時(shí)間吞吐優(yōu)先調(diào)到 10~100msbuffer.memory32MB緩沖區(qū)總內(nèi)存高吞吐場(chǎng)景調(diào)到 64MB~128MBmax.block.ms60ssend() 阻塞上限一般不用改除非要快速失敗request.timeout.ms30s請(qǐng)求超時(shí)跨機(jī)房可適當(dāng)調(diào)大delivery.timeout.ms120s消息投遞總超時(shí)應(yīng)大于 linger request.timeoutretriesMAX重試次數(shù)配合冪等使用max.in.flight.requests.per.connection5單連接未確認(rèn)請(qǐng)求數(shù)開(kāi)冪等時(shí)可保持 5compression.typenone壓縮算法推薦 lz4 或 zstdacksall確認(rèn)級(jí)別可靠性?xún)?yōu)先用 all4.2 batch.size 和 linger.ms 怎么配才合理這兩個(gè)參數(shù)是 Sender 線(xiàn)程效率的核心。我的經(jīng)驗(yàn)公式是先確定你能接受的最大延遲把 linger.ms 設(shè)成這個(gè)延遲的一半左右然后根據(jù)平均消息大小反推 batch.size。舉個(gè)例子假設(shè)你的消息平均 1KB目標(biāo)吞吐是 50MB/s能接受 10ms 延遲。那么linger.ms5一個(gè)批次在 5ms 內(nèi)大概能攢到 50MB/s × 5ms 250KB 的數(shù)據(jù)。但batch.size默認(rèn)才 16KB根本裝不下批次會(huì)頻繁滿(mǎn)。所以應(yīng)該把batch.size調(diào)到 256KB 左右讓批次能容納這段時(shí)間的數(shù)據(jù)。反過(guò)來(lái)如果消息很小比如 100 字節(jié)batch.size16KB能裝 160 條其實(shí)已經(jīng)夠了沒(méi)必要調(diào)太大否則低峰期批次遲遲不滿(mǎn)反而增加延遲。提示batch.size是每個(gè)批次的上限不是每個(gè)分區(qū)的上限。一個(gè)分區(qū)可以同時(shí)存在多個(gè)批次所以調(diào)大它不會(huì)導(dǎo)致內(nèi)存爆炸但會(huì)改變內(nèi)存的分配粒度。4.3 緩沖區(qū)滿(mǎn)了怎么辦buffer.memory 的取舍buffer.memory默認(rèn) 32MB在低吞吐場(chǎng)景夠用但在高吞吐場(chǎng)景經(jīng)常成為瓶頸。我遇到過(guò)一次線(xiàn)上問(wèn)題生產(chǎn)者 QPS 很高但 Sender 線(xiàn)程因?yàn)槟硞€(gè) broker 響應(yīng)慢而卡住導(dǎo)致緩沖區(qū)迅速填滿(mǎn)用戶(hù)線(xiàn)程大面積阻塞接口超時(shí)率飆升。解決辦法有兩個(gè)方向。一是加大 buffer.memory給 Sender 線(xiàn)程更多緩沖空間扛住短時(shí)的發(fā)送延遲。二是降低單條消息的阻塞敏感度比如把max.block.ms調(diào)小讓send()快速失敗由業(yè)務(wù)層做降級(jí)。前者治標(biāo)后者治本實(shí)際中往往兩者結(jié)合。但要注意buffer.memory是每個(gè)生產(chǎn)者實(shí)例的內(nèi)存不是全局的。如果你有 100 個(gè)生產(chǎn)者實(shí)例每個(gè) 32MB那就是 3.2GB 內(nèi)存。調(diào)大之前先算算總賬。4.4 冪等與重試Sender 線(xiàn)程的可靠性保障Sender 線(xiàn)程在遇到可重試錯(cuò)誤時(shí)會(huì)把批次放回隊(duì)列重發(fā)。如果沒(méi)有冪等保證重發(fā)可能導(dǎo)致 broker 端出現(xiàn)重復(fù)消息。開(kāi)啟enable.idempotencetrue后每個(gè)生產(chǎn)者會(huì)被分配一個(gè) PID每條消息帶一個(gè)序列號(hào)broker 會(huì)據(jù)此去重。開(kāi)啟冪等后acks會(huì)被強(qiáng)制設(shè)為allretries會(huì)被設(shè)為MAXmax.in.flight.requests.per.connection會(huì)被限制在 5 以?xún)?nèi)保證順序。這些約束其實(shí)都是合理的默認(rèn)值所以我一般建議直接開(kāi)冪等省心。注意冪等只在單個(gè)生產(chǎn)者會(huì)話(huà)內(nèi)有效。如果生產(chǎn)者重啟PID 會(huì)變跨會(huì)話(huà)的去重就失效了。要跨會(huì)話(huà)精確一次需要配合事務(wù)。5. 常見(jiàn)問(wèn)題與排查技巧實(shí)錄5.1 消息發(fā)送慢怎么定位是不是 Sender 線(xiàn)程的問(wèn)題排查發(fā)送慢我一般按這個(gè)順序看看record-queue-time-avg指標(biāo)這是消息在緩沖區(qū)里等待的平均時(shí)間。如果這個(gè)值很高說(shuō)明 Sender 線(xiàn)程發(fā)送不及時(shí)問(wèn)題在發(fā)送側(cè)?？磖equest-latency-avg這是網(wǎng)絡(luò)請(qǐng)求的平均往返時(shí)間。如果很高說(shuō)明網(wǎng)絡(luò)或 broker 端慢。看batch-size-avg平均批次大小。如果遠(yuǎn)小于batch.size說(shuō)明攢批效果差可能是linger.ms太小或流量太稀疏?？碽uffer-available-bytes如果經(jīng)常接近 0說(shuō)明緩沖區(qū)不夠用。這幾個(gè)指標(biāo)在 JMX 里都能拿到接上監(jiān)控面板一目了然。5.2 常見(jiàn)異常速查表異常常見(jiàn)原因解決方向TimeoutException: Failed to allocate memory緩沖區(qū)滿(mǎn)Sender 發(fā)送慢加大 buffer.memory優(yōu)化發(fā)送效率RecordTooLargeException單條消息超過(guò) max.request.size調(diào)大 max.request.size 或拆分消息NotLeaderForPartitionException分區(qū) leader 遷移正常重試即可檢查集群穩(wěn)定性NetworkException網(wǎng)絡(luò)抖動(dòng)或 broker 不可達(dá)檢查網(wǎng)絡(luò)適當(dāng)調(diào)大重試OutOfOrderSequenceException冪等序列號(hào)錯(cuò)亂通常意味著消息丟失需排查UnknownTopicOrPartitionExceptiontopic 不存在或元數(shù)據(jù)過(guò)期檢查 topic觸發(fā)元數(shù)據(jù)刷新5.3 踩過(guò)的坑linger.ms 設(shè)太大導(dǎo)致低峰期延遲有一次我把linger.ms設(shè)成了 100ms壓測(cè)時(shí)吞吐很漂亮。結(jié)果上線(xiàn)后低峰期用戶(hù)反饋接口偶爾卡頓。排查發(fā)現(xiàn)低峰期消息稀疏一個(gè)批次要等滿(mǎn) 100ms 才發(fā)導(dǎo)致每條消息都多了最多 100ms 延遲。后來(lái)改成動(dòng)態(tài)策略高峰期用大 linger低峰期用小 linger。但 Kafka 客戶(hù)端不支持動(dòng)態(tài)改最后是通過(guò)業(yè)務(wù)層根據(jù)流量切換兩套生產(chǎn)者配置實(shí)現(xiàn)的。這個(gè)坑告訴我參數(shù)調(diào)優(yōu)一定要考慮流量的波峰波谷不能只看壓測(cè)數(shù)據(jù)。5.4 踩過(guò)的坑max.in.flight 與順序性的關(guān)系max.in.flight.requests.per.connection控制單個(gè)連接上未確認(rèn)的請(qǐng)求數(shù)。設(shè)大了能提升吞吐但如果沒(méi)開(kāi)冪等一旦發(fā)生重試后面的請(qǐng)求可能先成功導(dǎo)致消息亂序。我見(jiàn)過(guò)一個(gè)團(tuán)隊(duì)為了追求吞吐把它設(shè)成 10又沒(méi)開(kāi)冪等結(jié)果下游消費(fèi)到的消息順序錯(cuò)亂業(yè)務(wù)邏輯出錯(cuò)。后來(lái)開(kāi)了冪等把它降回 5問(wèn)題解決。所以記住要順序又要重試就開(kāi)冪等并把 in-flight 控制在 5 以?xún)?nèi)。5.5 一個(gè)實(shí)用技巧用回調(diào)監(jiān)控 Sender 線(xiàn)程的健康度send()方法支持傳一個(gè)Callback消息發(fā)送成功或失敗時(shí)會(huì)回調(diào)。我習(xí)慣在回調(diào)里統(tǒng)計(jì)成功率和延遲分布尤其是失敗原因。如果發(fā)現(xiàn)大量TimeoutException說(shuō)明 Sender 線(xiàn)程壓力大如果大量RecordTooLargeException說(shuō)明消息體需要治理。這個(gè)回調(diào)是在 Sender 線(xiàn)程里執(zhí)行的所以回調(diào)里不要做耗時(shí)操作否則會(huì)拖慢整個(gè)發(fā)送循環(huán)。我一般只做計(jì)數(shù)和打點(diǎn)復(fù)雜的處理丟到單獨(dú)的線(xiàn)程池。6. 從源碼視角再看一眼 Sender 的循環(huán)如果你愿意翻一翻客戶(hù)端源碼會(huì)發(fā)現(xiàn)Sender.run()的主循環(huán)其實(shí)非常精煉。它每一輪都會(huì)先判斷生產(chǎn)者是否已關(guān)閉然后依次執(zhí)行sendProducerData()和client.poll()。sendProducerData()里封裝了前面說(shuō)的 ready、drain、構(gòu)造請(qǐng)求、發(fā)送這一整套邏輯。有意思的是client.poll()的阻塞時(shí)間不是固定的而是根據(jù)accumulator里最近一個(gè)批次的就緒時(shí)間來(lái)動(dòng)態(tài)計(jì)算的。如果下一個(gè)批次還有 3ms 就緒poll 最多阻塞 3ms 就返回保證不會(huì)錯(cuò)過(guò)發(fā)送時(shí)機(jī)。這種設(shè)計(jì)讓 Sender 線(xiàn)程既不空轉(zhuǎn)燒 CPU又能及時(shí)響應(yīng)。另外Sender 線(xiàn)程還負(fù)責(zé)元數(shù)據(jù)的定期刷新。metadata.max.age.ms默認(rèn) 5 分鐘到期后會(huì)強(qiáng)制刷新一次元數(shù)據(jù)即使沒(méi)有明顯的 leader 變更。這是為了防止客戶(hù)端長(zhǎng)時(shí)間使用過(guò)期的集群視圖。理解這些細(xì)節(jié)之后再回頭看那些調(diào)優(yōu)參數(shù)就不再是孤立的數(shù)字而是和 Sender 線(xiàn)程的具體行為一一對(duì)應(yīng)。比如你知道linger.ms影響的是 ready 判斷batch.size影響的是批次何時(shí)滿(mǎn)buffer.memory影響的是用戶(hù)線(xiàn)程會(huì)不會(huì)阻塞調(diào)起來(lái)就有方向了。我個(gè)人在實(shí)際操作中的體會(huì)是Kafka 生產(chǎn)者的調(diào)優(yōu)從來(lái)不是調(diào)一個(gè)參數(shù)而是理解 Sender 線(xiàn)程這個(gè)搬運(yùn)工的工作節(jié)奏然后讓各個(gè)參數(shù)配合它的節(jié)奏。批次攢得太小它跑得勤但每次搬得少批次攢得太大它搬得爽但消息等得久。找到那個(gè)平衡點(diǎn)才是真正的功夫。最后再分享一個(gè)小技巧壓測(cè)時(shí)別只看總吞吐一定要同時(shí)看 P99 延遲和緩沖區(qū)水位這三個(gè)指標(biāo)一起看才能判斷配置是不是真的健康。