Find N3入網(wǎng)：最高支持16+1TB
OPPO將于近期登場的Find N3折疊屏目前已經(jīng)正式入網(wǎng)，型號(hào)為PHN110。本次Find N3在外觀方面相比前兩代有很大的變化，不再是小號(hào)的橫向折疊屏，而是跟別的廠商一樣采用了較為常見的
這篇文章，筆者想聊聊 RocketMQ 最佳實(shí)踐之一：保證訂閱關(guān)系一致。
訂閱關(guān)系一致指的是同一個(gè)消費(fèi)者 Group ID 下所有 Consumer 實(shí)例所訂閱的 Topic 、Tag 必須完全一致。
如果訂閱關(guān)系不一致，消息消費(fèi)的邏輯就會(huì)混亂，甚至導(dǎo)致消息丟失。
首先我們展示正確的訂閱關(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)組，兩種不正確的場景，看看消費(fèi)者和 Broker 服務(wù)有什么異常。
訂閱主題不同，標(biāo)簽相同
訂閱主題相同，標(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.ConsumerManagerprivate 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)者客戶端無法拉取到新的消息。
C1消費(fèi)者無法消費(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ì)列；
圖片
這里的平均分配算法，類似于分頁的算法，將所有 MessageQueue 排好序類似于記錄，將所有消費(fèi)端排好序類似頁數(shù)，并求出每一頁需要包含的平均 size 和每個(gè)頁面記錄的范圍 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)快照)。
最后創(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 , 所以無法消費(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 端，同一個(gè)消費(fèi)組內(nèi)的各個(gè)消費(fèi)者客戶端的訂閱信息相互被覆蓋，所以這種消費(fèi)狀態(tài)非常混亂，偶爾也會(huì)切換成：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 并不相同。
啟動(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)者無法消費(fèi) TAG 值為 A 的消息 ，C2 消費(fèi)者只能消費(fèi)部分 TAG 值為 B 的消息。
想要理解原因，我們需要梳理消息過濾機(jī)制。
首先 ConsumeQueue 文件的格式如下 ：
圖片
我們模擬下消息過濾的過程：
圖片
首先，生產(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)者無法消費(fèi) TAG 值為 A 的消息。
C2 消費(fèi)者從隊(duì)列 2， 隊(duì)列 3 中拉取消息，整個(gè)邏輯鏈路是正常的 ，但是因?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í)行邏輯就會(huì)出現(xiàn)混亂。
規(guī)避訂閱關(guān)系不一致這個(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ī)制
https://blog.csdn/prestigeding/article/details/79255328
本文鏈接：http://www.tebozhan.com/showinfo-26-13354-0.html兩個(gè)實(shí)驗(yàn)讓我徹底弄懂了「訂閱關(guān)系一致」
聲明：本網(wǎng)頁內(nèi)容旨在傳播知識(shí)，若有侵權(quán)等問題請(qǐng)及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。郵件：2376512515@qq.com
Copyright ? 2016-2023 天津谷騏科技有限公司 版權(quán)所有 sitemap.xml
違法及侵權(quán)請(qǐng)聯(lián)系：2376512515@qq.com 津ICP備18001702號(hào)
津公網(wǎng)安備 12010102000574號(hào)