關于 RocketMQ ClientID 相同引發的消息堆積的問題
首先，造成這個問題的 BUG RocketMQ 官方已經在 3月16號 的這個提交中修復了，這里只是探討一下在修復之前造成問題的具體細節，更多的上下文可以參考我之前寫的 《RocketMQ Consumer 啟動時都干了些啥?》 ，這篇文章講解了 RocketMQ 的 Consumer 啟動之后都做了哪些操作，對理解本次要講解的 BUG 有一定的幫助。
其中講到了：
消息堆積
重復消費自不必說，你 ClientID 都相同了。本篇著重聊聊為什么會消息堆積。
文章中講到，初始化 Consumer 時，會初始化 Rebalance 的策略。你可以大致將 Rebalance 策略理解為如何將一個 Topic 下的 m 個 MessageQueue 分配給一個 ConsumerGroup 下的 n 個 Consumer 實例的策略，看著有些繞，其實就長這樣：
rebalance策略
而從 Consumer 初始化的源碼中可以看出，默認情況下 Consumer 采取的 Rebalance 策略是 AllocateMessageQueueAverage()。
默認的 Rebalance 策略
默認的策略很好理解，將 MessageQueue 平均的分配給 Consumer。舉個例子，假設有 8 個 MessageQueue，2 個 Consumer，那么每個 Consumer 就會被分配到 4 個 MessageQueue。
那如果分配不均勻怎么辦?例如只有 7 個 MessageQueue，但是 Consumer 仍然是 2 個。此時 RocketMQ 會將多出來的部分，對已經排好序的 Consumer 再做平均分配，一個一個分發給 Consumer，直到分發完。例如剛剛說的 7 個 MessageQueue 和 2 個 ConsumerGroup 這種 case，排在第一個的 Consumer 就會被分配到 4 個 MessageQueue，而第二個會被分配到 3 個 MessageQueue。
大家可以先理解一下 AllocateMessageQueueAveragely 的實現，作為默認的 Rebalance 的策略，其實現位于這里：
默認策略的實現位置
接下來我們看看，AllocateMessageQueueAveragely 內部具體都做了哪些事情。
其核心其實就是實現的 AllocateMessageQueueStrategy 接口中的 allocate 方法。實際上，RocketMQ 對該接口總共有 5 種實現：
- AllocateMachineRoomNearby
- AllocateMessageQueueAveragely
- AllocateMessageQueueAveragelyByCircle
- AllocateMessageQueueByConfig
- AllocateMessageQueueByMachineRoom
- AllocateMessageQueueConsistentHash
其默認的 AllocateMessageQueueAveragely 只是其中的一種實現而已，那執行 allocate 它需要什么參數呢?
入參
需要以下四個：
- ConsumerGroup 消費者組的名字
- currentCID 當前消費者的 clientID
- mqAll 當前 ConsumerGroup 所消費的 Topic 下的所有的 MessageQueue
- cidAll 當前 ConsumerGroup 下所有消費者的 ClientID
實際上是將某個 Topic 下的所有 MessageQueue 分配給屬于同一個消費者的所有消費者實例，粒度是 By Topic 的。
所以到這里剩下的事情就很簡單了，無非就是怎么樣把這一堆 MessageQueue 分配給這一堆 Consumer。這個怎么樣，就對應了 AllocateMessageQueueStrategy 的不同實現。
接下來我們就來看看 AllocateMessageQueueAveragely 是如何對 MessageQueue 進行分配的，之前講源碼我一般都會一步一步的來，結合源碼跟圖，但是這個源碼太短了，我就直接先給出來吧。
- public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {
- if (currentCID == null || currentCID.length() < 1) {
- throw new IllegalArgumentException("currentCID is empty");
- }
- if (mqAll == null || mqAll.isEmpty()) {
- throw new IllegalArgumentException("mqAll is null or mqAll empty");
- }
- if (cidAll == null || cidAll.isEmpty()) {
- throw new IllegalArgumentException("cidAll is null or cidAll empty");
- }
- List<MessageQueue> result = new ArrayList<MessageQueue>();
- // 判斷一下當前的客戶端是否在 cidAll 的集合當中
- if (!cidAll.contains(currentCID)) {
- log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
- consumerGroup,
- currentCID,
- cidAll);
- return result;
- }
- // 拿到當前消費者在所有的消費者實例數組中的位置
- int index = cidAll.indexOf(currentCID);
- // 用 messageQueue 的數量 對 消費者實例的數量取余數, 這個實際上就把不夠均勻分的 MessageQueue 的數量算出來了
- // 舉個例子, 12 個 MessageQueue, 有 5 個 Consumer, 12 % 5 = 2
- int mod = mqAll.size() % cidAll.size();
- int averageSize =
- mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size() + 1 : mqAll.size() / cidAll.size());
- int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
- int range = Math.min(averageSize, mqAll.size() - startIndex);
- for (int i = 0; i < range; i++) {
- result.add(mqAll.get((startIndex + i) % mqAll.size()));
- }
- return result;
- }
其實前半部分都是些常規的 check，可以忽略不看，從這里：
- int index = cidAll.indexOf(currentCID);
開始，才是核心邏輯。為了避免邏輯混亂，還是假設有 12 個 MessageQueue，5 個 Consumer，同時假設 index=0 。
那么 mod 的值就為 12 % 5 = 2 了。
而 averageSize 的值，稍微有點繞。如果 MessageQueue 的數量比消費者的數量還少，那么就為 1 ;否則，就走這一堆邏輯(mod > 0 && index < mod ? mqAll.size() / cidAll.size() + 1 : mqAll.size() / cidAll.size())。我們 index 是 0，而 mod 是 2，index < mod 則是成立的，那么最終 averageSize 的值就為 12 / 5 + 1 = 3。
接下來是 startIndex，由于這個三元運算符的條件是成立的，所以其值為 0 * 3 ，就為 0。
看了一大堆邏輯，是不是已經暈了?直接舉實例：
12 個 Message Queue
5 個 Consumer 實例
按照上面的分法：
排在第 1 的消費者 分到 3 個
排在第 2 的消費者 分到 3 個
排在第 3 的消費者 分到 2 個
排在第 4 的消費者 分到 2 個
排在第 5 的消費者 分到 2 個
具體分配流程
所以，你可以大致認為：
先“均分”，12 / 5 取整為 2。然后“均分”完之后還剩下 2 個，那么就從上往下，挨個再分配，這樣第 1、第 2 個消費者就會被多分到 1 個。
所以如果有 13 個 MessageQueue，5 個 Consumer，那么第 1、第 2、第 3 就會被分配 3 個。
但并不準確，因為分配的 MessageQueue 是一次性的，例如那 3 個 MessageQueue 是一次性獲取的，不會先給 2 個，再給 1 個。
而我們開篇提到的 Consumer 的 ClientID 相同，會造成什么?
當然是 index 的值相同，進而造成 mod、averageSize、startIndex、range 全部相同。那么最后 result.add(mqAll.get((startIndex + i) % mqAll.size())); 時，本來不同的 Consumer，會取到相同的 MessageQueue(舉個例子，Consumer 1 和 Consumer 2 都取到了前 3 個 MessageQueue)，從而造成有些 MessageQueue(如果有的話) 沒有 Consumer 對其消費，而沒有被消費，消息也在不停的投遞進來，就會造成消息的大量堆積。
當然，現在的新版本從代碼上看已經修復這個問題了，這個只是對之前的版本的原因做一個探索。
