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