為什么 ready 了還不發(fā)上一篇講了 batch 什么時(shí)候變 READY。但很多人遇到過這種情況監(jiān)控顯示 batch 已經(jīng) ready 了消息卻沒發(fā)出去延遲持續(xù)走高。原因很簡單accumulator.ready()返回的readyNodes只是這個(gè) Broker 上有 ready 的 batch不代表這個(gè) Broker 現(xiàn)在可以接收消息。Sender 拿到 readyNodes 后還要逐個(gè)節(jié)點(diǎn)調(diào)用client.ready(node)做三層過濾通過了才會真正 drain 發(fā)送。01 三層過濾全景從 readyNodes 到可發(fā)送節(jié)點(diǎn)要經(jīng)過三層過濾第一層連接狀態(tài)過濾。調(diào)用NetworkClient.isReady(node)檢查 TCP 連接是否已建立、SASL 認(rèn)證是否完成、握手是否完成。只有連接狀態(tài)為 READY 才通過。未連接的節(jié)點(diǎn)會調(diào)用initiateConnect()發(fā)起連接連接中的節(jié)點(diǎn)本輪跳過。第二層in-flight 過濾。調(diào)用inFlightRequests.canSendMore(node.id)檢查該節(jié)點(diǎn)當(dāng)前未確認(rèn)的請求數(shù)是否小于max.in.flight.requests.per.connection默認(rèn) 5。滿了說明 Broker 處理不過來繼續(xù)發(fā)只會導(dǎo)致請求排隊(duì)超時(shí)。第三層元數(shù)據(jù)更新過濾。檢查metadataUpdater.isUpdateDue(nowMs)如果元數(shù)據(jù)過期需要刷新本輪優(yōu)先發(fā) MetadataRequest暫停普通消息發(fā)送——避免用過期元數(shù)據(jù)把消息發(fā)到錯(cuò)誤的 Broker。三層都通過的節(jié)點(diǎn)才加入可發(fā)送節(jié)點(diǎn)集傳給accumulator.drain()拉取批次。02 NetworkClient.ready() 的判斷邏輯client.ready(node, nowMs)內(nèi)部是一個(gè)與邏輯三個(gè)條件同時(shí)滿足才返回 truepublicbooleanready(Nodenode,longnowMs){if(node.isEmpty())returnfalse;// 條件1連接就緒if(!connectionStates.isReady(node.id())){if(connectionStates.canConnect(node.id(),nowMs))initiateConnect(node,nowMs);returnfalse;}// 條件2in-flight 未滿if(!inFlightRequests.canSendMore(node.id()))returnfalse;// 條件3元數(shù)據(jù)未在更新if(metadataUpdater.isUpdateDue(nowMs))returnfalse;returntrue;}連接狀態(tài)機(jī)有 4 個(gè)狀態(tài)NOT_CONNECTED → CONNECTING → AUTHENTICATING → READY。只有 READY 狀態(tài)isReady()才返回 true。首次發(fā)送時(shí)要經(jīng)過 TCP 三次握手 SASL 認(rèn)證 TLS 握手如果啟用這就是為什么 Producer 啟動后第一條消息延遲特別高。in-flight 計(jì)數(shù)是每個(gè) Broker 連接獨(dú)立維護(hù)的。發(fā)一個(gè) ProduceRequest 計(jì)數(shù) 1收到 ACK 計(jì)數(shù) -1。默認(rèn)上限 5調(diào)大能提高吞吐但可能導(dǎo)致 Broker 壓力過大調(diào)小如 1能保證嚴(yán)格順序但犧牲吞吐。冪等生產(chǎn)者enable.idempotencetrue會自動限制為 ≤5。03 Sender.runOnce 中的完整流程把過濾邏輯放回 Sender 的主循環(huán)完整流程是這樣的voidrunOnce(longnowMs,longmetadataTimeoutMs){Clusterclustermetadata.fetch();// 1. 獲取有 ready batch 的節(jié)點(diǎn)RecordAccumulator.ReadyCheckResultresultaccumulator.ready(cluster,nowMs);// 2. 逐個(gè)節(jié)點(diǎn)過濾未通過的從集合中移除for(Nodenode:result.readyNodes){if(!client.ready(node,nowMs))result.readyNodes.remove(node);}// 3. 用過濾后的節(jié)點(diǎn)拉取批次MapInteger,ListProducerBatchbatchesaccumulator.drain(cluster,result.readyNodes,maxRequestSize,nowMs);// 4. 構(gòu)建請求并發(fā)送sendProduceRequests(batches,nowMs);}注意一個(gè)細(xì)節(jié)過濾是直接修改 readyNodes 集合remove 未通過的節(jié)點(diǎn)不是創(chuàng)建新集合。所以傳給 drain 的 readyNodes 已經(jīng)是過濾后的結(jié)果。04 為什么 ready 了但沒發(fā)出去4 種場景場景1連接未建立。首次發(fā)送或連接斷開重連時(shí)TCP 握手 認(rèn)證需要時(shí)間。監(jiān)控connection-count和connection-creation-total如果連接數(shù)持續(xù)增長說明在頻繁重連。解決Producer 啟動后預(yù)熱發(fā)一條空消息觸發(fā)連接建立或者接受首次延遲。場景2in-flight 已滿。這是最常見的吞吐瓶頸。監(jiān)控in-flight-requests-count如果持續(xù)接近max.in.flight默認(rèn)5說明 Broker 處理不過來。這不是 Producer 的問題是 Broker 端瓶頸——排查磁盤 IO、CPU、網(wǎng)絡(luò)。調(diào)大max.in.flight可能暫時(shí)緩解但根本解決要優(yōu)化 Broker。場景3元數(shù)據(jù)更新中。metadata.max.age.ms到期默認(rèn)5分鐘或 Broker 變化時(shí)觸發(fā)元數(shù)據(jù)刷新期間普通消息暫停。表現(xiàn)為間歇性延遲尖刺持續(xù)幾百 ms。這是正常保護(hù)機(jī)制更新完自動恢復(fù)。如果頻繁更新排查 Broker 穩(wěn)定性是否有頻繁的 Leader 切換。場景4Broker 不可用 / Leader 切換。Broker 宕機(jī)觸發(fā) Leader 重選舉舊 Leader 不再是 Leader發(fā)送會報(bào)NOT_LEADER_OR_FOLLOWER。元數(shù)據(jù)刷新后重新路由到新 Leader重試發(fā)送。retries足夠大時(shí)自動恢復(fù)超過delivery.timeout.ms則報(bào)錯(cuò)。寫在最后理解這三層過濾后“batch ready 了但沒發(fā)出去就不再是玄學(xué)。核心思路ready 只是可以發(fā)”真正發(fā)送還要過連接、in-flight、元數(shù)據(jù)三道關(guān)。排查時(shí)按順序檢查三個(gè)監(jiān)控指標(biāo)connection-count連接是否正常、in-flight-requests-count是否接近上限、metadata-age是否在更新。99% 的問題都能在這三個(gè)指標(biāo)里定位到。