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