線上發(fā)版或者業(yè)務(wù)高峰期有時(shí)會(huì)遇到一個(gè)奇怪的現(xiàn)象消費(fèi)端的服務(wù)進(jìn)程沒(méi)掛JVM 內(nèi)存正常日志也在正常輸出顯示業(yè)務(wù)代碼還在繼續(xù)處理數(shù)據(jù)。但 Kafka 服務(wù)端卻判定該消費(fèi)者已失效強(qiáng)行將其踢出消費(fèi)組并觸發(fā)了 Rebalance。這種在進(jìn)程和日志均正常的情況下依然被判定為失效的現(xiàn)象在技術(shù)上通常被稱(chēng)為假死。為什么有心跳也會(huì)被判定死亡Kafka 極早期的版本消費(fèi)端是單線程的。這個(gè)單線程要同時(shí)負(fù)責(zé)兩件事拉取數(shù)據(jù)并執(zhí)行寫(xiě)庫(kù)、打日志等業(yè)務(wù)邏輯還要給服務(wù)端發(fā)心跳包報(bào)告自己還活著。但這個(gè)設(shè)計(jì)在性能上存在缺陷如果業(yè)務(wù)邏輯卡住了比如查數(shù)據(jù)庫(kù)超時(shí)或者 JVM 發(fā)生 Full GC主線程被卡住就無(wú)法按時(shí)發(fā)心跳。服務(wù)端等不到心跳就會(huì)誤認(rèn)為客戶端掛了立刻觸發(fā) Rebalance。后來(lái) Kafka 把拉數(shù)據(jù)和發(fā)心跳徹底拆成了兩個(gè)線程。雙線程架構(gòu)重構(gòu)之后啟動(dòng)一個(gè) Kafka 消費(fèi)者底層實(shí)際上是兩個(gè)線程在協(xié)作。心跳線程只負(fù)責(zé)在后臺(tái)按照固定頻率給 Broker 發(fā)心跳包。只要心跳不斷服務(wù)端就認(rèn)為你的進(jìn)程還活著。它判斷存活的閾值是session.timeout.ms默認(rèn)是 45 秒。另一個(gè)業(yè)務(wù)主線程負(fù)責(zé)在循環(huán)里調(diào)用poll()拉取數(shù)據(jù)并執(zhí)行業(yè)務(wù)邏輯。它判斷是否卡死的閾值是max.poll.interval.ms默認(rèn)是 5 分鐘。這里有個(gè)最容易發(fā)生沖突的地方。假設(shè)你的消費(fèi)者一次性拉了 500 條消息因?yàn)橄掠螌?xiě)庫(kù)慢主業(yè)務(wù)線程處理這 500 條消息一共花了 6 分鐘。在這 6 分鐘里主線程正在處理數(shù)據(jù)并輸出日志。由于這 500 條數(shù)據(jù)還沒(méi)處理完它是沒(méi)辦法回到循環(huán)去調(diào)用下一次poll()的。可后臺(tái)的心跳線程依然在按時(shí)給服務(wù)端發(fā)心跳。但在服務(wù)端看來(lái)距離這臺(tái)機(jī)器上一次調(diào)用poll()已經(jīng)過(guò)去 6 分鐘了超過(guò)了規(guī)定的 5 分鐘死線。服務(wù)端就會(huì)判定這個(gè)消費(fèi)者雖然還有心跳但它已經(jīng)失去了消費(fèi)能力占著分區(qū)卻不拉取新數(shù)據(jù)導(dǎo)致數(shù)據(jù)積壓。為了不影響整個(gè)消費(fèi)組的吞吐必須判定它為假死強(qiáng)行將其踢出消費(fèi)組這就引發(fā)了 Rebalance。怎么避免假死知道是由于業(yè)務(wù)處理太慢導(dǎo)致poll()間隔超時(shí)其實(shí)解決思路非常清晰。1. 減小拉取批次客戶端配置里限制每次poll()拉取的最大消息數(shù)max.poll.records: 50把大批次拆成高頻的小批次。即使單條消息處理需要 100 毫秒50 條也只需要 5 秒鐘。主線程處理完能快速回到下一次poll()循環(huán)避免超出 5 分鐘超時(shí)。2. 調(diào)大間隔參數(shù)根據(jù)最壞的業(yè)務(wù)場(chǎng)景比如下游服務(wù)宕機(jī)、網(wǎng)絡(luò)抖動(dòng)等合理調(diào)大最大 poll 間隔時(shí)間max.poll.interval.ms: 600000不過(guò)這通常要配合max.poll.records一起微調(diào)不建議設(shè)得無(wú)限大。因?yàn)槿绻骶€程真的徹底死鎖卡死了服務(wù)端需要等 10 分鐘才能發(fā)現(xiàn)這期間對(duì)應(yīng)的分區(qū)就會(huì)被其他消費(fèi)者接管造成數(shù)據(jù)消費(fèi)中斷。3. 異步多線程消費(fèi)如果單條業(yè)務(wù)邏輯確實(shí)極其耗時(shí)無(wú)論怎么調(diào)小 records 都不行那就不要在 Kafka 的消費(fèi)主線程里直接干活?？梢灾蛔屩骶€程負(fù)責(zé)拉數(shù)據(jù)拉完立刻丟進(jìn)自定義的 Java 線程池里去并行處理讓主線程秒級(jí)回到下一次poll()。不過(guò)這個(gè)方案會(huì)引入消息丟失和重復(fù)消費(fèi)的風(fēng)險(xiǎn)自動(dòng)提交可能導(dǎo)致消息丟失如果開(kāi)啟了自動(dòng)提交主線程下一次poll()就會(huì)把上一批數(shù)據(jù)對(duì)應(yīng)的 offset 提交。但此時(shí)線程池里可能還有一堆任務(wù)在排隊(duì)。一旦服務(wù)此時(shí)重啟或崩潰這些還沒(méi)來(lái)得及處理的消息就徹底丟失了。手動(dòng)提交的亂序與重復(fù)消費(fèi)要知道線程池里的并發(fā)線程是亂序完成的如果在子線程里直接提交 offset會(huì)發(fā)生 offset 覆蓋導(dǎo)致重復(fù)消費(fèi)。要解決這個(gè)問(wèn)題需要自己設(shè)計(jì)一套滑窗提交機(jī)制或者基于每個(gè) Partition 單獨(dú)配內(nèi)存隊(duì)列開(kāi)發(fā)成本非常高。說(shuō)在最后分布式系統(tǒng)對(duì)活著其實(shí)有兩種定義一個(gè)是進(jìn)程存活代表進(jìn)程在端口通心跳還在跳另一個(gè)是業(yè)務(wù)存活代表業(yè)務(wù)還能響應(yīng)外部輸入推動(dòng)數(shù)據(jù)往下走。寫(xiě)業(yè)務(wù)消費(fèi)邏輯時(shí)不能僅憑進(jìn)程在日志在刷就認(rèn)為消費(fèi)者運(yùn)行正常需要合理評(píng)估m(xù)ax.poll.records和max.poll.interval.ms的配置避免因?yàn)閱未翁幚砗臅r(shí)過(guò)長(zhǎng)導(dǎo)致不必要的 Rebalance。