影音體驗(yàn)是真的強(qiáng) 簡單聊聊iQOO Pad
大公司的好處就是產(chǎn)品線豐富，非常細(xì)分化的東西也能給你做出來，例如早先我們看到了新的vivo Pad2，之后我們又在iQOO Neo8 Pro的發(fā)布會上看到了iQOO的首款平板產(chǎn)品iQOO Pad。雖
這篇文章，筆者想聊聊 RocketMQ 最佳實(shí)踐之一：保證訂閱關(guān)系一致。
訂閱關(guān)系一致指的是同一個消費(fèi)者 Group ID 下所有 Consumer 實(shí)例所訂閱的 Topic 、Tag 必須完全一致。
如果訂閱關(guān)系不一致，消息消費(fèi)的邏輯就會混亂，甚至導(dǎo)致消息丟失。
首先我們展示正確的訂閱關(guān)系：多個 Group ID 訂閱了多個 Topic，并且每個 Group ID 里的多個消費(fèi)者的訂閱關(guān)系保持了一致。
正確的訂閱關(guān)系
接下來，我們展示錯誤的訂閱關(guān)系。
錯誤的訂閱關(guān)系
從上圖中，單個 Group ID 訂閱了多個 Topic，但是該 Group ID 里的多個消費(fèi)者的訂閱關(guān)系并沒有保持一致。
代碼邏輯角度來看，每個消費(fèi)者實(shí)例內(nèi)訂閱方法的主題、 TAG、監(jiān)聽邏輯都需要保持一致。
圖片
接下來，我們實(shí)驗(yàn)相同消費(fèi)組，兩種不正確的場景，看看消費(fèi)者和 Broker 服務(wù)有什么異常。
訂閱主題不同，標(biāo)簽相同
訂閱主題相同，標(biāo)簽不同
圖片
當(dāng)我們啟動兩個消費(fèi)者后，消費(fèi)者組名：myconsumerGroup。
C1消費(fèi)者訂閱主題 TopicTest , C2消費(fèi)者訂閱主題 mytest。
在 Broker 端的日志里，會不停的打印拉取消息失敗的日志 ：
2023-10-09 14:52:53 WARN PullMessageThread_2 - the consumer's subscription not exist, group: myconsumerGroup, topic:TopicTest那么在這種情況下，C1 消費(fèi)者是不可能拉取到消息，也就不可能消費(fèi)到最新的消息。
為什么呢 ？我們知道客戶端會定時的發(fā)送心跳包到 Broker 服務(wù)，心跳包中會包含消費(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ù)會調(diào)用 ClientManageProcessor 的 heartBeat方法處理心跳請求。
最終跟蹤到代碼：org.apache.rocketmq.broker.client.ConsumerManager#registerConsumer
圖片
Broker 服務(wù)的會保存消費(fèi)者信息，消費(fèi)者信息存儲在消費(fèi)者表 consumerTable 。消費(fèi)者表以消費(fèi)組名為 key , 值為消費(fèi)者組信息 ConsumerGroupInfo 。
#org.apache.rocketmq.broker.client.ConsumerManagerprivate final ConcurrentMap<String/* Group */, ConsumerGroupInfo> consumerTable = new ConcurrentHashMap<String, ConsumerGroupInfo>(1024);如果消費(fèi)組的消費(fèi)者信息 ConsumerGroupInfo 為空，則新建新的對象。
更新訂閱信息時，訂閱信息是按照消費(fèi)組存放的，這步驟就會導(dǎo)致同一個消費(fèi)組內(nèi)的各個消費(fèi)者客戶端的訂閱信息相互被覆蓋。
回到消費(fèi)者客戶端，當(dāng)消費(fèi)者拉取消息時，Broker 服務(wù)會調(diào)用 PullMessageProcessor 的 processRequest 方法 。
首先會進(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)組的訂閱信息必須保持一致 , 否則同一個消費(fèi)組內(nèi)的各個消費(fèi)者客戶端的訂閱信息相互被覆蓋，從而導(dǎo)致某個消費(fèi)者客戶端無法拉取到新的消息。
C1消費(fèi)者無法消費(fèi)主題 TopicTest 的消息數(shù)據(jù)，那么 C2 消費(fèi)者訂閱主題 mytest，消費(fèi)會正常嗎 ？
圖片
從上圖來看，依然有問題。主題 mytest 有四個隊(duì)列，但只有兩個隊(duì)列被分配了， 另外兩個隊(duì)列的消息就沒有辦法消費(fèi)了。
要解釋這個問題，我們需要重新溫習(xí)負(fù)載均衡的原理。
負(fù)載均衡服務(wù)會根據(jù)消費(fèi)模式為”廣播模式”還是“集群模式”做不同的邏輯處理，這里主要來看下集群模式下的主要處理流程：
(1) 獲取該主題下的消息消費(fèi)隊(duì)列集合；
(2) 查詢 Broker 端獲取該消費(fèi)組下消費(fèi)者 Id 列表；
(3) 先對 Topic 下的消息消費(fèi)隊(duì)列、消費(fèi)者 Id 排序，然后用消息隊(duì)列分配策略算法（默認(rèn)為：消息隊(duì)列的平均分配算法），計算出待拉取的消息隊(duì)列；
圖片
這里的平均分配算法，類似于分頁的算法，將所有 MessageQueue 排好序類似于記錄，將所有消費(fèi)端排好序類似頁數(shù)，并求出每一頁需要包含的平均 size 和每個頁面記錄的范圍 range ，最后遍歷整個 range 而計算出當(dāng)前消費(fèi)端應(yīng)該分配到的記錄。
(4) 分配到的消息隊(duì)列集合與 processQueueTable 做一個過濾比對操作。
圖片
消費(fèi)者實(shí)例內(nèi) ，processQueueTable 對象存儲著當(dāng)前負(fù)載均衡的隊(duì)列 ，以及該隊(duì)列的處理隊(duì)列 processQueue (消費(fèi)快照)。
最后創(chuàng)建拉取消息請求列表，并將請求分發(fā)到消息拉取服務(wù)，進(jìn)入拉取消息環(huán)節(jié)。
通過上面的介紹 ，通過負(fù)載均衡的原理推導(dǎo)，原因就顯而易見了。
圖片
C1消費(fèi)者被分配了隊(duì)列 0、隊(duì)列 1 ，但是 C1消費(fèi)者本身并沒有訂閱主題 mytest , 所以無法消費(fèi)該主題的數(shù)據(jù)。
從本次實(shí)驗(yàn)來看，C1消費(fèi)者無法消費(fèi)主題 TopicTest 的消息數(shù)據(jù) , C2 消費(fèi)者只能部分消費(fèi)主題 mytest的消息數(shù)據(jù)。
但是因?yàn)樵?Broker 端，同一個消費(fèi)組內(nèi)的各個消費(fèi)者客戶端的訂閱信息相互被覆蓋，所以這種消費(fèi)狀態(tài)非常混亂，偶爾也會切換成：C1消費(fèi)者可以部分消費(fèi)主題 TopicTest 的消息數(shù)據(jù) , C2消費(fèi)者無法消費(fèi)主題 mytest的消息數(shù)據(jù)。
圖片
如圖，C1 消費(fèi)者和 C2 消費(fèi)者訂閱主題 TopicTest ，但兩者的標(biāo)簽 TAG 并不相同。
啟動消費(fèi)者服務(wù)之后，從控制臺觀察，負(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)者無法消費(fèi) TAG 值為 A 的消息 ，C2 消費(fèi)者只能消費(fèi)部分 TAG 值為 B 的消息。
想要理解原因，我們需要梳理消息過濾機(jī)制。
首先 ConsumeQueue 文件的格式如下 ：
圖片
我們模擬下消息過濾的過程：
圖片
首先，生產(chǎn)者將不同的消息發(fā)送到 Broker 端，不同的 TAG 的消息會發(fā)送到保存的不同的隊(duì)列中。
C1 消費(fèi)者從隊(duì)列 0 ，隊(duì)列 1 中拉取消息時，因?yàn)?Broker 端該主題的訂閱信息中 TAG 值為 B ，經(jīng)過服務(wù)端過濾后， C1 消費(fèi)者拉取到的消息的 TAG 值都是 B , 但消費(fèi)者在收到過濾的消息后，也需要進(jìn)行客戶端過濾，A 并不等于 B ，所以 C1 消費(fèi)者無法消費(fèi) TAG 值為 A 的消息。
C2 消費(fèi)者從隊(duì)列 2， 隊(duì)列 3 中拉取消息，整個邏輯鏈路是正常的 ，但是因?yàn)樨?fù)載均衡的緣故，它無法消費(fèi)隊(duì)列 0 ，隊(duì)列 1的消息。
什么是消費(fèi)組 ？消費(fèi)同一類消息且消費(fèi)邏輯一致 。
RocketMQ 4.X 源碼實(shí)現(xiàn)就是為了和消費(fèi)組的定義保持一致 ，假如訂閱關(guān)系不一致，那么代碼執(zhí)行邏輯就會出現(xiàn)混亂。
規(guī)避訂閱關(guān)系不一致這個問題有兩種方式 :
最后的思考：
假如從基礎(chǔ)架構(gòu)層面來思考，將訂閱關(guān)系信息中心化來設(shè)計，應(yīng)該也可以實(shí)現(xiàn) ，但成本較高，對于中小企業(yè)來講，并不合算。
參考資料：
RocketMQ為什么要保證訂閱關(guān)系的一致性
https://cloud.tencent.com/developer/article/1474885
RocketMQ最佳實(shí)踐之坑？
https://mp.weixin.qq.com/s/Ypk-U8uVu4aZKMinbfU3xQ
源碼分析RocketMQ消息過濾機(jī)制
https://blog.csdn/prestigeding/article/details/79255328
本文鏈接：http://www.tebozhan.com/showinfo-26-13354-0.html兩個實(shí)驗(yàn)讓我徹底弄懂了「訂閱關(guān)系一致」
聲明：本網(wǎng)頁內(nèi)容旨在傳播知識，若有侵權(quán)等問題請及時與本網(wǎng)聯(lián)系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
Copyright ? 2016-2023 天津谷騏科技有限公司 版權(quán)所有 sitemap.xml
違法及侵權(quán)請聯(lián)系：2376512515@qq.com 津ICP備18001702號
津公網(wǎng)安備 12010102000574號