[[436290]]

首先，造成這個問題的 BUG RocketMQ 官方已經(jīng)在 3月16號 的這個提交中修復了，這里只是探討一下在修復之前造成問題的具體細節(jié)，更多的上下文可以參考我之前寫的 《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 會將多出來的部分，對已經(jīng)排好序的 Consumer 再做平均分配，一個一個分發(fā)給 Consumer，直到分發(fā)完。例如剛剛說的 7 個 MessageQueue 和 2 個 ConsumerGroup 這種 case，排在第一個的 Consumer 就會被分配到 4 個 MessageQueue，而第二個會被分配到 3 個 MessageQueue。

大家可以先理解一下 AllocateMessageQueueAveragely 的實現(xiàn)，作為默認的 Rebalance 的策略，其實現(xiàn)位于這里：

默認策略的實現(xiàn)位置

接下來我們看看，AllocateMessageQueueAveragely 內(nèi)部具體都做了哪些事情。

其核心其實就是實現(xiàn)的 AllocateMessageQueueStrategy 接口中的 allocate 方法。實際上，RocketMQ 對該接口總共有 5 種實現(xiàn)：

• AllocateMachineRoomNearby
• AllocateMessageQueueAveragely
• AllocateMessageQueueAveragelyByCircle
• AllocateMessageQueueByConfig
• AllocateMessageQueueByMachineRoom
• AllocateMessageQueueConsistentHash

其默認的 AllocateMessageQueueAveragely 只是其中的一種實現(xiàn)而已，那執(zhí)行 allocate 它需要什么參數(shù)呢?

入?yún)?/p>

需要以下四個：

• ConsumerGroup 消費者組的名字
• currentCID 當前消費者的 clientID
• mqAll 當前 ConsumerGroup 所消費的 Topic 下的所有的 MessageQueue
• cidAll 當前 ConsumerGroup 下所有消費者的 ClientID

實際上是將某個 Topic 下的所有 MessageQueue 分配給屬于同一個消費者的所有消費者實例，粒度是 By Topic 的。

所以到這里剩下的事情就很簡單了，無非就是怎么樣把這一堆 MessageQueue 分配給這一堆 Consumer。這個怎么樣，就對應了 AllocateMessageQueueStrategy 的不同實現(xiàn)。

接下來我們就來看看 AllocateMessageQueueAveragely 是如何對 MessageQueue 進行分配的，之前講源碼我一般都會一步一步的來，結(jié)合源碼跟圖，但是這個源碼太短了，我就直接先給出來吧。

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; 
  } 
 
  // 拿到當前消費者在所有的消費者實例數(shù)組中的位置 
  int index = cidAll.indexOf(currentCID); 
  // 用 messageQueue 的數(shù)量 對 消費者實例的數(shù)量取余數(shù), 這個實際上就把不夠均勻分的 MessageQueue 的數(shù)量算出來了 
  // 舉個例子, 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; 
} 

其實前半部分都是些常規(guī)的 check，可以忽略不看，從這里：

int index = cidAll.indexOf(currentCID); 

開始，才是核心邏輯。為了避免邏輯混亂，還是假設有 12 個 MessageQueue，5 個 Consumer，同時假設 index=0 。

那么 mod 的值就為 12 % 5 = 2 了。

而 averageSize 的值，稍微有點繞。如果 MessageQueue 的數(shù)量比消費者的數(shù)量還少，那么就為 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。

看了一大堆邏輯，是不是已經(jīng)暈了?直接舉實例：

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 對其消費，而沒有被消費，消息也在不停的投遞進來，就會造成消息的大量堆積。

當然，現(xiàn)在的新版本從代碼上看已經(jīng)修復這個問題了，這個只是對之前的版本的原因做一個探索。