1. 從“消費(fèi)”這件事說起Consumer到底在解決什么問題很多人學(xué)KafkaProducer那關(guān)過得挺順——發(fā)消息嘛指定個Topic把數(shù)據(jù)塞進(jìn)去就完事了。到了Consumer就開始犯迷糊為什么我起了三個消費(fèi)者只有一個在干活為什么重啟之后消息又從頭消費(fèi)了一遍為什么明明消費(fèi)了Lag卻一直不降這些問題的根子都在于沒搞清楚Consumer的設(shè)計定位。Consumer不是“被動接收消息的管道”它是一個主動拉取、自主管理位點(diǎn)、參與組內(nèi)協(xié)調(diào)的復(fù)雜角色。你可以把它想象成一個快遞站的取件員不是快遞到了就自動推到你手上而是你拿著取件碼主動去站點(diǎn)把包裹拉回來同時還要記錄“我上次取到哪一件了”免得重復(fù)取或者漏取。Kafka的Consumer有幾個核心特征理解了它們后面所有操作都不會迷路Pull模式而非Push模式Consumer自己決定什么時候去Broker拉數(shù)據(jù)拉多少。這跟很多消息隊列的Push模式完全不同。好處是Consumer不會被壓垮——我處理得慢就少拉點(diǎn)處理得快就多拉點(diǎn)主動權(quán)在我手里。位點(diǎn)Offset由Consumer自己維護(hù)Broker不關(guān)心你消費(fèi)到哪了它只負(fù)責(zé)存消息。你消費(fèi)到哪是你自己的事。這就解釋了為什么“重啟后重復(fù)消費(fèi)”——因為你的位點(diǎn)沒提交成功或者提交到了一個錯誤的位置。Consumer Group是邏輯上的“一個消費(fèi)者”同一個組內(nèi)的多個Consumer實例共同瓜分所有分區(qū)的消息。一個分區(qū)在同一時刻只能被組內(nèi)一個Consumer消費(fèi)。這就是為什么“起了三個消費(fèi)者只有一個在干活”——如果你的Topic只有一個分區(qū)那確實只能有一個消費(fèi)者分到活。我見過太多人把Consumer Group當(dāng)成“負(fù)載均衡器”來理解結(jié)果在分區(qū)數(shù)不夠的時候百思不得其解。記住一句話并行度的上限是分區(qū)數(shù)不是消費(fèi)者數(shù)。2. 消費(fèi)位點(diǎn)提交自動提交的“舒適區(qū)”與“雷區(qū)”位點(diǎn)提交是Consumer里最容易出問題的地方?jīng)]有之一。很多線上事故——重復(fù)消費(fèi)、消息丟失——追到根上都是位點(diǎn)提交的時機(jī)和方式?jīng)]搞對。2.1 自動提交到底在什么時候提交先看一段最常見的配置enable.auto.committrue auto.commit.interval.ms5000這兩行配置的意思是Consumer在后臺每隔5秒自動提交一次當(dāng)前已拉取消息的最大位點(diǎn)。注意關(guān)鍵詞——已拉取不是已處理完。這就埋了一個雷。假設(shè)你的Consumer拉了一批消息位點(diǎn)100到200交給業(yè)務(wù)線程去處理。業(yè)務(wù)線程處理到第150條的時候卡住了花了10秒。而這10秒內(nèi)后臺的自動提交線程已經(jīng)跑了兩次把位點(diǎn)提交到了200。然后Consumer崩了。重啟之后從200開始消費(fèi)第151到200條消息——丟了。反過來如果自動提交間隔設(shè)得特別短比如100毫秒那又會出現(xiàn)另一個問題你剛拉下來還沒處理完位點(diǎn)就提交了一旦處理失敗消息同樣丟失。而且頻繁提交位點(diǎn)會給Broker帶來額外壓力。所以自動提交的“舒適區(qū)”是你對消息丟失有一定容忍度且處理邏輯足夠快、足夠穩(wěn)定。比如日志采集、監(jiān)控指標(biāo)上報這類場景丟幾條無所謂用自動提交省心省力。2.2 手動提交的兩種姿勢同步與異步一旦業(yè)務(wù)要求“至少一次”甚至“精確一次”就必須關(guān)掉自動提交自己控制時機(jī)enable.auto.commitfalse然后有兩種提交方式同步提交consumer.commitSync()。調(diào)用后會阻塞直到Broker確認(rèn)收到提交請求。優(yōu)點(diǎn)是可靠提交失敗會拋異常你能感知到缺點(diǎn)是吞吐量受影響每次提交都要等一個網(wǎng)絡(luò)往返。異步提交consumer.commitAsync()。調(diào)用后立即返回不阻塞?？梢詡饕粋€回調(diào)在提交成功或失敗時做處理。優(yōu)點(diǎn)是快缺點(diǎn)是失敗時不會自動重試——因為如果異步提交失敗后你重試可能提交的是一個更舊的位點(diǎn)反而造成重復(fù)消費(fèi)。實際項目中我通常這樣組合使用try { while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(100)); for (ConsumerRecordString, String record : records) { process(record); // 業(yè)務(wù)處理 } consumer.commitAsync(); // 正常循環(huán)用異步追求吞吐 } } catch (Exception e) { log.error(消費(fèi)異常, e); } finally { try { consumer.commitSync(); // 退出前用同步確保位點(diǎn)落盤 } finally { consumer.close(); } }這個模式的核心邏輯是日常用異步保吞吐退出用同步保可靠。因為退出時如果異步提交失敗你沒有機(jī)會再重試了所以必須用同步兜底。2.3 位點(diǎn)提交與分區(qū)再均衡的“致命交叉”比自動提交更隱蔽的坑是位點(diǎn)提交和分區(qū)再均衡Rebalance的交互。當(dāng)有新Consumer加入組或者有Consumer離開組或者訂閱的Topic分區(qū)數(shù)發(fā)生變化時就會觸發(fā)Rebalance。Rebalance期間所有Consumer都會停止消費(fèi)等協(xié)調(diào)者重新分配分區(qū)。問題來了如果一個Consumer在Rebalance之前拉了一批消息處理了一部分還沒提交位點(diǎn)Rebalance發(fā)生了這個分區(qū)被分配給了另一個Consumer。新的Consumer從上次提交的位點(diǎn)開始消費(fèi)——那些已經(jīng)處理但沒提交的消息會被重復(fù)消費(fèi)。更嚴(yán)重的情況是Consumer A正在處理分區(qū)P的消息位點(diǎn)還沒提交Rebalance把分區(qū)P分給了Consumer B。Consumer A處理完后提交位點(diǎn)但這個提交是針對分區(qū)P的——而分區(qū)P已經(jīng)不屬于A了。這個提交要么被拒絕要么造成位點(diǎn)混亂。解決辦法是使用ConsumerRebalanceListener在分區(qū)被回收之前提交位點(diǎn)consumer.subscribe(Collections.singletonList(my-topic), new ConsumerRebalanceListener() { Override public void onPartitionsRevoked(CollectionTopicPartition partitions) { // 分區(qū)被回收前同步提交位點(diǎn) consumer.commitSync(); } Override public void onPartitionsAssigned(CollectionTopicPartition partitions) { // 分區(qū)分配后可以從外部存儲恢復(fù)位點(diǎn) for (TopicPartition tp : partitions) { long offset loadOffsetFromDB(tp); if (offset 0) { consumer.seek(tp, offset); } } } });onPartitionsRevoked是在分區(qū)被拿走之前調(diào)用的這是你最后提交位點(diǎn)的機(jī)會。onPartitionsAssigned是在拿到新分區(qū)之后調(diào)用的你可以在這里從數(shù)據(jù)庫或Redis里恢復(fù)位點(diǎn)實現(xiàn)更精確的控制。我踩過的一個坑在onPartitionsRevoked里用異步提交結(jié)果Rebalance太快提交還沒完成分區(qū)就被收走了。后來改成同步提交雖然會稍微延長Rebalance時間但至少不會丟位點(diǎn)。3. Rebalance機(jī)制為什么你的消費(fèi)者組總是“卡住”Rebalance是Kafka Consumer里最讓人頭疼的機(jī)制。它本身是為了實現(xiàn)彈性伸縮和高可用但觸發(fā)條件多、過程復(fù)雜稍不注意就會導(dǎo)致消費(fèi)停滯。3.1 三種觸發(fā)條件與對應(yīng)的排查方向Rebalance的觸發(fā)條件有三類觸發(fā)條件典型場景排查方向組成員數(shù)量變化消費(fèi)者啟動、崩潰、主動退出檢查消費(fèi)者實例是否頻繁重啟訂閱Topic數(shù)量變化使用正則訂閱新建了匹配的Topic檢查訂閱方式是否為正則訂閱Topic分區(qū)數(shù)變化運(yùn)維給Topic增加了分區(qū)檢查是否有分區(qū)變更操作第一類是最常見的。消費(fèi)者崩潰、網(wǎng)絡(luò)抖動導(dǎo)致心跳超時、GC停頓過長導(dǎo)致心跳線程被掛起都會讓協(xié)調(diào)者認(rèn)為這個消費(fèi)者“死了”從而觸發(fā)Rebalance。這里有個關(guān)鍵參數(shù)session.timeout.ms默認(rèn)45秒不同版本有差異。消費(fèi)者需要在這個時間內(nèi)發(fā)送心跳給協(xié)調(diào)者。如果超時協(xié)調(diào)者就認(rèn)為消費(fèi)者下線。還有一個參數(shù)max.poll.interval.ms默認(rèn)5分鐘。這是兩次poll()調(diào)用的最大間隔。如果你的業(yè)務(wù)處理邏輯太慢兩次poll之間超過了5分鐘消費(fèi)者會被認(rèn)為“卡住了”主動離開組觸發(fā)Rebalance。這兩個參數(shù)的區(qū)別很重要session.timeout.ms管的是心跳線程max.poll.interval.ms管的是處理線程。心跳線程是后臺獨(dú)立運(yùn)行的只要進(jìn)程活著就會發(fā)心跳。但如果處理線程卡住poll間隔超時消費(fèi)者會主動發(fā)起離開組的請求。實際排查經(jīng)驗如果日志里看到“Member ... sending LeaveGroup request due to consumer poll timeout”那就是max.poll.interval.ms超了需要優(yōu)化處理邏輯或者調(diào)大這個參數(shù)。如果是“Disconnecting from node ... due to request timeout”那可能是網(wǎng)絡(luò)問題。3.2 分區(qū)分配策略Range、RoundRobin與StickyRebalance的核心動作是“把分區(qū)重新分配給消費(fèi)者”。分配策略決定了怎么分。Range策略默認(rèn)按Topic維度把每個Topic的分區(qū)按范圍分給消費(fèi)者。假設(shè)Topic有7個分區(qū)組內(nèi)有3個消費(fèi)者那么消費(fèi)者A分到分區(qū)0-2B分到3-5C分到6。問題在于如果組內(nèi)訂閱了多個Topic每個Topic都這樣分會導(dǎo)致前面的消費(fèi)者總是多分到分區(qū)負(fù)載不均。RoundRobin策略把所有訂閱的Topic的所有分區(qū)放在一起輪詢分配給消費(fèi)者。這樣負(fù)載更均衡但有一個前提——組內(nèi)所有消費(fèi)者訂閱的Topic必須完全一致否則會出現(xiàn)分區(qū)分配錯亂。Sticky策略推薦在Rebalance時盡量保留原有的分配方案只把需要變動的分區(qū)重新分配。比如消費(fèi)者C崩潰了Sticky策略會把C原來負(fù)責(zé)的分區(qū)盡量均勻地分給A和B而不是把所有分區(qū)打散重來。這大大減少了Rebalance的開銷。配置方式partition.assignment.strategyorg.apache.kafka.clients.consumer.StickyAssignorSticky策略還有一個好處它支持“ cooperative rebalance”協(xié)作再均衡。傳統(tǒng)的Rebalance是“全部停止重新分配”而協(xié)作再均衡允許消費(fèi)者在Rebalance期間繼續(xù)消費(fèi)那些不會被回收的分區(qū)只有需要變動的分區(qū)才暫停。這進(jìn)一步減少了消費(fèi)停滯的時間。3.3 如何避免“Rebalance風(fēng)暴”Rebalance本身不可怕可怕的是頻繁Rebalance。每次Rebalance都會導(dǎo)致消費(fèi)暫停如果頻繁發(fā)生消費(fèi)Lag會飆升業(yè)務(wù)會受影響。避免Rebalance風(fēng)暴的幾個實操要點(diǎn)合理設(shè)置session.timeout.ms設(shè)太小網(wǎng)絡(luò)稍微抖動就觸發(fā)Rebalance設(shè)太大消費(fèi)者真掛了要很久才能被發(fā)現(xiàn)。一般建議6-10秒根據(jù)網(wǎng)絡(luò)質(zhì)量調(diào)整。合理設(shè)置max.poll.interval.ms根據(jù)業(yè)務(wù)處理的最大耗時來設(shè)。如果一批消息最多處理2分鐘那就設(shè)成3-5分鐘留足余量?？刂苖ax.poll.records每次poll拉取的最大消息數(shù)。拉太多處理時間就長容易超時。一般設(shè)500-1000根據(jù)單條處理耗時調(diào)整。避免消費(fèi)者頻繁重啟如果是容器化部署檢查健康檢查是否過于激進(jìn)導(dǎo)致消費(fèi)者被反復(fù)殺掉重啟。使用靜態(tài)成員Static MembershipKafka 2.3支持給消費(fèi)者配置group.instance.id這樣消費(fèi)者重啟時在session.timeout.ms內(nèi)重新加入組不會觸發(fā)Rebalance。group.instance.idconsumer-instance-1 session.timeout.ms30000靜態(tài)成員特別適合Kubernetes環(huán)境下的消費(fèi)者部署。Pod重啟時只要在超時時間內(nèi)回來協(xié)調(diào)者就認(rèn)為還是原來的成員不觸發(fā)Rebalance。4. 消費(fèi)位移與Lag監(jiān)控怎么知道消費(fèi)者“健康不健康”Consumer跑起來容易跑得健康難。很多團(tuán)隊是等到業(yè)務(wù)方反饋“數(shù)據(jù)怎么沒更新”才發(fā)現(xiàn)消費(fèi)者掛了。其實通過Lag監(jiān)控完全可以提前發(fā)現(xiàn)問題。4.1 Lag到底是什么怎么算Lag Log End Offset (LEO) - Current Offset。LEO是分區(qū)最新消息的位點(diǎn)Current Offset是消費(fèi)者當(dāng)前提交的位點(diǎn)。兩者之差就是還沒消費(fèi)的消息數(shù)。Lag為0說明消費(fèi)者追上了最新消息。Lag持續(xù)增長說明消費(fèi)速度跟不上生產(chǎn)速度。Lag突然飆升可能是消費(fèi)者掛了或者Rebalance了。獲取Lag的方式有幾種命令行工具kafka-consumer-groups.sh --bootstrap-server localhost:9092 \ --describe --group my-consumer-group輸出里會顯示每個分區(qū)的CURRENT-OFFSET、LOG-END-OFFSET和LAG。Java APIMapTopicPartition, OffsetAndMetadata committed consumer.committed(partitions); for (TopicPartition tp : partitions) { long leo consumer.endOffsets(Collections.singleton(tp)).get(tp); long current committed.get(tp).offset(); long lag leo - current; System.out.println(tp lag: lag); }JMX指標(biāo)Kafka Consumer會暴露records-lag-max等JMX指標(biāo)可以直接接入監(jiān)控系統(tǒng)。4.2 用Lag做告警的幾種策略Lag監(jiān)控不是簡單設(shè)個閾值就完事了。不同業(yè)務(wù)對Lag的容忍度完全不同。絕對閾值告警Lag超過某個值就告警。適合消息量穩(wěn)定的場景。比如Lag超過10萬就告警。增長率告警Lag的增長速度超過某個速率就告警。適合消息量波動大的場景。比如Lag在5分鐘內(nèi)從0漲到5萬說明消費(fèi)完全停滯了。消費(fèi)時間告警根據(jù)Lag和消費(fèi)速率估算“還需要多久才能追平”。如果估算時間超過業(yè)務(wù)容忍度就告警。我一般會組合使用絕對閾值做兜底增長率做靈敏告警。同時把Lag的歷史曲線畫出來方便排查是“緩慢增長”還是“突然飆升”。一個容易忽略的點(diǎn)Lag為0不一定代表健康。如果消費(fèi)者掛了但位點(diǎn)已經(jīng)提交到了LEOLag也是0。所以還要結(jié)合消費(fèi)者的存活狀態(tài)一起判斷。4.3 消費(fèi)停滯的常見原因排查清單當(dāng)發(fā)現(xiàn)Lag持續(xù)增長時按這個順序排查消費(fèi)者進(jìn)程是否存活檢查進(jìn)程、容器狀態(tài)、日志有無異常退出。是否在頻繁Rebalance查看消費(fèi)者日志有無“Rebalance”相關(guān)記錄檢查session.timeout.ms和max.poll.interval.ms配置。業(yè)務(wù)處理是否變慢檢查下游依賴數(shù)據(jù)庫、外部接口的響應(yīng)時間看是否有慢查詢或超時。分區(qū)數(shù)是否足夠如果消費(fèi)者數(shù)已經(jīng)等于分區(qū)數(shù)再加消費(fèi)者也沒用需要增加分區(qū)。是否有消息堆積導(dǎo)致GC頻繁大量消息積壓會導(dǎo)致內(nèi)存壓力大GC頻繁進(jìn)一步拖慢處理速度。5. 多線程消費(fèi)提升吞吐的正確打開方式單線程Consumer的吞吐量是有上限的。當(dāng)Lag持續(xù)增長加消費(fèi)者又受限于分區(qū)數(shù)時就需要考慮多線程消費(fèi)。5.1 為什么Kafka Consumer不是線程安全的Kafka Consumer設(shè)計上就是單線程的。它內(nèi)部維護(hù)了網(wǎng)絡(luò)連接、位點(diǎn)狀態(tài)、分區(qū)分配等復(fù)雜狀態(tài)多個線程同時操作會導(dǎo)致狀態(tài)混亂。官方明確說了Consumer不是線程安全的多線程共享一個Consumer實例會拋ConcurrentModificationException。所以多線程消費(fèi)的模式要么是“每個線程一個Consumer”要么是“一個Consumer拉取多個Worker處理”。5.2 模式一每線程一個Consumer這是最簡單的多線程模式。每個線程創(chuàng)建自己的Consumer實例訂閱相同的Topic但屬于不同的Consumer Group或者同一個Group但線程數(shù)不超過分區(qū)數(shù)。int threadNum 4; for (int i 0; i threadNum; i) { new Thread(() - { KafkaConsumerString, String consumer new KafkaConsumer(props); consumer.subscribe(Collections.singletonList(my-topic)); while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(100)); for (ConsumerRecordString, String record : records) { process(record); } consumer.commitSync(); } }).start(); }這種模式的優(yōu)點(diǎn)是簡單每個Consumer獨(dú)立管理位點(diǎn)互不干擾。缺點(diǎn)是Consumer實例多了網(wǎng)絡(luò)連接和內(nèi)存開銷也大。而且如果線程數(shù)超過分區(qū)數(shù)多出來的線程分不到分區(qū)白白浪費(fèi)。5.3 模式二一個Consumer拉取Worker池處理這種模式用一個Consumer負(fù)責(zé)拉取消息然后把消息提交給線程池處理。位點(diǎn)提交需要等Worker處理完再統(tǒng)一提交。KafkaConsumerString, String consumer new KafkaConsumer(props); consumer.subscribe(Collections.singletonList(my-topic)); ExecutorService executor Executors.newFixedThreadPool(8); while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(100)); MapTopicPartition, OffsetAndMetadata currentOffsets new HashMap(); for (ConsumerRecordString, String record : records) { executor.submit(() - { process(record); // 記錄處理完成的位點(diǎn) synchronized (currentOffsets) { currentOffsets.put( new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() 1) ); } }); } // 等待所有任務(wù)完成后再提交位點(diǎn) // 實際實現(xiàn)需要更精細(xì)的同步控制 consumer.commitSync(currentOffsets); }這種模式的坑在于位點(diǎn)提交的時機(jī)。如果不等Worker處理完就提交會丟消息如果等所有Worker處理完再提交吞吐量又上不去。實際項目中通常需要引入一個“滑動窗口”機(jī)制跟蹤每個分區(qū)的處理進(jìn)度按分區(qū)粒度提交位點(diǎn)。我個人更傾向于模式一除非單線程處理確實扛不住。模式二的復(fù)雜度高位點(diǎn)管理容易出錯而且一旦某個Worker卡住整個消費(fèi)都會受影響。5.4 多線程消費(fèi)的位點(diǎn)管理要點(diǎn)無論哪種模式多線程下的位點(diǎn)管理都要注意位點(diǎn)提交必須串行化多個線程同時提交位點(diǎn)會導(dǎo)致位點(diǎn)跳躍或回退。要么加鎖要么只由一個線程負(fù)責(zé)提交。按分區(qū)粒度跟蹤進(jìn)度不同分區(qū)的消費(fèi)進(jìn)度不同不能用一個全局位點(diǎn)來代表所有分區(qū)。處理失敗要有補(bǔ)償機(jī)制Worker處理失敗的消息不能簡單跳過要記錄到重試隊列或死信隊列。6. 消費(fèi)端冪等重復(fù)消費(fèi)不可怕重復(fù)處理才可怕Kafka只能保證“至少一次”消費(fèi)語義在手動提交且處理成功后再提交的情況下。這意味著消息重復(fù)是常態(tài)不是異常。網(wǎng)絡(luò)抖動、Rebalance、消費(fèi)者重啟都可能導(dǎo)致重復(fù)消費(fèi)。所以消費(fèi)端必須做冪等。冪等的核心思想是同一條消息處理多次結(jié)果和處理一次一樣。6.1 用唯一ID做去重最常見的方案是給每條消息分配一個全局唯一的業(yè)務(wù)ID處理前先查這個ID是否已經(jīng)處理過。// 偽代碼 String msgId record.key(); // 假設(shè)key是唯一ID if (redis.setnx(consumed: msgId, 1) 1) { // 設(shè)置成功說明沒處理過 process(record); redis.expire(consumed: msgId, 24 * 3600); // 設(shè)置過期時間 } else { // 已經(jīng)處理過跳過 log.info(重復(fù)消息跳過: {}, msgId); }這個方案的關(guān)鍵是setnx的原子性以及過期時間的設(shè)置。過期時間太短可能還沒過重復(fù)窗口就失效了太長Redis內(nèi)存壓力大。一般根據(jù)業(yè)務(wù)的重試窗口來定24小時是個比較安全的默認(rèn)值。6.2 數(shù)據(jù)庫唯一約束兜底如果業(yè)務(wù)操作是寫數(shù)據(jù)庫可以利用數(shù)據(jù)庫的唯一索引來做冪等。比如訂單創(chuàng)建用訂單號做唯一索引重復(fù)插入會報唯一鍵沖突捕獲這個異常即可。INSERT INTO orders (order_id, user_id, amount) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE update_time NOW();這種方式的好處是不依賴外部組件數(shù)據(jù)庫自身就能保證。缺點(diǎn)是只適用于寫數(shù)據(jù)庫的場景而且如果業(yè)務(wù)邏輯復(fù)雜比如涉及多個表的寫入單靠唯一索引不夠。6.3 狀態(tài)機(jī)冪等對于有狀態(tài)流轉(zhuǎn)的業(yè)務(wù)可以用狀態(tài)機(jī)來做冪等。比如訂單狀態(tài)從“待支付”到“已支付”如果已經(jīng)是“已支付”狀態(tài)再次收到支付消息就忽略。// 偽代碼 Order order orderDao.selectById(orderId); if (order.getStatus() OrderStatus.PAID) { log.info(訂單已支付忽略重復(fù)消息); return; } order.setStatus(OrderStatus.PAID); orderDao.update(order);這種方式需要配合樂觀鎖或悲觀鎖防止并發(fā)狀態(tài)變更。冪等方案的選擇取決于業(yè)務(wù)場景。我的經(jīng)驗是能用數(shù)據(jù)庫唯一約束就用數(shù)據(jù)庫簡單可靠需要跨服務(wù)去重就用Redis狀態(tài)流轉(zhuǎn)類業(yè)務(wù)用狀態(tài)機(jī)最自然。7. 消費(fèi)者配置調(diào)優(yōu)那些默認(rèn)值不一定適合你Kafka Consumer有一堆配置參數(shù)默認(rèn)值是為通用場景設(shè)計的。實際項目中有幾個參數(shù)幾乎必須調(diào)整。7.1 fetch相關(guān)配置控制拉取行為fetch.min.bytes1 fetch.max.wait.ms500 max.partition.fetch.bytes1048576 max.poll.records500fetch.min.bytes一次拉取的最小字節(jié)數(shù)。默認(rèn)1字節(jié)意味著只要有消息就拉。設(shè)大一點(diǎn)比如1MB可以減少網(wǎng)絡(luò)往返但會增加延遲。適合吞吐優(yōu)先的場景。fetch.max.wait.ms如果數(shù)據(jù)不夠fetch.min.bytes最多等多久。默認(rèn)500ms。和fetch.min.bytes配合使用。max.partition.fetch.bytes每個分區(qū)一次拉取的最大字節(jié)數(shù)。默認(rèn)1MB。如果單條消息很大需要調(diào)大這個值否則會一直拉不到完整消息。max.poll.records一次poll返回的最大消息數(shù)。默認(rèn)500。這個值直接影響處理時間進(jìn)而影響max.poll.interval.ms。7.2 心跳與超時配置平衡靈敏度和穩(wěn)定性session.timeout.ms10000 heartbeat.interval.ms3000 max.poll.interval.ms300000session.timeout.ms協(xié)調(diào)者等待消費(fèi)者心跳的超時時間。默認(rèn)45秒新版本或10秒老版本。建議設(shè)6-10秒根據(jù)網(wǎng)絡(luò)質(zhì)量調(diào)整。heartbeat.interval.ms消費(fèi)者發(fā)送心跳的間隔。一般是session.timeout.ms的三分之一。設(shè)太小會增加網(wǎng)絡(luò)開銷設(shè)太大會導(dǎo)致心跳超時被誤判。max.poll.interval.ms兩次poll的最大間隔。默認(rèn)5分鐘。根據(jù)業(yè)務(wù)處理的最大耗時來設(shè)留2-3倍余量。7.3 不同場景的配置模板低延遲場景如實時監(jiān)控fetch.min.bytes1 fetch.max.wait.ms100 max.poll.records100 session.timeout.ms6000 heartbeat.interval.ms2000高吞吐場景如日志采集fetch.min.bytes1048576 fetch.max.wait.ms1000 max.poll.records1000 max.partition.fetch.bytes5242880 session.timeout.ms30000 heartbeat.interval.ms10000大消息場景如文件傳輸max.partition.fetch.bytes10485760 fetch.max.bytes52428800 max.poll.records10調(diào)參這件事沒有銀彈。我的建議是先用默認(rèn)值跑起來觀察Lag和消費(fèi)速率然后針對瓶頸調(diào)整。每次只調(diào)一個參數(shù)觀察效果避免多個參數(shù)一起調(diào)導(dǎo)致無法定位問題。8. 從Consumer視角看消息丟失與重復(fù)的完整鏈路最后把消息丟失和重復(fù)的鏈路完整梳理一遍。理解了這條鏈路排查問題就有章可循。消息丟失的可能環(huán)節(jié)Producer端acks0或acks1時Broker還沒落盤就確認(rèn)Broker掛了消息丟失。Broker端副本數(shù)不夠Leader掛了未同步的副本被選為新Leader消息丟失。Consumer端自動提交位點(diǎn)消息還沒處理完位點(diǎn)就提交了處理失敗后消息丟失。消息重復(fù)的可能環(huán)節(jié)Producer端重試導(dǎo)致同一條消息發(fā)送多次。Consumer端位點(diǎn)提交失敗后重啟從舊位點(diǎn)重新消費(fèi)。Consumer端Rebalance時位點(diǎn)未及時提交新Consumer從舊位點(diǎn)消費(fèi)。Consumer端手動提交時處理成功但提交失敗下次重新消費(fèi)。精確一次Exactly Once的實現(xiàn)思路Kafka 0.11支持事務(wù)和冪等Producer可以實現(xiàn)端到端的精確一次。核心思路是Producer開啟冪等enable.idempotencetrue保證單分區(qū)內(nèi)不重復(fù)。Producer開啟事務(wù)transactional.id保證跨分區(qū)寫入的原子性。Consumer設(shè)置isolation.levelread_committed只讀已提交的事務(wù)消息。消費(fèi)端將位點(diǎn)提交和業(yè)務(wù)處理放在同一個事務(wù)里比如數(shù)據(jù)庫事務(wù)保證“處理成功”和“位點(diǎn)提交”要么都成功要么都失敗。# Producer enable.idempotencetrue transactional.idmy-tx-id # Consumer isolation.levelread_committed enable.auto.commitfalse精確一次的實現(xiàn)成本很高對性能有影響。大多數(shù)業(yè)務(wù)場景“至少一次消費(fèi)端冪等”已經(jīng)足夠。只有在金融、計費(fèi)等對準(zhǔn)確性要求極高的場景才值得上精確一次。我在實際項目中90%的場景都是用“手動提交冪等”的方案。簡單、可靠、性能好。精確一次只在少數(shù)核心鏈路用過配置復(fù)雜排查問題也麻煩。Consumer這塊的內(nèi)容展開講能講一本書。上面這些是我在實際項目中踩過坑、填過坑之后覺得最值得分享的部分。每個參數(shù)、每個配置背后都有它的設(shè)計意圖理解了意圖才能根據(jù)業(yè)務(wù)場景做出正確的取舍。