這篇文章，筆者想聊聊 RocketMQ 最佳實踐之一：保證訂閱關係一致。

訂閱關係一致指的是同一個消費者 Group ID 下所有 Consumer 範例所訂閱的 Topic 、Tag 必須完全一致。

如果訂閱關係不一致，訊息消費的邏輯就會混亂，甚至導致訊息丟失。

1 訂閱關係演示

首先我們展示正確的訂閱關係：多個 Group ID 訂閱了多個 Topic，並且每個 Group ID 裡的多個消費者的訂閱關係保持了一致。

接下來，我們展示錯誤的訂閱關係。

從上圖中，單個 Group ID 訂閱了多個 Topic，但是該 Group ID 裡的多個消費者的訂閱關係並沒有保持一致。

程式碼邏輯角度來看，每個消費者範例內訂閱方法的主題、 TAG、監聽邏輯都需要保持一致。

接下來，我們實驗相同消費組，兩種不正確的場景，看看消費者和 Broker 服務有什麼異常。

訂閱主題不同，標籤相同
訂閱主題相同，標籤不同

2 訂閱主題不同，標籤相同

當我們啟動兩個消費者後，消費者組名：myconsumerGroup。C1消費者訂閱主題 TopicTest , C2消費者訂閱主題 mytest。

在 Broker 端的紀錄檔裡，會不停的列印拉取訊息失敗的紀錄檔：

2023-10-09 14:52:53 WARN PullMessageThread_2 - 
the consumer's subscription not exist, group: myconsumerGroup, topic:TopicTest

那麼在這種情況下，C1 消費者是不可能拉取到訊息，也就不可能消費到最新的訊息。

為什麼呢？我們知道使用者端會定時的傳送心跳包到 Broker 服務，心跳包中會包含消費者訂閱資訊，資料格式樣例如下：

"subscriptionDataSet": [
  {
    "classFilterMode": false,
    "codeSet": [],
    "expressionType": "TAG",
    "subString": "*",
    "subVersion": 1696832107020,
    "tagsSet": [],
    "topic": "TopicTest"
  },
  {
    "classFilterMode": false,
    "codeSet": [],
    "expressionType": "TAG",
    "subString": "*",
    "subVersion": 1696832098221,
    "tagsSet": [],
    "topic": "%RETRY%myconsumerGroup"
  }
]

Broker 服務會呼叫 ClientManageProcessor 的 heartBeat方法處理心跳請求。

最終跟蹤到程式碼： org.apache.rocketmq.broker.client.ConsumerManager#registerConsumer

Broker 服務的會儲存消費者資訊，消費者資訊儲存在消費者表 consumerTable 。消費者表以消費組名為 key , 值為消費者組資訊 ConsumerGroupInfo 。

#org.apache.rocketmq.broker.client.ConsumerManager
private final ConcurrentMap<String/* Group */, ConsumerGroupInfo> consumerTable =
    new ConcurrentHashMap<String, ConsumerGroupInfo>(1024);

如果消費組的消費者資訊 ConsumerGroupInfo 為空，則新建新的物件。

更新訂閱資訊時，訂閱資訊是按照消費組存放的，這步驟就會導致同一個消費組內的各個消費者使用者端的訂閱資訊相互被覆蓋。

回到消費者使用者端，當消費者拉取訊息時，Broker 服務會呼叫 PullMessageProcessor 的 processRequest 方法。

首先會進行前置判斷，查詢當前的主題的訂閱資訊若該主題的訂閱資訊為空，則列印告警紀錄檔，並返回異常的響應結果。

subscriptionData = consumerGroupInfo.findSubscriptionData(requestHeader.getTopic());    
if (null == subscriptionData) {
     log.warn("the consumer's subscription not exist, group: {}, topic:{}", requestHeader.getConsumerGroup(), 
     response.setCode(ResponseCode.SUBSCRIPTION_NOT_EXIST);
     response.setRemark("the consumer's subscription not exist" + FAQUrl.suggestTodo(FAQUrl.SAME_GROUP_DIFFERENT_TOPIC));
     return response;
}

通過調研 Broker 端的程式碼，我們發現：相同消費組的訂閱資訊必須保持一致 , 否則同一個消費組內的各個消費者使用者端的訂閱資訊相互被覆蓋，從而導致某個消費者使用者端無法拉取到新的訊息。

C1消費者無法消費主題 TopicTest 的訊息資料，那麼 C2 消費者訂閱主題 mytest，消費會正常嗎？

從上圖來看，依然有問題。主題 mytest 有四個佇列，但只有兩個佇列被分配了，另外兩個佇列的訊息就沒有辦法消費了。

要解釋這個問題，我們需要重新溫習負載均衡的原理。

負載均衡服務會根據消費模式為」廣播模式」還是「叢集模式」做不同的邏輯處理，這裡主要來看下叢集模式下的主要處理流程：

(1) 獲取該主題下的訊息消費佇列集合；

(2) 查詢 Broker 端獲取該消費組下消費者 Id 列表；

(3) 先對 Topic 下的訊息消費佇列、消費者 Id 排序，然後用訊息佇列分配策略演演算法（預設為：訊息佇列的平均分配演演算法），計算出待拉取的訊息佇列；

這裡的平均分配演演算法，類似於分頁的演演算法，將所有 MessageQueue 排好序類似於記錄，將所有消費端排好序類似頁數，並求出每一頁需要包含的平均 size 和每個頁面記錄的範圍 range ，最後遍歷整個 range 而計算出當前消費端應該分配到的記錄。

(4) 分配到的訊息佇列集合與 processQueueTable 做一個過濾比對操作。

消費者範例內，processQueueTable 物件儲存著當前負載均衡的佇列，以及該佇列的處理佇列 processQueue (消費快照)。

標紅的 Entry 部分表示與分配到的訊息佇列集合互不包含，則需要將這些紅色佇列 Dropped 屬性為 true , 然後從 processQueueTable 物件中移除。
綠色的 Entry 部分表示與分配到的訊息佇列集合的交集，processQueueTable 物件中已經存在該佇列。
黃色的 Entry 部分表示這些佇列需要新增到 processQueueTable 物件中，為每個分配的新佇列建立一個訊息拉取請求 pullRequest , 在訊息拉取請求中儲存一個處理佇列 processQueue （佇列消費快照），內部是紅黑樹（TreeMap），用來儲存拉取到的訊息。

最後建立拉取訊息請求列表，並將請求分發到訊息拉取服務，進入拉取訊息環節。

通過上面的介紹，通過負載均衡的原理推導，原因就顯而易見了。

C1消費者被分配了佇列 0、佇列 1 ，但是 C1消費者本身並沒有訂閱主題 mytest , 所以無法消費該主題的資料。

從本次實驗來看，C1消費者無法消費主題 TopicTest 的訊息資料 , C2 消費者只能部分消費主題 mytest的訊息資料。

但是因為在 Broker 端，同一個消費組內的各個消費者使用者端的訂閱資訊相互被覆蓋，所以這種消費狀態非常混亂，偶爾也會切換成：C1消費者可以部分消費主題 TopicTest 的訊息資料 , C2消費者無法消費主題 mytest的訊息資料。

3 訂閱主題相同，標籤不同

如圖，C1 消費者和 C2 消費者訂閱主題 TopicTest ，但兩者的標籤 TAG 並不相同。

啟動消費者服務之後，從控制檯觀察，負載均衡的效果也如預期一般正常。

筆者在 Broker 端列印埋點紀錄檔，發現主題 TopicTest 的訂閱資訊為：

{
  "classFilterMode": false,
  "codeSet": [66],
  "expressionType": "TAG",
  "subString": "B",
  "subVersion": 1696901014319,
  "tagsSet": ["B"],
  "topic": "TopicTest"
}

那麼這種狀態，消費正常嗎？筆者做了一組實驗，消費依然混亂：

C1 消費者無法消費 TAG 值為 A 的訊息，C2 消費者只能消費部分 TAG 值為 B 的訊息。

想要理解原因，我們需要梳理訊息過濾機制。

首先 ConsumeQueue 檔案的格式如下：

Broker 端在接收到拉取請求後，根據請求引數定位 ConsumeQueue 檔案，然後遍歷 ConsumeQueue 待檢索的條目，判斷條目中儲存 Tag 的 hashcode 是否和訂閱資訊中 TAG 的 hashcode 是否相同，若不符合，則跳過，繼續對比下一個，符合條件的聚合後返回給消費者使用者端。
消費者在收到過濾後的訊息後，也要執行過濾機制，只不過過濾的是 TAG 字串的值，而不是 hashcode 。

我們模擬下訊息過濾的過程：

首先，生產者將不同的訊息傳送到 Broker 端，不同的 TAG 的訊息會傳送到儲存的不同的佇列中。

C1 消費者從佇列 0 ，佇列 1 中拉取訊息時，因為 Broker 端該主題的訂閱資訊中 TAG 值為 B ，經過伺服器端過濾後， C1 消費者拉取到的訊息的 TAG 值都是 B , 但消費者在收到過濾的訊息後，也需要進行使用者端過濾，A 並不等於 B ，所以 C1 消費者無法消費 TAG 值為 A 的訊息。

C2 消費者從佇列 2，佇列 3 中拉取訊息，整個邏輯鏈路是正常的，但是因為負載均衡的緣故，它無法消費佇列 0 ，佇列 1的訊息。

4 總結

什麼是消費組？消費同一類訊息且消費邏輯一致。RocketMQ 4.X 原始碼實現就是為了和消費組的定義保持一致 。

規避訂閱關係不一致這個問題有兩種方式:

合理定義好主題和標籤

當我們定義好主題和標籤後，需要新增新的標籤時，是否可以換一個思路：換一個新的消費組或者新建一個主題。

嚴格規範上線流程

在上線之前，梳理好相關依賴服務，梳理好上線流程，做好上線評審，並嚴格按照流程執行。

最後的思考：

假如從基礎架構層面來思考，將訂閱關係資訊中心化來設計，應該也可以實現，但成本較高，對於中小企業來講，並不合算。

參考資料：

RocketMQ為什麼要保證訂閱關係的一致性 :

https://cloud.tencent.com/developer/article/1474885

RocketMQ最佳實踐之坑？

https://mp.weixin.qq.com/s/Ypk-U8uVu4aZKMinbfU3xQ

原始碼分析RocketMQ訊息過濾機制

https://blog.csdn.net/prestigeding/article/details/79255328

RocketMQ為什麼要保證訂閱關係一致

1 訂閱關係演示

2 訂閱主題不同，標籤相同

3 訂閱主題相同，標籤不同

4 總結