Apache RocketMQ 是阿里開源的一款高性能、高吞吐量的分布式消息中間件。

整體架構(gòu)

RocketMQ主要由 Producer、Broker、Consumer 三部分組成，其中Producer 負(fù)責(zé)生產(chǎn)消息，Consumer 負(fù)責(zé)消費(fèi)消息，Broker 負(fù)責(zé)存儲(chǔ)消息。每個(gè) Broker 可以存儲(chǔ)多個(gè)Topic的消息，每個(gè)Topic的消息也可以分片存儲(chǔ)于集群中的不同的Broker Group。

Namesrv

說道Namesrv首先會(huì)想到服務(wù)注冊與發(fā)現(xiàn)。分布式服務(wù)SOA架構(gòu)體系中會(huì)有服務(wù)注冊與發(fā)現(xiàn)中心。主要作用是指導(dǎo)服務(wù)調(diào)用方找到服務(wù)提供者提供的服務(wù)實(shí)例。RocketMQ體系中Namesrv主要作用是：為producer和consumer提供關(guān)于topic的路由信息。管理broker節(jié)點(diǎn)：監(jiān)控更新broker的實(shí)時(shí)狀態(tài)。路由注冊、路由刪除（故障剔除）。

Namesrv充當(dāng)路由消息的提供者。Namesrv是一個(gè)幾乎無狀態(tài)節(jié)點(diǎn)，多個(gè)Namesrv實(shí)例組成集群，但相互獨(dú)立，沒有信息交換。

路由元信息

topicQueueTable：topic 消息隊(duì)列路由信息。

brokerAddrTable：broker基礎(chǔ)信息。包含broker name，所屬集群名稱，主broker地址等。

clusterAddrTable：broker集群信息，存儲(chǔ)集群中所有broker的名稱。

brokerLiveTable：broker狀態(tài)信息。

filterServerTable：broker上的filterServer列表。filterServer用于消息過濾。

路由注冊 RocketMQ路由注冊是通過broker與Namesrv的心跳功能實(shí)現(xiàn)的。broker啟動(dòng)時(shí)向集群中所有Namesrv發(fā)送心跳包，之后每隔30秒向集群中所有Namesrv發(fā)送心跳包。心跳包中包含：broker集群信息、broker信息、topic配置信息、broker關(guān)聯(lián)的FilterServer列表等。如果brokerA為Master。并且brokerA上的topic1的配置信息發(fā)生變化或初次注冊，Namesrv會(huì)根據(jù)報(bào)文創(chuàng)建或更新Topic路由元數(shù)據(jù)，填充topicQueueTable。

路由刪除 Namesrv收到brokerA的心跳包會(huì)更新brokerLiveTable中的brokerA對(duì)應(yīng)的BrokerLiveInfo中的lastUpdateTimestamp。Namesrv每隔10秒掃描brokerLiveTable一次。如果brokerA對(duì)應(yīng)的BrokerLiveInfo 中 lastUpdateTimestamp距當(dāng)前時(shí)間超過 120秒，Namesrv認(rèn)為brokerA失效，會(huì)將brokerA的路由信息移除并關(guān)閉與broker的socket連接。更新：topicQueueInfo、brokerAddrTable、brokerLiveTable、filterServerTable等。

路由發(fā)現(xiàn) RocketMQ路由發(fā)現(xiàn)是非實(shí)時(shí)的。當(dāng)Topic路由信息發(fā)生變化是，Namesrv不會(huì)主動(dòng)推送給客戶端（Producer、Consumer）。而是由客戶端定時(shí)到Namesrv拉去最新的路由信息并緩存（包含Topic路由信息）。

與kafka對(duì)比

kafka 由zookeeper集群提供命名服務(wù)（Naming Service）。

Kafka通過 ZooKeeper 管理集群配置、選舉 Leader 以及在 consumer g

Broker

消息中轉(zhuǎn)角色，負(fù)責(zé)存儲(chǔ)消息、轉(zhuǎn)發(fā)消息。代理服務(wù)器在RocketMQ系統(tǒng)中負(fù)責(zé)接收從生產(chǎn)者發(fā)送來的消息并存儲(chǔ)、同時(shí)為消費(fèi)者的拉取請求作準(zhǔn)備。代理服務(wù)器也存儲(chǔ)消息相關(guān)的元數(shù)據(jù)，包括消費(fèi)者組、消費(fèi)進(jìn)度偏移和主題和隊(duì)列消息等。

Broker是以group為單位提供服務(wù)。一個(gè)group里面分Master和Slave。Master和Slave存儲(chǔ)的數(shù)據(jù)一樣，slave從master同步數(shù)據(jù)（同步雙寫或異步復(fù)制看配置）。一個(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與Namesrv集群中的所有節(jié)點(diǎn)建立長連接，定時(shí)發(fā)送心跳包到所有Namesrv，更新broker信息、topic路由信息等。一個(gè)Topic的不同queue（分區(qū)）可分布到集群中不同的broker group上。

與kafka對(duì)比：

kafka和RocketMQ的broker都可以容納多個(gè)一個(gè)或多個(gè)分區(qū)數(shù)據(jù)（kafka分區(qū)：partition；RocketMQ分區(qū)：queue）。

kafka基于partition（分區(qū)） 做備份/高可用（partition follower）。

RocketMQ增加了broker group的概念，基于broker（可能包含多個(gè)分區(qū)）。

Producer

（消息）生產(chǎn)者。Producer與Namesrv集群中的其中一個(gè)節(jié)點(diǎn)（隨機(jī)選擇）建立長連接，定期從Name Server取Topic路由信息，并向提供Topic服務(wù)的broker master建立長連接，且定時(shí)向broker master發(fā)送心跳。Producer完全無狀態(tài)，可集群部署。

Producer負(fù)責(zé)生產(chǎn)消息，一般由業(yè)務(wù)系統(tǒng)負(fù)責(zé)生產(chǎn)消息。一個(gè)消息生產(chǎn)者會(huì)把業(yè)務(wù)應(yīng)用系統(tǒng)里產(chǎn)生的消息發(fā)送到broker服務(wù)器。RocketMQ提供多種發(fā)送方式，同步發(fā)送、異步發(fā)送、順序發(fā)送、單向發(fā)送。同步和異步方式均需要Broker返回確認(rèn)信息，單向發(fā)送不需要。

Consumer

（消息）消費(fèi)者 Consumer與Namesrv集群中的其中一個(gè)節(jié)點(diǎn)（隨機(jī)選擇）建立長連接，定期從Name Server取Topic路由信息，并向提供Topic服務(wù)的Master、Slave建立長連接，且定時(shí)向Master、Slave發(fā)送心跳。Consumer既可以從Master訂閱消息，也可以從Slave訂閱消息，訂閱規(guī)則由Broker配置決定。

Consumer負(fù)責(zé)消費(fèi)消息，一般是后臺(tái)系統(tǒng)負(fù)責(zé)異步消費(fèi)。一個(gè)消息消費(fèi)者會(huì)從Broker服務(wù)器拉取消息、并將其提供給應(yīng)用程序。從用戶應(yīng)用的角度而言提供了兩種消費(fèi)形式：拉取式消費(fèi)、推動(dòng)式消費(fèi)。

集群模式下：相同Consumer Group的每個(gè)Consumer實(shí)例平均分?jǐn)傁ⅰＲ粋€(gè)條消息僅能被一個(gè)Consumer Group消費(fèi)一次。

Producer、Consumer都只需要和集群中一個(gè)Namesrv建立長連接。Broker需要向集群中所有的Namesrv發(fā)送心跳包。 其實(shí)很好理解： Namesrv集群提供高可用的命名服務(wù)。 Producer、Consumer只需要從其中一臺(tái)定期同步路由信息。 如果Broker只隨機(jī)調(diào)一臺(tái)發(fā)送心跳包。那么不同的Namesrv保存的路由信息會(huì)出現(xiàn)

消費(fèi)者類型：

拉取式消費(fèi)（Pull Consumer） Consumer消費(fèi)的一種類型，應(yīng)用通常主動(dòng)調(diào)用Consumer的拉消息方法從Broker服務(wù)器拉消息、主動(dòng)權(quán)由應(yīng)用控制。一旦獲取了批量消息，應(yīng)用就會(huì)啟動(dòng)消費(fèi)過程。Pull方式里，取消息的過程需要用戶自己寫（包括提交offset等操作）。

推動(dòng)式消費(fèi)（Push Consumer） Consumer消費(fèi)的一種類型，該模式下Broker收到數(shù)據(jù)后會(huì)主動(dòng)推送給消費(fèi)端，該消費(fèi)模式一般實(shí)時(shí)性較高。Push Consumer原理上也是采取pull模式。實(shí)際上就是長輪詢的pull模式。

一些概念

主題（Topic） 表示一類消息的集合，每個(gè)主題包含若干條消息，每條消息只能屬于一個(gè)主題，是RocketMQ進(jìn)行消息訂閱的基本單位。每個(gè)topic可分為若干個(gè)分區(qū)（queue）。

生產(chǎn)者組（Producer Group） 同一類Producer的集合，這類Producer發(fā)送同一類消息且發(fā)送邏輯一致。如果發(fā)送的是事務(wù)消息且原始生產(chǎn)者在發(fā)送之后崩潰，則Broker服務(wù)器會(huì)聯(lián)系同一生產(chǎn)者組的其他生產(chǎn)者實(shí)例以提交或回溯消費(fèi)。

消費(fèi)者組（Consumer Group） 同一類Consumer的集合，這類Consumer通常消費(fèi)同一類消息且消費(fèi)邏輯一致。消費(fèi)者組使得在消息消費(fèi)方面，實(shí)現(xiàn)負(fù)載均衡和容錯(cuò)的目標(biāo)變得非常容易。要注意的是，消費(fèi)者組的消費(fèi)者實(shí)例必須訂閱完全相同的Topic。RocketMQ 支持兩種消息模式：集群消費(fèi)（Clustering）和廣播消費(fèi)（Broadcasting）。

普通順序消息（Normal Ordered Message） 普通順序消費(fèi)模式下，消費(fèi)者通過同一個(gè)消費(fèi)隊(duì)列收到的消息是有順序的，不同消息隊(duì)列收到的消息則可能是無順序的。

嚴(yán)格順序消息（Strictly Ordered Message） 嚴(yán)格順序消息模式下，消費(fèi)者收到的所有消息均是有順序的。

消息（Message） 消息系統(tǒng)所傳輸信息的物理載體，生產(chǎn)和消費(fèi)數(shù)據(jù)的最小單位，每條消息必須屬于一個(gè)主題。RocketMQ中每個(gè)消息擁有唯一的Message ID，且可以攜帶具有業(yè)務(wù)標(biāo)識(shí)的Key。系統(tǒng)提供了通過Message ID和Key查詢消息的功能。

標(biāo)簽（Tag） 為消息設(shè)置的標(biāo)志，用于同一主題下區(qū)分不同類型的消息。來自同一業(yè)務(wù)單元的消息，可以根據(jù)不同業(yè)務(wù)目的在同一主題下設(shè)置不同標(biāo)簽。標(biāo)簽?zāi)軌蛴行У乇３执a的清晰度和連貫性，并優(yōu)化RocketMQ提供的查詢系統(tǒng)。消費(fèi)者可以根據(jù)Tag實(shí)現(xiàn)對(duì)不同子主題的不同消費(fèi)邏輯，實(shí)現(xiàn)更好的擴(kuò)展性。

關(guān)于消息中間件

消息中間件需要解決的問題：異步化、削峰填谷。

消息中間件應(yīng)具備的基礎(chǔ)能力是：消息發(fā)布、訂閱、消費(fèi)。概念相對(duì)簡單這里不過多描述。

消息中間件的一些重要的機(jī)制：

1. 消息優(yōu)先級(jí)（Message Priority；RocketMQ不支持）

優(yōu)先級(jí)是指在一個(gè)消息隊(duì)列中，每條消息都有不同的優(yōu)先級(jí)，一般用整數(shù)來描述，優(yōu)先級(jí)高的消息先投遞，如果消息完全在一個(gè)內(nèi)存隊(duì)列中，那么在投遞前可以按照優(yōu)先級(jí)排序，令優(yōu)先級(jí)高的先投遞。由于RocketMQ所有消息都是持久化的，所以如果按照優(yōu)先級(jí)來排序，開銷會(huì)非常大，因此RocketMQ沒有特意支持消息優(yōu)先級(jí)，但是可以通過變通的方式實(shí)現(xiàn)類似功能，即單獨(dú)配置一個(gè)優(yōu)先級(jí)高的隊(duì)列，和一個(gè)普通優(yōu)先級(jí)的隊(duì)列，將不同優(yōu)先級(jí)發(fā)送到不同隊(duì)列即可。

2. 順序消息（Message Order）

消息有序指的是一類消息消費(fèi)時(shí)，能按照發(fā)送的順序來消費(fèi)。例如：一個(gè)訂單產(chǎn)生了3條消息，分別是訂單創(chuàng)建，訂單付款，訂單完成。消費(fèi)時(shí)，要按照這個(gè)順序消費(fèi)才能有意義。但是同時(shí)訂單之間是可以并行消費(fèi)的。RocketMQ可以嚴(yán)格的保證消息有序。

投遞消息的順序性：投遞消息的順序性可通過將一組消息投遞到同一分區(qū)實(shí)現(xiàn)。例如：借助MessageQueueSelector將對(duì)相同訂單的操作消息投放到同一分區(qū)。

消費(fèi)消息的順序性：RoctetMQ特性保障：特定分區(qū)（queue）中的消息不能同時(shí)被同一個(gè)消費(fèi)者組中的多個(gè)Consumer消費(fèi)，以避免重復(fù)消費(fèi)。通過自定義或使用預(yù)置的AllocateQueueStrategy可設(shè)定分區(qū)的分配策略（哪些分區(qū)分配給哪個(gè)消費(fèi)者消費(fèi)）。

3. 高可用、消息可靠性

3.1 消息持久化

RocketMQ、Kafka 以文件記錄形式持久化。

RocketMQ采用了單一的日志文件，即把同1個(gè)broker上面所有topic的所有queue的消息，存放在一個(gè)文件里面，從而避免了隨機(jī)的磁盤寫入。

如上圖所示，所有消息都存在一個(gè)單一的CommitLog文件里面，然后有后臺(tái)線程異步的同步到ConsumeQueue，再由Consumer進(jìn)行消費(fèi)。

TODO 同步、異步刷盤。

TODO RocketMQ充分利用Linux文件系統(tǒng)內(nèi)存cache來提高性能。TODO CommitLog index Commitlog segment的大小與頁緩存一致。

RocketMQ消息存儲(chǔ)機(jī)制會(huì)在后面的文章詳細(xì)說明。

3.2 broker master/salve

TODO broker group master/salveTODO Async/Sync Master；

4. 高并發(fā)、可擴(kuò)展 ==》 分布式

提高并發(fā)效率 =》 提高生產(chǎn)、消費(fèi)并行度=》提高分區(qū)數(shù)量。

RocketMQ、kafka都支持topic數(shù)據(jù)分區(qū)存放、動(dòng)態(tài)擴(kuò)展。

以RocketMQ為例：

topic創(chuàng)建的時(shí)候可以用集群模式去創(chuàng)建（這樣集群里面每個(gè)broker的queue的數(shù)量相同），也可以用單個(gè)broker模式去創(chuàng)建（這樣每個(gè)broker的queue數(shù)量可以不一致）。

4.1 生產(chǎn)并行度

RocketMQ的生產(chǎn)并行度是由其自身機(jī)制及broker的數(shù)量決定的。這塊后面的文章會(huì)詳細(xì)分析。

4.2 消費(fèi)并行度

廣播模式下所有消費(fèi)者會(huì)接受并消費(fèi)當(dāng)前topic下所有Queue的消息。

集群模式下，一個(gè)queue只分配給一個(gè)consumer實(shí)例：這是由于拉取消息是consumer主動(dòng)控制的，如果多個(gè)實(shí)例同時(shí)消費(fèi)一個(gè)queue的消息，會(huì)導(dǎo)致同一個(gè)消息在不同的實(shí)例下被消費(fèi)多次，所以算法上都是一個(gè)queue只分給一個(gè)consumer實(shí)例，一個(gè)consumer實(shí)例可以允許同時(shí)分到不同的queue。

Kafka的消費(fèi)并行度依賴Topic配置的分區(qū)數(shù)，如分區(qū)數(shù)為10，那么最多10臺(tái)機(jī)器來并行消費(fèi)（每臺(tái)機(jī)器只能開啟一個(gè)線程），或者一臺(tái)機(jī)器消費(fèi)（10個(gè)線程并行消費(fèi)）。即消費(fèi)并行度和分區(qū)數(shù)一致。RocketMQ消費(fèi)并行度分兩種情況：順序消費(fèi)方式并行度同卡夫卡完全一致；亂序方式并行度取決于Consumer的線程數(shù)，如Topic配置10個(gè)隊(duì)列，10臺(tái)機(jī)器消費(fèi)，每臺(tái)機(jī)器100個(gè)線程，那么并行度為1000。

4.3 消息隊(duì)列分配策略

Producer使用MessageQueueSelector選擇將消息投放到哪個(gè)分區(qū) 使用AllocateMessageQueueStrategy將不同分區(qū)分配給Consumer Group中的不同Consumer。一個(gè)分區(qū)（queue）僅允許分配給同一個(gè)Consumer Group下的一個(gè)Consumer（防止重復(fù)消費(fèi)）。

MessageQueueSelector

內(nèi)置實(shí)現(xiàn)類：SelectMessageQueueByMachineRoom SelectMessageQueueByHash SelectMessageQueueByRandom

可以通過實(shí)現(xiàn)MessageQueueSelector接口，來自定義Producer投遞消息時(shí)選擇分區(qū)的算法。

AllocateMessageQueueStrategy

內(nèi)置實(shí)現(xiàn)類：

AllocateMessageQueueAveragely：平均分配算法 AllocateMessageQueueAveragelyByCircle：基于環(huán)形平均分配算法AllocateMachineRoomNearby：基于機(jī)房臨近原則算法AllocateMessageQueueByMachineRoom：基于機(jī)房分配算法AllocateMessageQueueConsistentHash：基于一致性hash算法AllocateMessageQueueByConfig：基于配置分配算法

可以通過實(shí)現(xiàn)AllocateMessageQueueStrategy來自定義queue 分配給特定Consumer Group下不同Consumer的策略。

參考：

https://github.com/apache/rocketmq/blob/master/docs/cn/https://juejin.im/post/6844903589819875336https://jaskey.github.io/blog/2016/12/19/rocketmq-rebalance/http://objcoding.com/2019/09/13/kafka-partition-and-rmq-queue/http://www.itmuch.com/books/rocketmq

作者：RyanLee86799

來源：https://juejin.im/post/6844904130822029320

文章轉(zhuǎn)載：JAVA高級(jí)架構(gòu)

（版權(quán)歸原作者所有，侵刪）

編輯：jq