消息的順序消費(fèi)在很多交易型的業(yè)務(wù)場景中都會(huì)被要求實(shí)現(xiàn),而且,消息隊(duì)列的順序消費(fèi)解決方案在很多互聯(lián)網(wǎng)公司的面試中經(jīng)常會(huì)被問到。
在使用了多個(gè)消息隊(duì)列后發(fā)現(xiàn),雖然每個(gè)消息隊(duì)列都有各自的順序消費(fèi)解決方案,但是RocketMQ經(jīng)過了多年電商的洗禮,其功能性的要求,已經(jīng)設(shè)計(jì)的非常全面。這樣的全面可以通過RocketMQ消息模型的架構(gòu)設(shè)計(jì)得以體現(xiàn)。我們看看RocketMQ是怎么解決消息的順序消費(fèi)。
一、RocketMQ的消息模型
1.技術(shù)架構(gòu)
RocketMQ架構(gòu)上主要分為四部分,如上圖所示:
NameServer:NameServer是一個(gè)非常簡單的Topic路由注冊(cè)中心,其角色類似Dubbo中的zookeeper,支持Broker的動(dòng)態(tài)注冊(cè)與發(fā)現(xiàn)。主要包括兩個(gè)功能:Broker管理,NameServer接受Broker集群的注冊(cè)信息并且保存下來作為路由信息的基本數(shù)據(jù)。然后提供心跳檢測機(jī)制,檢查Broker是否還存活;路由信息管理,每個(gè)NameServer將保存關(guān)于Broker集群的整個(gè)路由信息和用于客戶端查詢的隊(duì)列信息。然后Producer和Conumser通過NameServer就可以知道整個(gè)Broker集群的路由信息,從而進(jìn)行消息的投遞和消費(fèi)。NameServer通常也是集群的方式部署,各實(shí)例間相互不進(jìn)行信息通訊。Broker是向每一臺(tái)NameServer注冊(cè)自己的路由信息,所以每一個(gè)NameServer實(shí)例上面都保存一份完整的路由信息。當(dāng)某個(gè)NameServer因某種原因下線了,Broker仍然可以向其它NameServer同步其路由信息,Producer,Consumer仍然可以動(dòng)態(tài)感知Broker的路由的信息。
Producer:消息的發(fā)布者,支持分布式集群方式部署。消息發(fā)布者通過消息隊(duì)列的負(fù)載均衡模塊選擇相應(yīng)的Broker集群進(jìn)行消息投遞,投遞的過程低延遲并且支持快速失敗。
Consumer:消息的消費(fèi)者,支持分布式集群方式部署。RocketMQ同時(shí)支持push推和pull拉的兩種消息消費(fèi)模式。消息消費(fèi)者也支持集群和廣播兩種消費(fèi)方式,同時(shí)提供了實(shí)時(shí)消息的訂閱機(jī)制,滿足大多數(shù)用戶的需求。
BrokerServer:Broker主要負(fù)責(zé)消息的存儲(chǔ)、投遞和查詢以及服務(wù)高可用保證,為了實(shí)現(xiàn)這些功能,Broker包含了以下幾個(gè)重要子模塊。
Remoting Module:整個(gè)Broker的實(shí)體,負(fù)責(zé)處理來自clients端的請(qǐng)求。
Client Manager:負(fù)責(zé)管理客戶端(Producer/Consumer)和維護(hù)Consumer的Topic訂閱信息。
Store Service:提供方便簡單的API接口處理消息存儲(chǔ)到物理硬盤和查詢功能。
HA Service:高可用服務(wù),提供Master Broker 和 Slave Broker之間的數(shù)據(jù)同步功能。
Index Service:根據(jù)特定的Message key對(duì)投遞到Broker的消息進(jìn)行索引服務(wù),以提供消息的快速查詢。
2. 部署架構(gòu)
RocketMQ 網(wǎng)絡(luò)部署特點(diǎn)
NameServer是一個(gè)幾乎無狀態(tài)節(jié)點(diǎn),可集群部署,節(jié)點(diǎn)之間無任何信息同步。
Broker部署相對(duì)復(fù)雜,Broker分為Master與Slave,一個(gè)Master可以對(duì)應(yīng)多個(gè)Slave,但是一個(gè)Slave只能對(duì)應(yīng)一個(gè)Master,Master與Slave 的對(duì)應(yīng)關(guān)系通過指定相同的BrokerName,不同的BrokerId 來定義,BrokerId為0表示Master,非0表示Slave。Master也可以部署多個(gè)。每個(gè)Broker與NameServer集群中的所有節(jié)點(diǎn)建立長連接,定時(shí)注冊(cè)Topic信息到所有NameServer。注意:當(dāng)前RocketMQ版本在部署架構(gòu)上支持一Master多Slave,但只有BrokerId=1的從服務(wù)器才會(huì)參與消息的讀負(fù)載。
Producer與NameServer集群中的其中一個(gè)節(jié)點(diǎn)(隨機(jī)選擇)建立長連接,定期從NameServer獲取Topic路由信息,并向提供Topic 服務(wù)的Master建立長連接,且定時(shí)向Master發(fā)送信息。Producer完全無狀態(tài),可集群部署。
Consumer與NameServer集群中的其中一個(gè)節(jié)點(diǎn)(隨機(jī)選擇)建立長連接,定期從NameServer獲取Topic路由信息,并向提供Topic服務(wù)的Master、Slave建立長連接,且定時(shí)向Master、Slave發(fā)送信息。Consumer既可以從Master訂閱消息,也可以從Slave訂閱消息,消費(fèi)者在向Master拉取消息時(shí),Master服務(wù)器會(huì)根據(jù)拉取偏移量與最大偏移量的距離(判斷是否讀老消息,產(chǎn)生讀I/O),以及從服務(wù)器是否可讀等因素建議下一次是從Master還是Slave拉取。
結(jié)合部署架構(gòu)圖,描述集群工作流程:
啟動(dòng)NameServer,NameServer起來后監(jiān)聽端口,等待Broker、Producer、Consumer連上來,相當(dāng)于一個(gè)路由控制中心。
Broker啟動(dòng),跟所有的NameServer保持長連接,定時(shí)發(fā)送心跳包。心跳包中包含當(dāng)前Broker信息(IP+端口等)以及存儲(chǔ)所有Topic信息。注冊(cè)成功后,NameServer集群中就有Topic跟Broker的映射關(guān)系。
收發(fā)消息前,先創(chuàng)建Topic,創(chuàng)建Topic時(shí)需要指定該Topic要存儲(chǔ)在哪些Broker上,也可以在發(fā)送消息時(shí)自動(dòng)創(chuàng)建Topic。
Producer發(fā)送消息,啟動(dòng)時(shí)先跟NameServer集群中的其中一臺(tái)建立長連接,并從NameServer中獲取當(dāng)前發(fā)送的Topic存在哪些Broker上,輪詢從隊(duì)列列表中選擇一個(gè)隊(duì)列,然后與隊(duì)列所在的Broker建立長連接從而向Broker發(fā)消息。
Consumer跟Producer類似,跟其中一臺(tái)NameServer建立長連接,獲取當(dāng)前訂閱Topic存在哪些Broker上,然后直接跟Broker建立連接通道,開始消費(fèi)消息。
二、順序消息的應(yīng)用場景
順序消費(fèi)的應(yīng)用場景有很多,無論是在面試,或是實(shí)際生產(chǎn)環(huán)境中要解決順序消費(fèi)問題,都需要對(duì)消息隊(duì)列的應(yīng)用場景有一定的理解。
1.AIOT中消息的順序消費(fèi)
比如有這么一個(gè)物聯(lián)網(wǎng)的應(yīng)用場景,IOT中的設(shè)備在初始化時(shí)需要按順序接收這樣的消息:
操作1:設(shè)置設(shè)備名稱
操作2:設(shè)置設(shè)備的網(wǎng)絡(luò)
操作3:重啟設(shè)備使配置生效
如果這個(gè)順序顛倒了,可能就沒有辦法讓設(shè)備的配置生效,因?yàn)橹挥兄貑⒃O(shè)備才能讓配置生效,但重啟的消息卻在設(shè)置設(shè)備消息之前被消費(fèi)。
2.用戶服務(wù)中的順序消費(fèi)
應(yīng)用為了提高用戶粘合度,往往通過積分制度提供用戶活躍度。比如接下來有這一系列的場景:
操作1:新注冊(cè)用戶,將用戶積分設(shè)置為10分。
操作2:獎(jiǎng)勵(lì)行為,比如用戶完善了個(gè)人信息,則積分+5分。
操作3:懲罰行為,比如用戶發(fā)表違規(guī)言論,則積分-3分。
如果上述的操作是順序進(jìn)行的,則用戶積分為12分。如果上述操作的步驟順序發(fā)生了變化,比如消費(fèi)者先消費(fèi)了操作2的消息,再消費(fèi)操作1和3的消息,則分?jǐn)?shù)變?yōu)?分,此時(shí)就出現(xiàn)了因?yàn)閬y序消費(fèi)導(dǎo)致的錯(cuò)誤結(jié)果。
三、如何實(shí)現(xiàn)順序消息
順序消息要求消費(fèi)者消費(fèi)消息的順序按照發(fā)送者發(fā)送消息的順序執(zhí)行。RocketMQ中實(shí)現(xiàn)的消息順序有兩個(gè)維度,分別是局部有序和全局有序。
1.局部有序
局部消息指的是消費(fèi)者消費(fèi)某個(gè)topic的某個(gè)隊(duì)列中的消息是順序的。在圖上的八個(gè)隊(duì)列中,消費(fèi)者可以隨機(jī)的消費(fèi)這8個(gè)隊(duì)列,也就是說消費(fèi)者不能保證消費(fèi)隊(duì)列的順序。但是消費(fèi)者在消費(fèi)某一個(gè)隊(duì)列的時(shí)候,一定可以進(jìn)行順序消費(fèi)。也就是說,在消費(fèi)者兩次訪問一個(gè)隊(duì)列中的消息時(shí),一定是按照從左到右的順序依次消費(fèi)消息。所以針對(duì)于某一個(gè)隊(duì)列來講,消費(fèi)是順序的。但如果把問題定位在整個(gè)隊(duì)列中時(shí),不同的消息在不同的隊(duì)列中,又不能保證消息的有序性,不如消息A到了隊(duì)列2,消息B到了隊(duì)列1,消費(fèi)者先消費(fèi)了隊(duì)列1再消費(fèi)隊(duì)列2,就不能保證有序性。但如果消息C隨著消息B進(jìn)入隊(duì)列1,那么消息C一定是在消息B之后被消費(fèi)。
消費(fèi)者使用MessageListenerOrderly類做消息監(jiān)聽,實(shí)現(xiàn)局部順序。
2.全局有序
消費(fèi)者消費(fèi)全部消息都是順序的,只能通過一個(gè)某個(gè)topic只有一個(gè)隊(duì)列才能實(shí)現(xiàn),這種應(yīng)用場景較少,且性能較差。
3.亂序消費(fèi)
消費(fèi)者消費(fèi)消息不需要關(guān)注消息的順序。消費(fèi)者使用MessageListenerConcurrently類做消息監(jiān)聽。