兩個(gè)實(shí)驗(yàn)讓我徹底弄懂了「訂閱關(guān)系一致」
這篇文章，筆者想聊聊 RocketMQ 最佳實(shí)踐之一：保證訂閱關(guān)系一致。
訂閱關(guān)系一致指的是同一個(gè)消費(fèi)者 Group ID 下所有 Consumer 實(shí)例所訂閱的 Topic 、Tag 必須完全一致。
如果訂閱關(guān)系不一致，消息消費(fèi)的邏輯就會(huì)混亂，甚至導(dǎo)致消息丟失。
1 訂閱關(guān)系演示
首先我們展示正確的訂閱關(guān)系：多個(gè) Group ID 訂閱了多個(gè) Topic，并且每個(gè) Group ID 里的多個(gè)消費(fèi)者的訂閱關(guān)系保持了一致。
正確的訂閱關(guān)系
接下來，我們展示錯(cuò)誤的訂閱關(guān)系。
錯(cuò)誤的訂閱關(guān)系
從上圖中，單個(gè) Group ID 訂閱了多個(gè) Topic，但是該 Group ID 里的多個(gè)消費(fèi)者的訂閱關(guān)系并沒有保持一致。
代碼邏輯角度來看，每個(gè)消費(fèi)者實(shí)例內(nèi)訂閱方法的主題、 TAG、監(jiān)聽邏輯都需要保持一致。
圖片
接下來，我們實(shí)驗(yàn)相同消費(fèi)組，兩種不正確的場(chǎng)景，看看消費(fèi)者和 Broker 服務(wù)有什么異常。
訂閱主題不同，標(biāo)簽相同
訂閱主題相同，標(biāo)簽不同
2 訂閱主題不同，標(biāo)簽相同
圖片
當(dāng)我們啟動(dòng)兩個(gè)消費(fèi)者后，消費(fèi)者組名：myconsumerGroup。
C1消費(fèi)者訂閱主題 TopicTest , C2消費(fèi)者訂閱主題 mytest。
在 Broker 端的日志里，會(huì)不停的打印拉取消息失敗的日志 ：
2023-10-09 14:52:53 WARN PullMessageThread_2 -
the consumer's subscription not exist, group: myconsumerGroup, topic:TopicTest那么在這種情況下，C1 消費(fèi)者是不可能拉取到消息，也就不可能消費(fèi)到最新的消息。
為什么呢 ？我們知道客戶端會(huì)定時(shí)的發(fā)送心跳包到 Broker 服務(wù)，心跳包中會(huì)包含消費(fèi)者訂閱信息，數(shù)據(jù)格式樣例如下：
"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 服務(wù)會(huì)調(diào)用 ClientManageProcessor 的 heartBeat方法處理心跳請(qǐng)求。
最終跟蹤到代碼：org.apache.rocketmq.broker.client.ConsumerManager#registerConsumer
圖片
Broker 服務(wù)的會(huì)保存消費(fèi)者信息，消費(fèi)者信息存儲(chǔ)在消費(fèi)者表 consumerTable 。消費(fèi)者表以消費(fèi)組名為 key , 值為消費(fèi)者組信息 ConsumerGroupInfo 。
#org.apache.rocketmq.broker.client.ConsumerManager
private final ConcurrentMap<String/* Group */, ConsumerGroupInfo> consumerTable =
new ConcurrentHashMap<String, ConsumerGroupInfo>(1024);如果消費(fèi)組的消費(fèi)者信息 ConsumerGroupInfo 為空，則新建新的對(duì)象。
更新訂閱信息時(shí)，訂閱信息是按照消費(fèi)組存放的，這步驟就會(huì)導(dǎo)致同一個(gè)消費(fèi)組內(nèi)的各個(gè)消費(fèi)者客戶端的訂閱信息相互被覆蓋。
回到消費(fèi)者客戶端，當(dāng)消費(fèi)者拉取消息時(shí)，Broker 服務(wù)會(huì)調(diào)用 PullMessageProcessor 的 processRequest 方法 。
首先會(huì)進(jìn)行前置判斷，查詢當(dāng)前的主題的訂閱信息若該主題的訂閱信息為空，則打印告警日志，并返回異常的響應(yīng)結(jié)果。
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;
}通過調(diào)研 Broker 端的代碼，我們發(fā)現(xiàn)：相同消費(fèi)組的訂閱信息必須保持一致 , 否則同一個(gè)消費(fèi)組內(nèi)的各個(gè)消費(fèi)者客戶端的訂閱信息相互被覆蓋，從而導(dǎo)致某個(gè)消費(fèi)者客戶端無(wú)法拉取到新的消息。
C1消費(fèi)者無(wú)法消費(fèi)主題 TopicTest 的消息數(shù)據(jù)，那么 C2 消費(fèi)者訂閱主題 mytest，消費(fèi)會(huì)正常嗎 ？
圖片
從上圖來看，依然有問題。主題 mytest 有四個(gè)隊(duì)列，但只有兩個(gè)隊(duì)列被分配了， 另外兩個(gè)隊(duì)列的消息就沒有辦法消費(fèi)了。
要解釋這個(gè)問題，我們需要重新溫習(xí)負(fù)載均衡的原理。
負(fù)載均衡服務(wù)會(huì)根據(jù)消費(fèi)模式為”廣播模式”還是“集群模式”做不同的邏輯處理，這里主要來看下集群模式下的主要處理流程：
(1) 獲取該主題下的消息消費(fèi)隊(duì)列集合；
(2) 查詢 Broker 端獲取該消費(fèi)組下消費(fèi)者 Id 列表；
(3) 先對(duì) Topic 下的消息消費(fèi)隊(duì)列、消費(fèi)者 Id 排序，然后用消息隊(duì)列分配策略算法（默認(rèn)為：消息隊(duì)列的平均分配算法），計(jì)算出待拉取的消息隊(duì)列；
圖片
這里的平均分配算法，類似于分頁(yè)的算法，將所有 MessageQueue 排好序類似于記錄，將所有消費(fèi)端排好序類似頁(yè)數(shù)，并求出每一頁(yè)需要包含的平均 size 和每個(gè)頁(yè)面記錄的范圍 range ，最后遍歷整個(gè) range 而計(jì)算出當(dāng)前消費(fèi)端應(yīng)該分配到的記錄。
(4) 分配到的消息隊(duì)列集合與 processQueueTable 做一個(gè)過濾比對(duì)操作。
圖片
消費(fèi)者實(shí)例內(nèi) ，processQueueTable 對(duì)象存儲(chǔ)著當(dāng)前負(fù)載均衡的隊(duì)列 ，以及該隊(duì)列的處理隊(duì)列 processQueue (消費(fèi)快照)。
- 標(biāo)紅的 Entry 部分表示與分配到的消息隊(duì)列集合互不包含，則需要將這些紅色隊(duì)列 Dropped 屬性為 true , 然后從 processQueueTable 對(duì)象中移除。
- 綠色的 Entry 部分表示與分配到的消息隊(duì)列集合的交集，processQueueTable 對(duì)象中已經(jīng)存在該隊(duì)列。
- 黃色的 Entry 部分表示這些隊(duì)列需要添加到 processQueueTable 對(duì)象中，為每個(gè)分配的新隊(duì)列創(chuàng)建一個(gè)消息拉取請(qǐng)求 pullRequest , 在消息拉取請(qǐng)求中保存一個(gè)處理隊(duì)列 processQueue （隊(duì)列消費(fèi)快照），內(nèi)部是紅黑樹（TreeMap），用來保存拉取到的消息。
最后創(chuàng)建拉取消息請(qǐng)求列表，并將請(qǐng)求分發(fā)到消息拉取服務(wù)，進(jìn)入拉取消息環(huán)節(jié)。
通過上面的介紹 ，通過負(fù)載均衡的原理推導(dǎo)，原因就顯而易見了。
圖片
C1消費(fèi)者被分配了隊(duì)列 0、隊(duì)列 1 ，但是 C1消費(fèi)者本身并沒有訂閱主題 mytest , 所以無(wú)法消費(fèi)該主題的數(shù)據(jù)。
從本次實(shí)驗(yàn)來看，C1消費(fèi)者無(wú)法消費(fèi)主題 TopicTest 的消息數(shù)據(jù) , C2 消費(fèi)者只能部分消費(fèi)主題 mytest的消息數(shù)據(jù)。
但是因?yàn)樵?Broker 端，同一個(gè)消費(fèi)組內(nèi)的各個(gè)消費(fèi)者客戶端的訂閱信息相互被覆蓋，所以這種消費(fèi)狀態(tài)非常混亂，偶爾也會(huì)切換成：C1消費(fèi)者可以部分消費(fèi)主題 TopicTest 的消息數(shù)據(jù) , C2消費(fèi)者無(wú)法消費(fèi)主題 mytest的消息數(shù)據(jù)。
3 訂閱主題相同，標(biāo)簽不同
圖片
如圖，C1 消費(fèi)者和 C2 消費(fèi)者訂閱主題 TopicTest ，但兩者的標(biāo)簽 TAG 并不相同。
啟動(dòng)消費(fèi)者服務(wù)之后，從控制臺(tái)觀察，負(fù)載均衡的效果也如預(yù)期一般正常。
圖片
筆者在 Broker 端打印埋點(diǎn)日志，發(fā)現(xiàn)主題 TopicTest 的訂閱信息為 ：
{
"classFilterMode": false,
"codeSet": [66],
"expressionType": "TAG",
"subString": "B",
"subVersion": 1696901014319,
"tagsSet": ["B"],
"topic": "TopicTest"
}那么這種狀態(tài)，消費(fèi)正常嗎 ？筆者做了一組實(shí)驗(yàn)，消費(fèi)依然混亂：
C1 消費(fèi)者無(wú)法消費(fèi) TAG 值為 A 的消息 ，C2 消費(fèi)者只能消費(fèi)部分 TAG 值為 B 的消息。
想要理解原因，我們需要梳理消息過濾機(jī)制。
首先 ConsumeQueue 文件的格式如下 ：
圖片
- Broker 端在接收到拉取請(qǐng)求后，根據(jù)請(qǐng)求參數(shù)定位 ConsumeQueue 文件，然后遍歷 ConsumeQueue 待檢索的條目， 判斷條目中存儲(chǔ) Tag 的 hashcode 是否和訂閱信息中 TAG 的 hashcode 是否相同，若不符合，則跳過，繼續(xù)對(duì)比下一個(gè)， 符合條件的聚合后返回給消費(fèi)者客戶端。
- 消費(fèi)者在收到過濾后的消息后，也要執(zhí)行過濾機(jī)制，只不過過濾的是 TAG 字符串的值，而不是 hashcode 。
我們模擬下消息過濾的過程：
圖片
首先，生產(chǎn)者將不同的消息發(fā)送到 Broker 端，不同的 TAG 的消息會(huì)發(fā)送到保存的不同的隊(duì)列中。
C1 消費(fèi)者從隊(duì)列 0 ，隊(duì)列 1 中拉取消息時(shí)，因?yàn)?Broker 端該主題的訂閱信息中 TAG 值為 B ，經(jīng)過服務(wù)端過濾后， C1 消費(fèi)者拉取到的消息的 TAG 值都是 B , 但消費(fèi)者在收到過濾的消息后，也需要進(jìn)行客戶端過濾，A 并不等于 B ，所以 C1 消費(fèi)者無(wú)法消費(fèi) TAG 值為 A 的消息。
C2 消費(fèi)者從隊(duì)列 2， 隊(duì)列 3 中拉取消息，整個(gè)邏輯鏈路是正常的 ，但是因?yàn)樨?fù)載均衡的緣故，它無(wú)法消費(fèi)隊(duì)列 0 ，隊(duì)列 1的消息。
4 總結(jié)
什么是消費(fèi)組 ？消費(fèi)同一類消息且消費(fèi)邏輯一致 。
RocketMQ 4.X 源碼實(shí)現(xiàn)就是為了和消費(fèi)組的定義保持一致 ，假如訂閱關(guān)系不一致，那么代碼執(zhí)行邏輯就會(huì)出現(xiàn)混亂。
規(guī)避訂閱關(guān)系不一致這個(gè)問題有兩種方式 :
- 嚴(yán)格規(guī)范上線流程在上線之前，梳理好相關(guān)依賴服務(wù)，梳理好上線流程，做好上線評(píng)審，并嚴(yán)格按照流程執(zhí)行。
- 合理定義好主題和標(biāo)簽當(dāng)我們定義好主題和標(biāo)簽后，需要添加新的標(biāo)簽時(shí)，是否可以換一個(gè)思路：換一個(gè)新的消費(fèi)組或者新建一個(gè)主題。
最后的思考：
假如從基礎(chǔ)架構(gòu)層面來思考，將訂閱關(guān)系信息中心化來設(shè)計(jì)，應(yīng)該也可以實(shí)現(xiàn) ，但成本較高，對(duì)于中小企業(yè)來講，并不合算。
參考資料：
RocketMQ為什么要保證訂閱關(guān)系的一致性
https://cloud.tencent.com/developer/article/1474885
RocketMQ最佳實(shí)踐之坑？
https://mp.weixin.qq.com/s/Ypk-U8uVu4aZKMinbfU3xQ
源碼分析RocketMQ消息過濾機(jī)制
