載均衡機(jī)制深度解析：Producer 發(fā)送與 Consumer 訂閱的全鏈路原理)
RocketMQ 客戶端負(fù)載均衡機(jī)制深度解析Producer 發(fā)送與 Consumer 訂閱的全鏈路原理【免費(fèi)下載鏈接】rocketmqApache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.項(xiàng)目地址: https://gitcode.com/gh_mirrors/ro/rocketmq導(dǎo)讀Apache RocketMQ 的負(fù)載均衡完全在客戶端完成不依賴 Broker 參與調(diào)度這與多數(shù)消息中間件在服務(wù)端分發(fā)消息的設(shè)計(jì)截然不同。本文基于倉(cāng)庫(kù)文檔 Design_LoadBalancing.md系統(tǒng)拆解兩套核心機(jī)制Producer 發(fā)送側(cè)的隊(duì)列選擇與延遲容錯(cuò)MQFaultStrategy以及Consumer 訂閱側(cè)的定時(shí)重平衡RebalanceService RebalanceImpl并對(duì)照 client 模塊 的源碼與測(cè)試說(shuō)明每條關(guān)鍵路徑的底層實(shí)現(xiàn)。讀完本文你將掌握 RocketMQ 客戶端負(fù)載均衡的完整數(shù)據(jù)流、默認(rèn)分配算法細(xì)節(jié)、容錯(cuò)參數(shù)含義以及如何在代碼中調(diào)整分配策略。一、負(fù)載均衡的總體架構(gòu)一切都在客戶端完成在 RocketMQ 中負(fù)載均衡Load Balancing并非由 NameServer 或 Broker 統(tǒng)一計(jì)算而是拆成兩條獨(dú)立的客戶端鏈路Producer 側(cè)負(fù)載均衡發(fā)送消息時(shí)在 Topic 的多個(gè)寫(xiě)隊(duì)列MessageQueue中選擇一個(gè)作為本次發(fā)送目標(biāo)Consumer 側(cè)負(fù)載均衡訂閱消息時(shí)將 Topic 下的一組讀隊(duì)列按消費(fèi)組內(nèi)的消費(fèi)者數(shù)量進(jìn)行劃分決定哪個(gè)消費(fèi)者消費(fèi)哪些隊(duì)列。Broker 和 NameServer 只負(fù)責(zé)提供路由元數(shù)據(jù)Topic 與隊(duì)列的對(duì)應(yīng)關(guān)系和消費(fèi)組心跳信息真正的分配計(jì)算、容錯(cuò)決策全部發(fā)生在 SDK 進(jìn)程內(nèi)部。這也意味著負(fù)載均衡的最終效果取決于客戶端配置與版本而不是集群端參數(shù)。二、Producer 側(cè)負(fù)載均衡隊(duì)列選擇與延遲容錯(cuò)2.1 從路由信息到隊(duì)列選擇TopicPublishInfoProducer 發(fā)送消息時(shí)第一步是根據(jù) Topic 從 NameServer 拉取路由信息構(gòu)建出TopicPublishInfo對(duì)象。該對(duì)象內(nèi)部維護(hù)了一個(gè)messageQueueList該 Topic 可寫(xiě)隊(duì)列列表和一個(gè)自增索引ThreadLocalIndex sendWhichQueue。隊(duì)列選擇的默認(rèn)邏輯位于 TopicPublishInfo.javaint index Math.abs(sendQueue.incrementAndGet() % messageQueueList.size()); return messageQueueList.get(index);即隨機(jī)遞增取模每次發(fā)送讓索引自增再對(duì)隊(duì)列總數(shù)取模。這種輪流 取模的方式能天然打散消息到不同隊(duì)列結(jié)合ThreadLocal特性每個(gè)發(fā)送線程維護(hù)自己的索引游標(biāo)避免多線程競(jìng)爭(zhēng)同一計(jì)數(shù)器從而獲得較好的并發(fā)性能。2.2 MQFaultStrategy發(fā)送容錯(cuò)的核心類(lèi)默認(rèn)的隊(duì)列選擇算法只負(fù)責(zé)輪轉(zhuǎn)并不感知 Broker 的可用性。為了讓發(fā)送具備高可用能力RocketMQ 在 MQFaultStrategy.java 中實(shí)現(xiàn)了**延遲故障容錯(cuò)LatencyFaultTolerance**機(jī)制由sendLatencyFaultEnable開(kāi)關(guān)控制對(duì)應(yīng) ClientConfig 中的sendLatencyEnable配置關(guān)閉默認(rèn)直接按隨機(jī)遞增取模從messageQueueList中選一個(gè)隊(duì)列發(fā)送邏輯最簡(jiǎn)單開(kāi)啟優(yōu)先過(guò)濾掉當(dāng)前不可達(dá)isReachable或暫不可用isAvailable的 Broker再執(zhí)行取模選擇。具體選擇順序見(jiàn) MQFaultStrategy.selectOneMessageQueue先嘗試在可用且非上次失敗 Broker的隊(duì)列中挑選availableFilterBrokerFilter若無(wú)結(jié)果再放寬到可達(dá)的隊(duì)列reachableFilter最后兜底直接取模選擇任意隊(duì)列。2.3 延遲分級(jí)與規(guī)避時(shí)間表所謂延遲故障容錯(cuò)是指根據(jù)上一次請(qǐng)求的耗時(shí)為某 Broker 設(shè)置一段規(guī)避期在規(guī)避期內(nèi)不向該 Broker 發(fā)送消息。文檔給出兩個(gè)典型示例上次請(qǐng)求延遲超過(guò) 550ms 時(shí)規(guī)避 30000ms超過(guò) 1000ms 時(shí)規(guī)避 60000ms。倉(cāng)庫(kù)源碼中的完整分級(jí)表定義在 MQFaultStrategy.java上次請(qǐng)求延遲latencyMaxms觸發(fā)規(guī)避時(shí)長(zhǎng)notAvailableDurationms≥ 500不規(guī)避≥ 1000不規(guī)避≥ 5502000≥ 18005000≥ 30006000≥ 500010000≥ 1500030000updateFaultItem方法在每次發(fā)送完成后被回調(diào)將本次耗時(shí)寫(xiě)入容錯(cuò)表computeNotAvailableDuration從延遲表尾部向前遍歷命中第一檔即返回對(duì)應(yīng)規(guī)避時(shí)長(zhǎng)MQFaultStrategy.java。若因異常觸發(fā)隔離isolation則直接按 10000ms 的固定時(shí)長(zhǎng)規(guī)避。這套按延遲分級(jí)、分級(jí)規(guī)避的設(shè)計(jì)是 RocketMQ 發(fā)送側(cè)高可用的關(guān)鍵慢 Broker 被自動(dòng)隔離一段時(shí)間快 Broker 獲得更多發(fā)送機(jī)會(huì)從而顯著降低發(fā)送超時(shí)率。2.4 從文檔到源碼兩個(gè)易混淆的點(diǎn)文檔中超過(guò) 550Lms 規(guī)避 30000Lms、超過(guò) 1000L 規(guī)避 60000L的表述是早期版本的示意。當(dāng)前倉(cāng)庫(kù)的實(shí)際分級(jí)表以源碼latencyMax/notAvailableDuration數(shù)組為準(zhǔn)如上表兩者核心思想一致延遲越高、規(guī)避越久LatencyFaultTolerance是抽象接口其實(shí)現(xiàn) LatencyFaultToleranceImpl 與配套單元測(cè)試 LatencyFaultToleranceImplTest.java 一起構(gòu)成了容錯(cuò)表的存取與淘汰邏輯可通過(guò)測(cè)試用例驗(yàn)證各分級(jí)的命中行為。三、Consumer 側(cè)負(fù)載均衡Push/Pull 統(tǒng)一的拉模型3.1 Push 模式本質(zhì)上是 Pull 的封裝RocketMQ 的兩種消費(fèi)模式Push / Pull底層都基于 Pull 拉取。Push 模式DefaultMQPushConsumer只是在客戶端做了一層封裝內(nèi)部的消息拉取線程從 Broker 拉回一批消息后提交給消費(fèi)線程池處理然后繼續(xù)嘗試?yán)∠乱慌舯敬挝蠢较t延遲一段時(shí)間再次重試。因此無(wú)論哪種模式Consumer 都必須先明確從哪個(gè) MessageQueue 拉取這就引出了消費(fèi)側(cè)的負(fù)載均衡——把 Broker 上的多個(gè)讀隊(duì)列按消費(fèi)組內(nèi)的消費(fèi)者數(shù)量進(jìn)行分配保證同一時(shí)刻一個(gè)隊(duì)列只歸屬一個(gè)消費(fèi)者。3.2 第一步Consumer 心跳上報(bào)Broker 建立消費(fèi)組視圖Consumer 啟動(dòng)后會(huì)通過(guò)定時(shí)任務(wù)持續(xù)向集群內(nèi)所有 Broker 實(shí)例發(fā)送心跳包。心跳包內(nèi)容包含消費(fèi)組名稱consumerGroup訂閱關(guān)系集合subscription 集合即訂閱了哪些 Topic消息通信模式MessageModel集群/廣播客戶端實(shí)例標(biāo)識(shí)clientId。Broker 收到心跳后在ConsumerManager的本地緩存consumerTable中維護(hù)消費(fèi)組與消費(fèi)者的對(duì)應(yīng)關(guān)系同時(shí)在channelInfoTable中保存封裝好的客戶端網(wǎng)絡(luò)通道信息。這兩張表就是后續(xù) Consumer 側(cè)負(fù)載均衡所需的元數(shù)據(jù)來(lái)源——Broker 據(jù)此回答當(dāng)前消費(fèi)組下有哪些消費(fèi)者在線。3.3 第二步RebalanceService 定時(shí)觸發(fā)重平衡Consumer 實(shí)例啟動(dòng)過(guò)程中會(huì)初始化MQClientInstance并隨之啟動(dòng)負(fù)載均衡服務(wù)線程RebalanceService。RebalanceService.java 默認(rèn)每 20 秒執(zhí)行一次重平衡兩個(gè)關(guān)鍵參數(shù)可通過(guò) JVM 系統(tǒng)屬性調(diào)整系統(tǒng)屬性默認(rèn)值作用rocketmq.client.rebalance.waitInterval20000 ms正常情況下的重平衡周期rocketmq.client.rebalance.minInterval1000 ms重平衡未收斂時(shí)分配結(jié)果與當(dāng)前工作隊(duì)列不一致的最短重試間隔其run()方法的核心邏輯是調(diào)用MQClientInstance.doRebalance()若返回結(jié)果表示已平衡則按waitInterval休眠否則按minInterval盡快再次觸發(fā)直到消費(fèi)組內(nèi)隊(duì)列分配收斂。3.4 第三步RebalanceImpl.rebalanceByTopic 主流程MQClientInstance.doRebalance()最終調(diào)用 RebalanceImpl.rebalanceByTopic —— 這是 Consumer 端負(fù)載均衡的核心類(lèi)與核心方法。該方法根據(jù)消費(fèi)模式走不同分支廣播模式BROADCASTINGTopic 下所有隊(duì)列全量分配給每個(gè)消費(fèi)者無(wú)需協(xié)商直接對(duì)比并更新本地processQueueTable集群模式CLUSTERING需要走完整的獲取隊(duì)列 → 獲取消費(fèi)者列表 → 計(jì)算分配 → 更新本地緩存流程這也是文檔重點(diǎn)講解的分支具體分四步1) 獲取 Topic 的消費(fèi)隊(duì)列集合mqSet從 RebalanceImpl 實(shí)例的本地緩存變量topicSubscribeInfoTable中取出該 Topic 下的消息隊(duì)列集合。該表由MQClientInstance定時(shí)從 NameServer 刷新路由信息后填充。2) 獲取消費(fèi)組內(nèi)的消費(fèi)者 ID 列表cidAll調(diào)用mQClientFactory.findConsumerIdList(topic, consumerGroup)向 Broker 發(fā)起一次 RPC 請(qǐng)求業(yè)務(wù)請(qǐng)求碼GET_CONSUMER_LIST_BY_GROUPBroker 根據(jù)前面心跳數(shù)據(jù)構(gòu)建的consumerTable應(yīng)答返回該消費(fèi)組下所有在線的消費(fèi)者 ID 列表。該步驟本質(zhì)上是讓 Broker 充當(dāng)在線成員發(fā)現(xiàn)角色。3) 排序后按分配策略計(jì)算隊(duì)列歸屬將mqAll消息隊(duì)列集合與cidAll消費(fèi)者列表分別排序再調(diào)用消息隊(duì)列分配策略算法計(jì)算當(dāng)前消費(fèi)者應(yīng)該拉取哪些隊(duì)列。默認(rèn)使用平均分配算法AllocateMessageQueueAveragely策略名 AVG。文檔將平均分配算法比作分頁(yè)把所有 MessageQueue 當(dāng)作記錄、所有消費(fèi)者當(dāng)作頁(yè)碼先算出每頁(yè)的平均大小與每頁(yè)的記錄區(qū)間再遍歷區(qū)間確定當(dāng)前消費(fèi)者應(yīng)分配到的隊(duì)列。其實(shí)現(xiàn)位于 AllocateMessageQueueAveragely.java分配結(jié)果AVG通過(guò)AllocateMessageQueueStrategy接口AllocateMessageQueueStrategy.java的allocate(consumerGroup, currentCID, mqAll, cidAll)方法返回。4) 更新本地消費(fèi)隊(duì)列表updateProcessQueueTableInRebalance拿到本次分配結(jié)果mqSet后調(diào)用updateProcessQueueTableInRebalance與本地緩存processQueueTable做差集比對(duì)RebalanceImpl.java紅色部分不再屬于我的隊(duì)列即processQueueTable中存在、但不在新mqSet中的隊(duì)列。將這些隊(duì)列對(duì)應(yīng)的ProcessQueue的Dropped屬性置為true隨后執(zhí)行removeUnnecessaryMessageQueue嘗試從緩存中移除。該方法的移除條件是每隔 1s 檢查一次能否取到當(dāng)前消費(fèi)處理隊(duì)列的鎖能取到則返回true并移除對(duì)應(yīng) Entry等待 1s 后仍取不到鎖則返回false暫不移除這是為了保證正在消費(fèi)中的隊(duì)列不被粗暴打斷屬于優(yōu)雅摘除綠色部分交集隊(duì)列processQueueTable與mqSet的交集。這里會(huì)判斷ProcessQueue是否已過(guò)期Pull 模式無(wú)此判斷若是 Push 模式且已過(guò)期同樣置Dropped true并按上述方式嘗試移除。之后對(duì)過(guò)濾后mqSet中的每個(gè)MessageQueue新建ProcessQueue對(duì)象并放入processQueueTable。創(chuàng)建時(shí)需要調(diào)用computePullFromWhere(mq)計(jì)算該隊(duì)列的下一條消費(fèi)進(jìn)度 offset填充到即將創(chuàng)建的PullRequest對(duì)象中。所有PullRequest匯總成pullRequestList后調(diào)用dispatchPullRequest方法將它們依次放入PullMessageService服務(wù)線程的阻塞隊(duì)列pullRequestQueue由該線程取出并真正向 Broker 發(fā)起 Pull 請(qǐng)求。至此一次完整的 Consumer 負(fù)載均衡閉環(huán)完成。3.5 核心設(shè)計(jì)思想整個(gè)消費(fèi)隊(duì)列分配體系建立在兩條樸素但關(guān)鍵的原則之上同一消費(fèi)組內(nèi)一個(gè)消息消費(fèi)隊(duì)列在同一時(shí)刻只能被一個(gè)消費(fèi)者消費(fèi)——這是防止消息被重復(fù)消費(fèi)到不同消費(fèi)者的前提也是分配算法正確性的根基一個(gè)消息消費(fèi)者可以同時(shí)消費(fèi)多個(gè)消息隊(duì)列——消費(fèi)者與隊(duì)列是多對(duì)一的關(guān)系因此當(dāng)消費(fèi)組內(nèi)消費(fèi)者數(shù)量少于隊(duì)列數(shù)量時(shí)部分消費(fèi)者會(huì)承擔(dān)多個(gè)隊(duì)列從而實(shí)現(xiàn)吞吐的最大化。此外從 RebalanceImpl.java 的整體結(jié)構(gòu)還可以看到隊(duì)列鎖lock/unlock/lockAll、臟偏移清理removeDirtyOffset、以及面向 POP 消費(fèi)模式的PopProcessQueue/popProcessQueueTable分支都是圍繞分配結(jié)果收斂 消費(fèi)進(jìn)度正確這一目標(biāo)設(shè)計(jì)的配套機(jī)制clientRebalance為true時(shí)走客戶端本地分配否則可回退到由 Broker 端計(jì)算分配結(jié)果的getRebalanceResultFromBroker路徑。四、實(shí)踐要點(diǎn)與配置指引4.1 發(fā)送側(cè)需要更高的發(fā)送可用性時(shí)開(kāi)啟延遲故障容錯(cuò)開(kāi)關(guān)clientConfig.setSendLatencyEnable(true)對(duì)應(yīng) MQFaultStrategy 中的sendLatencyFaultEnable并可通過(guò)setLatencyMax/setNotAvailableDuration按業(yè)務(wù)容忍度自定義分級(jí)表注意容錯(cuò)開(kāi)關(guān)會(huì)略微增加每次發(fā)送的過(guò)濾開(kāi)銷(xiāo)且慢 Broker 規(guī)避會(huì)改變消息的隊(duì)列分布適用場(chǎng)景是對(duì)發(fā)送成功率要求高、允許一定延遲傾斜的生產(chǎn)鏈路。4.2 消費(fèi)側(cè)默認(rèn)每 20s 重平衡一次消費(fèi)組擴(kuò)縮容后最多等待一個(gè)周期即可完成隊(duì)列再分配若希望加快收斂可調(diào)小rocketmq.client.rebalance.waitInterval默認(rèn)分配策略為平均分配AVG可通過(guò)DefaultMQPushConsumer#setAllocateMessageQueueStrategy注入自定義的AllocateMessageQueueStrategy實(shí)現(xiàn)實(shí)現(xiàn)灰度、粘性等定制化分配若 Topic 隊(duì)列數(shù)遠(yuǎn)大于消費(fèi)組消費(fèi)者數(shù)建議隊(duì)列數(shù)保持為消費(fèi)者數(shù)的整數(shù)倍以充分利用平均分配算法反之若消費(fèi)者數(shù)超過(guò)隊(duì)列數(shù)多出的消費(fèi)者將暫時(shí)分不到隊(duì)列。五、總結(jié)RocketMQ 的負(fù)載均衡以客戶端自治為設(shè)計(jì)哲學(xué)Producer 用隨機(jī)遞增取模 延遲故障容錯(cuò)保障發(fā)送高可用Consumer 用心跳上報(bào) 定時(shí)重平衡 隊(duì)列分配策略保障消費(fèi)高并發(fā)與不重復(fù)歸屬。其中MQFaultStrategy、RebalanceService、RebalanceImpl與AllocateMessageQueueStrategy構(gòu)成了整套機(jī)制的骨架全部實(shí)現(xiàn)集中在 client/src/main/java/org/apache/rocketmq/client 目錄下讀者可結(jié)合 LatencyFaultToleranceImplTest.java 等測(cè)試用例進(jìn)一步驗(yàn)證各分支行為。【免費(fèi)下載鏈接】rocketmqApache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.項(xiàng)目地址: https://gitcode.com/gh_mirrors/ro/rocketmq創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 RocketMQ 客戶端負(fù)載均衡機(jī)制深度解析：Producer 發(fā)送與 Consumer 訂閱的全鏈路原理
RocketMQ 客戶端負(fù)載均衡機(jī)制深度解析Producer 發(fā)送與 Consumer 訂閱的全鏈路原理【免費(fèi)下載鏈接】rocketmqApache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.項(xiàng)目地址: https://gitcode.com/gh_mirrors/ro/rocketmq導(dǎo)讀Apache RocketMQ 的負(fù)載均衡完全在客戶端完成不依賴 Broker 參與調(diào)度這與多數(shù)消息中間件在服務(wù)端分發(fā)消息的設(shè)計(jì)截然不同。本文基于倉(cāng)庫(kù)文檔 Design_LoadBalancing.md系統(tǒng)拆解兩套核心機(jī)制Producer 發(fā)送側(cè)的隊(duì)列選擇與延遲容錯(cuò)MQFaultStrategy以及Consumer 訂閱側(cè)的定時(shí)重平衡RebalanceService RebalanceImpl并對(duì)照 client 模塊 的源碼與測(cè)試說(shuō)明每條關(guān)鍵路徑的底層實(shí)現(xiàn)。讀完本文你將掌握 RocketMQ 客戶端負(fù)載均衡的完整數(shù)據(jù)流、默認(rèn)分配算法細(xì)節(jié)、容錯(cuò)參數(shù)含義以及如何在代碼中調(diào)整分配策略。一、負(fù)載均衡的總體架構(gòu)一切都在客戶端完成在 RocketMQ 中負(fù)載均衡Load Balancing并非由 NameServer 或 Broker 統(tǒng)一計(jì)算而是拆成兩條獨(dú)立的客戶端鏈路Producer 側(cè)負(fù)載均衡發(fā)送消息時(shí)在 Topic 的多個(gè)寫(xiě)隊(duì)列MessageQueue中選擇一個(gè)作為本次發(fā)送目標(biāo)Consumer 側(cè)負(fù)載均衡訂閱消息時(shí)將 Topic 下的一組讀隊(duì)列按消費(fèi)組內(nèi)的消費(fèi)者數(shù)量進(jìn)行劃分決定哪個(gè)消費(fèi)者消費(fèi)哪些隊(duì)列。Broker 和 NameServer 只負(fù)責(zé)提供路由元數(shù)據(jù)Topic 與隊(duì)列的對(duì)應(yīng)關(guān)系和消費(fèi)組心跳信息真正的分配計(jì)算、容錯(cuò)決策全部發(fā)生在 SDK 進(jìn)程內(nèi)部。這也意味著負(fù)載均衡的最終效果取決于客戶端配置與版本而不是集群端參數(shù)。二、Producer 側(cè)負(fù)載均衡隊(duì)列選擇與延遲容錯(cuò)2.1 從路由信息到隊(duì)列選擇TopicPublishInfoProducer 發(fā)送消息時(shí)第一步是根據(jù) Topic 從 NameServer 拉取路由信息構(gòu)建出TopicPublishInfo對(duì)象。該對(duì)象內(nèi)部維護(hù)了一個(gè)messageQueueList該 Topic 可寫(xiě)隊(duì)列列表和一個(gè)自增索引ThreadLocalIndex sendWhichQueue。隊(duì)列選擇的默認(rèn)邏輯位于 TopicPublishInfo.javaint index Math.abs(sendQueue.incrementAndGet() % messageQueueList.size()); return messageQueueList.get(index);即隨機(jī)遞增取模每次發(fā)送讓索引自增再對(duì)隊(duì)列總數(shù)取模。這種輪流 取模的方式能天然打散消息到不同隊(duì)列結(jié)合ThreadLocal特性每個(gè)發(fā)送線程維護(hù)自己的索引游標(biāo)避免多線程競(jìng)爭(zhēng)同一計(jì)數(shù)器從而獲得較好的并發(fā)性能。2.2 MQFaultStrategy發(fā)送容錯(cuò)的核心類(lèi)默認(rèn)的隊(duì)列選擇算法只負(fù)責(zé)輪轉(zhuǎn)并不感知 Broker 的可用性。為了讓發(fā)送具備高可用能力RocketMQ 在 MQFaultStrategy.java 中實(shí)現(xiàn)了**延遲故障容錯(cuò)LatencyFaultTolerance**機(jī)制由sendLatencyFaultEnable開(kāi)關(guān)控制對(duì)應(yīng) ClientConfig 中的sendLatencyEnable配置關(guān)閉默認(rèn)直接按隨機(jī)遞增取模從messageQueueList中選一個(gè)隊(duì)列發(fā)送邏輯最簡(jiǎn)單開(kāi)啟優(yōu)先過(guò)濾掉當(dāng)前不可達(dá)isReachable或暫不可用isAvailable的 Broker再執(zhí)行取模選擇。具體選擇順序見(jiàn) MQFaultStrategy.selectOneMessageQueue先嘗試在可用且非上次失敗 Broker的隊(duì)列中挑選availableFilterBrokerFilter若無(wú)結(jié)果再放寬到可達(dá)的隊(duì)列reachableFilter最后兜底直接取模選擇任意隊(duì)列。2.3 延遲分級(jí)與規(guī)避時(shí)間表所謂延遲故障容錯(cuò)是指根據(jù)上一次請(qǐng)求的耗時(shí)為某 Broker 設(shè)置一段規(guī)避期在規(guī)避期內(nèi)不向該 Broker 發(fā)送消息。文檔給出兩個(gè)典型示例上次請(qǐng)求延遲超過(guò) 550ms 時(shí)規(guī)避 30000ms超過(guò) 1000ms 時(shí)規(guī)避 60000ms。倉(cāng)庫(kù)源碼中的完整分級(jí)表定義在 MQFaultStrategy.java上次請(qǐng)求延遲latencyMaxms觸發(fā)規(guī)避時(shí)長(zhǎng)notAvailableDurationms≥ 500不規(guī)避≥ 1000不規(guī)避≥ 5502000≥ 18005000≥ 30006000≥ 500010000≥ 1500030000updateFaultItem方法在每次發(fā)送完成后被回調(diào)將本次耗時(shí)寫(xiě)入容錯(cuò)表computeNotAvailableDuration從延遲表尾部向前遍歷命中第一檔即返回對(duì)應(yīng)規(guī)避時(shí)長(zhǎng)MQFaultStrategy.java。若因異常觸發(fā)隔離isolation則直接按 10000ms 的固定時(shí)長(zhǎng)規(guī)避。這套按延遲分級(jí)、分級(jí)規(guī)避的設(shè)計(jì)是 RocketMQ 發(fā)送側(cè)高可用的關(guān)鍵慢 Broker 被自動(dòng)隔離一段時(shí)間快 Broker 獲得更多發(fā)送機(jī)會(huì)從而顯著降低發(fā)送超時(shí)率。2.4 從文檔到源碼兩個(gè)易混淆的點(diǎn)文檔中超過(guò) 550Lms 規(guī)避 30000Lms、超過(guò) 1000L 規(guī)避 60000L的表述是早期版本的示意。當(dāng)前倉(cāng)庫(kù)的實(shí)際分級(jí)表以源碼latencyMax/notAvailableDuration數(shù)組為準(zhǔn)如上表兩者核心思想一致延遲越高、規(guī)避越久LatencyFaultTolerance是抽象接口其實(shí)現(xiàn) LatencyFaultToleranceImpl 與配套單元測(cè)試 LatencyFaultToleranceImplTest.java 一起構(gòu)成了容錯(cuò)表的存取與淘汰邏輯可通過(guò)測(cè)試用例驗(yàn)證各分級(jí)的命中行為。三、Consumer 側(cè)負(fù)載均衡Push/Pull 統(tǒng)一的拉模型3.1 Push 模式本質(zhì)上是 Pull 的封裝RocketMQ 的兩種消費(fèi)模式Push / Pull底層都基于 Pull 拉取。Push 模式DefaultMQPushConsumer只是在客戶端做了一層封裝內(nèi)部的消息拉取線程從 Broker 拉回一批消息后提交給消費(fèi)線程池處理然后繼續(xù)嘗試?yán)∠乱慌舯敬挝蠢较t延遲一段時(shí)間再次重試。因此無(wú)論哪種模式Consumer 都必須先明確從哪個(gè) MessageQueue 拉取這就引出了消費(fèi)側(cè)的負(fù)載均衡——把 Broker 上的多個(gè)讀隊(duì)列按消費(fèi)組內(nèi)的消費(fèi)者數(shù)量進(jìn)行分配保證同一時(shí)刻一個(gè)隊(duì)列只歸屬一個(gè)消費(fèi)者。3.2 第一步Consumer 心跳上報(bào)Broker 建立消費(fèi)組視圖Consumer 啟動(dòng)后會(huì)通過(guò)定時(shí)任務(wù)持續(xù)向集群內(nèi)所有 Broker 實(shí)例發(fā)送心跳包。心跳包內(nèi)容包含消費(fèi)組名稱consumerGroup訂閱關(guān)系集合subscription 集合即訂閱了哪些 Topic消息通信模式MessageModel集群/廣播客戶端實(shí)例標(biāo)識(shí)clientId。Broker 收到心跳后在ConsumerManager的本地緩存consumerTable中維護(hù)消費(fèi)組與消費(fèi)者的對(duì)應(yīng)關(guān)系同時(shí)在channelInfoTable中保存封裝好的客戶端網(wǎng)絡(luò)通道信息。這兩張表就是后續(xù) Consumer 側(cè)負(fù)載均衡所需的元數(shù)據(jù)來(lái)源——Broker 據(jù)此回答當(dāng)前消費(fèi)組下有哪些消費(fèi)者在線。3.3 第二步RebalanceService 定時(shí)觸發(fā)重平衡Consumer 實(shí)例啟動(dòng)過(guò)程中會(huì)初始化MQClientInstance并隨之啟動(dòng)負(fù)載均衡服務(wù)線程RebalanceService。RebalanceService.java 默認(rèn)每 20 秒執(zhí)行一次重平衡兩個(gè)關(guān)鍵參數(shù)可通過(guò) JVM 系統(tǒng)屬性調(diào)整系統(tǒng)屬性默認(rèn)值作用rocketmq.client.rebalance.waitInterval20000 ms正常情況下的重平衡周期rocketmq.client.rebalance.minInterval1000 ms重平衡未收斂時(shí)分配結(jié)果與當(dāng)前工作隊(duì)列不一致的最短重試間隔其run()方法的核心邏輯是調(diào)用MQClientInstance.doRebalance()若返回結(jié)果表示已平衡則按waitInterval休眠否則按minInterval盡快再次觸發(fā)直到消費(fèi)組內(nèi)隊(duì)列分配收斂。3.4 第三步RebalanceImpl.rebalanceByTopic 主流程MQClientInstance.doRebalance()最終調(diào)用 RebalanceImpl.rebalanceByTopic —— 這是 Consumer 端負(fù)載均衡的核心類(lèi)與核心方法。該方法根據(jù)消費(fèi)模式走不同分支廣播模式BROADCASTINGTopic 下所有隊(duì)列全量分配給每個(gè)消費(fèi)者無(wú)需協(xié)商直接對(duì)比并更新本地processQueueTable集群模式CLUSTERING需要走完整的獲取隊(duì)列 → 獲取消費(fèi)者列表 → 計(jì)算分配 → 更新本地緩存流程這也是文檔重點(diǎn)講解的分支具體分四步1) 獲取 Topic 的消費(fèi)隊(duì)列集合mqSet從 RebalanceImpl 實(shí)例的本地緩存變量topicSubscribeInfoTable中取出該 Topic 下的消息隊(duì)列集合。該表由MQClientInstance定時(shí)從 NameServer 刷新路由信息后填充。2) 獲取消費(fèi)組內(nèi)的消費(fèi)者 ID 列表cidAll調(diào)用mQClientFactory.findConsumerIdList(topic, consumerGroup)向 Broker 發(fā)起一次 RPC 請(qǐng)求業(yè)務(wù)請(qǐng)求碼GET_CONSUMER_LIST_BY_GROUPBroker 根據(jù)前面心跳數(shù)據(jù)構(gòu)建的consumerTable應(yīng)答返回該消費(fèi)組下所有在線的消費(fèi)者 ID 列表。該步驟本質(zhì)上是讓 Broker 充當(dāng)在線成員發(fā)現(xiàn)角色。3) 排序后按分配策略計(jì)算隊(duì)列歸屬將mqAll消息隊(duì)列集合與cidAll消費(fèi)者列表分別排序再調(diào)用消息隊(duì)列分配策略算法計(jì)算當(dāng)前消費(fèi)者應(yīng)該拉取哪些隊(duì)列。默認(rèn)使用平均分配算法AllocateMessageQueueAveragely策略名 AVG。文檔將平均分配算法比作分頁(yè)把所有 MessageQueue 當(dāng)作記錄、所有消費(fèi)者當(dāng)作頁(yè)碼先算出每頁(yè)的平均大小與每頁(yè)的記錄區(qū)間再遍歷區(qū)間確定當(dāng)前消費(fèi)者應(yīng)分配到的隊(duì)列。其實(shí)現(xiàn)位于 AllocateMessageQueueAveragely.java分配結(jié)果AVG通過(guò)AllocateMessageQueueStrategy接口AllocateMessageQueueStrategy.java的allocate(consumerGroup, currentCID, mqAll, cidAll)方法返回。4) 更新本地消費(fèi)隊(duì)列表updateProcessQueueTableInRebalance拿到本次分配結(jié)果mqSet后調(diào)用updateProcessQueueTableInRebalance與本地緩存processQueueTable做差集比對(duì)RebalanceImpl.java紅色部分不再屬于我的隊(duì)列即processQueueTable中存在、但不在新mqSet中的隊(duì)列。將這些隊(duì)列對(duì)應(yīng)的ProcessQueue的Dropped屬性置為true隨后執(zhí)行removeUnnecessaryMessageQueue嘗試從緩存中移除。該方法的移除條件是每隔 1s 檢查一次能否取到當(dāng)前消費(fèi)處理隊(duì)列的鎖能取到則返回true并移除對(duì)應(yīng) Entry等待 1s 后仍取不到鎖則返回false暫不移除這是為了保證正在消費(fèi)中的隊(duì)列不被粗暴打斷屬于優(yōu)雅摘除綠色部分交集隊(duì)列processQueueTable與mqSet的交集。這里會(huì)判斷ProcessQueue是否已過(guò)期Pull 模式無(wú)此判斷若是 Push 模式且已過(guò)期同樣置Dropped true并按上述方式嘗試移除。之后對(duì)過(guò)濾后mqSet中的每個(gè)MessageQueue新建ProcessQueue對(duì)象并放入processQueueTable。創(chuàng)建時(shí)需要調(diào)用computePullFromWhere(mq)計(jì)算該隊(duì)列的下一條消費(fèi)進(jìn)度 offset填充到即將創(chuàng)建的PullRequest對(duì)象中。所有PullRequest匯總成pullRequestList后調(diào)用dispatchPullRequest方法將它們依次放入PullMessageService服務(wù)線程的阻塞隊(duì)列pullRequestQueue由該線程取出并真正向 Broker 發(fā)起 Pull 請(qǐng)求。至此一次完整的 Consumer 負(fù)載均衡閉環(huán)完成。3.5 核心設(shè)計(jì)思想整個(gè)消費(fèi)隊(duì)列分配體系建立在兩條樸素但關(guān)鍵的原則之上同一消費(fèi)組內(nèi)一個(gè)消息消費(fèi)隊(duì)列在同一時(shí)刻只能被一個(gè)消費(fèi)者消費(fèi)——這是防止消息被重復(fù)消費(fèi)到不同消費(fèi)者的前提也是分配算法正確性的根基一個(gè)消息消費(fèi)者可以同時(shí)消費(fèi)多個(gè)消息隊(duì)列——消費(fèi)者與隊(duì)列是多對(duì)一的關(guān)系因此當(dāng)消費(fèi)組內(nèi)消費(fèi)者數(shù)量少于隊(duì)列數(shù)量時(shí)部分消費(fèi)者會(huì)承擔(dān)多個(gè)隊(duì)列從而實(shí)現(xiàn)吞吐的最大化。此外從 RebalanceImpl.java 的整體結(jié)構(gòu)還可以看到隊(duì)列鎖lock/unlock/lockAll、臟偏移清理removeDirtyOffset、以及面向 POP 消費(fèi)模式的PopProcessQueue/popProcessQueueTable分支都是圍繞分配結(jié)果收斂 消費(fèi)進(jìn)度正確這一目標(biāo)設(shè)計(jì)的配套機(jī)制clientRebalance為true時(shí)走客戶端本地分配否則可回退到由 Broker 端計(jì)算分配結(jié)果的getRebalanceResultFromBroker路徑。四、實(shí)踐要點(diǎn)與配置指引4.1 發(fā)送側(cè)需要更高的發(fā)送可用性時(shí)開(kāi)啟延遲故障容錯(cuò)開(kāi)關(guān)clientConfig.setSendLatencyEnable(true)對(duì)應(yīng) MQFaultStrategy 中的sendLatencyFaultEnable并可通過(guò)setLatencyMax/setNotAvailableDuration按業(yè)務(wù)容忍度自定義分級(jí)表注意容錯(cuò)開(kāi)關(guān)會(huì)略微增加每次發(fā)送的過(guò)濾開(kāi)銷(xiāo)且慢 Broker 規(guī)避會(huì)改變消息的隊(duì)列分布適用場(chǎng)景是對(duì)發(fā)送成功率要求高、允許一定延遲傾斜的生產(chǎn)鏈路。4.2 消費(fèi)側(cè)默認(rèn)每 20s 重平衡一次消費(fèi)組擴(kuò)縮容后最多等待一個(gè)周期即可完成隊(duì)列再分配若希望加快收斂可調(diào)小rocketmq.client.rebalance.waitInterval默認(rèn)分配策略為平均分配AVG可通過(guò)DefaultMQPushConsumer#setAllocateMessageQueueStrategy注入自定義的AllocateMessageQueueStrategy實(shí)現(xiàn)實(shí)現(xiàn)灰度、粘性等定制化分配若 Topic 隊(duì)列數(shù)遠(yuǎn)大于消費(fèi)組消費(fèi)者數(shù)建議隊(duì)列數(shù)保持為消費(fèi)者數(shù)的整數(shù)倍以充分利用平均分配算法反之若消費(fèi)者數(shù)超過(guò)隊(duì)列數(shù)多出的消費(fèi)者將暫時(shí)分不到隊(duì)列。五、總結(jié)RocketMQ 的負(fù)載均衡以客戶端自治為設(shè)計(jì)哲學(xué)Producer 用隨機(jī)遞增取模 延遲故障容錯(cuò)保障發(fā)送高可用Consumer 用心跳上報(bào) 定時(shí)重平衡 隊(duì)列分配策略保障消費(fèi)高并發(fā)與不重復(fù)歸屬。其中MQFaultStrategy、RebalanceService、RebalanceImpl與AllocateMessageQueueStrategy構(gòu)成了整套機(jī)制的骨架全部實(shí)現(xiàn)集中在 client/src/main/java/org/apache/rocketmq/client 目錄下讀者可結(jié)合 LatencyFaultToleranceImplTest.java 等測(cè)試用例進(jìn)一步驗(yàn)證各分支行為。【免費(fèi)下載鏈接】rocketmqApache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.項(xiàng)目地址: https://gitcode.com/gh_mirrors/ro/rocketmq創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考