實驗現象

準備

 建立的topicA、topicB，每個都是16個訊息佇列;

 建立了同一個消費者組：ntm-hxy-group

 預設訊息模式：負載均衡

消費者一（clientId1）

先啟動一個消費者一：clientId:192.168.31.182@10962#386194072446460

public static void main(String[] args) throws Exception {
    String nameAddress = "192.168.11.180:9876;192.168.11.181:9876;192.168.11.182:9876";
    DefaultMQPushConsumer concurrentlyPushConsumer2 
 
=
            createConcurrentlyPushConsumer("ntm-hxy-group", nameAddress, "topicA", "topicB");
}

這個clientId1 訂閱了topicA、topicB，屬於組ntm-hxy-group

這裡代表訂閱的clientId的數量

 上圖訂閱關係：client1訂閱了兩個topic，自動添加了一個重試的topic

該消費者組ntm-hxy-group消費佇列的情況如下：

這是重試topic，只有一個訊息佇列，由clientId1消費

 topicA所有16個佇列，都由clientId消費

 topicB 所有16個訊息佇列都由clientId1 消費

 由此可見：兩個topic都被這一個clientId1佔用了，還有一個重試topic也是被該clinetId1佔用

 再啟動一個消費者（clientId2）

在啟動一個clientId2，同一個group，只訂閱topicA。clientId:192.168.31.182@11084#386532403737756

控制檯介面：

同一個消費者組group，clientId變為了兩個 

 topic只有topicA了，也就是最新訂閱資訊，但是clientId還有兩個。

此時，該消費者組ntm-hxy-group消費佇列的情況如下：

 如上圖，重試訊息佇列消費物件沒有變，因為重試topic是服務端（broker）基於group產生的，所以一直和第一個啟動的clientId繫結的，也就是重試佇列會一直由第一個clientId消費，不會由於心跳引起訂閱的改變而改變。

這時剛啟動第二個的時候，只有topicA了，並且訊息佇列的消費物件被負載均衡給了兩個clientId

也就是每個clientId可以消費topicA一半的訊息，但是此時topicB丟失了，也就是傳送給topicB的訊息會被堆積，不能被消費了。

預設每30秒傳送心跳：

 下次clientId1傳送心跳時，topicB又訂閱上了，此時的情況如下：

 這個重試group跟心跳無關，還是clientId1消費沒變。

topicA還是由兩個clientId進行負載均衡

但是新增了訂閱topicB，如下：

 原因：雖然client1傳送心跳時，訂閱資訊重新整理了，topicA和topicB都訂閱了，但是消費佇列繫結關係是根據group來進行替換的，因為負載均衡時關鍵兩點：

①根據topic獲取MesssageQueue列表

在 RebalanceImpl 維護了一份 map 結構的本地快取 topicSubscribeInfoTable，以 topic 維度儲存了對應的 MesssageQueue列表；

②根據topic、group獲取線上的消費者終端列表（即線上的clientId）

findConsumerIdList 方法接受兩個引數：Topic 主題和 ConsumerGroup 消費組

底層通過傳送 RequestCode.GET_CONSUMER_LIST_BY_GROUP 請求碼的 RemotingCommand 到 Broker 查詢線上消費者列表，拿到結果後反序列化

所以對tpicB進行負載均衡時，根據topicB在本地快取找到了16個佇列（即mqSet），然後根據groupName去查詢線上的clientId，找到了兩個（clientId:192.168.31.182@10962#386194072446460、clientId:192.168.31.182@11084#386532403737756）即cidAll的值，之後拿mqSet、cidAll進行預設的平均負載均衡策略進行負載均衡時，邏輯如下：

 這個方法預設負載均衡策略具體實現如下：

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>();
    if (!cidAll.contains(currentCID)) {
        log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
            consumerGroup,
            currentCID,
            cidAll);
        return result;
    }

    int index = cidAll.indexOf(currentCID);
    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;
}

所以就是16個佇列，被兩個clientId平均分配了，clientId1先啟動，所以分配topicB前八個訊息佇列，而clientId2應該分配topicB後八個訊息佇列的，但是發現topicB並沒有被clientId2消費端訂閱，所以topicB的後八個訊息佇列的消費端就丟失了，也就是後八個佇列的訊息會堆積，不會有消費者消費。

這是通過控制檯看到的現象，至於為什麼明明topicB分配給了clientId2的訊息佇列，卻由於clientId2沒有訂閱topicB的原因而導致丟失了消費端的繫結？這個原因在下一個實驗通過程式碼解釋。

• 中文（簡體）
• 中文（繁體）
• 丹麥語
• 烏克蘭語
• 烏爾都語
• 亞美尼亞語
• 俄語
• 保加利亞語
• 克羅埃西亞語
• 冰島語
• 加泰羅尼亞語
• 匈牙利語
• 卡納達語
• 印地語
• 印尼語
• 古吉拉特語
• 哈薩克語
• 土耳其語
• 威爾士語
• 孟加拉語
• 尼泊爾語
• 布林語(南非荷蘭語)
• 希伯來語
• 希臘語
• 庫爾德語
• 德語
• 義大利語
• 拉脫維亞語
• 挪威語
• 捷克語
• 斯洛伐克語
• 斯洛維尼亞語
• 旁遮普語
• 日語
• 普什圖語
• 毛利語
• 法語
• 波蘭語
• 波斯語
• 泰盧固語
• 泰米爾語
• 泰語
• 海地克里奧爾語
• 愛沙尼亞語
• 瑞典語
• 立陶宛語
• 緬甸語
• 羅馬尼亞語
• 寮國語
• 芬蘭語
• 英語
• 荷蘭語
• 薩摩亞語
• 葡萄牙語
• 西班牙語
• 越南語
• 亞塞拜然語
• 阿姆哈拉語
• 阿爾巴尼亞語
• 阿拉伯語
• 韓語
• 馬爾加什語
• 馬拉地語
• 馬拉雅拉姆語
• 馬來語
• 馬耳他語
• 高棉語