北京it運(yùn)維服務(wù)商教您深入解讀RocketMQ

2020-04-24 14:13 作者：艾銻無(wú)限 瀏覽量：

 
艾銻無(wú)限科技專業(yè)：IT外包、企業(yè)外包、北京IT外包、桌面運(yùn)維、弱電工程、網(wǎng)站開(kāi)發(fā)、wifi覆蓋方案,網(wǎng)絡(luò)外包,網(wǎng)絡(luò)管理服務(wù),網(wǎng)管外包,綜合布線,服務(wù)器運(yùn)維服務(wù),中小企業(yè)it外包服務(wù),服務(wù)器維保公司,硬件運(yùn)維,網(wǎng)站運(yùn)維服務(wù)
 

 
RocketMQ是一個(gè)純java、分布式、隊(duì)列模型的開(kāi)源消息中間件，前身是MetaQ，是阿里研發(fā)的一個(gè)隊(duì)列模型的消息中間件，后開(kāi)源給apache基金會(huì)成為了apache的頂級(jí)開(kāi)源項(xiàng)目，具有高性能、高可靠、高實(shí)時(shí)、分布式特點(diǎn)，


RocketMQ是一個(gè)純java、分布式、隊(duì)列模型的開(kāi)源消息中間件，前身是MetaQ，是阿里研發(fā)的一個(gè)隊(duì)列模型的消息中間件，后開(kāi)源給apache基金會(huì)成為了apache的頂級(jí)開(kāi)源項(xiàng)目，具有高性能、高可靠、高實(shí)時(shí)、分布式特點(diǎn)，

同時(shí)，廣泛應(yīng)用于多個(gè)領(lǐng)域，包括異步通信解耦、企業(yè)解決方案、金融支付、電信、電子商務(wù)、快遞物流、廣告營(yíng)銷、社交、即時(shí)通信、移動(dòng)應(yīng)用、手游、視頻、物聯(lián)網(wǎng)、車聯(lián)網(wǎng)等。


具有以下特點(diǎn)：

· 能夠保證嚴(yán)格的消息順序

· 提供豐富的消息拉取模式

· 高效的訂閱者水平擴(kuò)展能力

· 實(shí)時(shí)的消息訂閱機(jī)制

· 億級(jí)消息堆積能力

RocketMQ 架構(gòu)原理分析

RocketMQ 架構(gòu)

NameServer (名稱服務(wù)器)：

· 提供輕量級(jí)的服務(wù)發(fā)現(xiàn)和路由。NameServer接受來(lái)自Broker群集的注冊(cè)，并提供檢測(cè)信號(hào)機(jī)制以檢查Broker是否還存在

· 每個(gè)NameServer記錄完整的路由信息(Broker 相關(guān) Topic 等元信息，并給 Producer 提供 Consumer 查找 Broker 信息)，提供相應(yīng)的讀寫服務(wù)。

Broker(消息服務(wù)器)： 消息存儲(chǔ)中心，接收來(lái)自 Producer 的消息并存儲(chǔ)， Consumer 從這里取得消息

· 單個(gè)Broker節(jié)點(diǎn)與所有的NameServer節(jié)點(diǎn)保持長(zhǎng)連接及心跳，并會(huì)定時(shí)將Topic信息注冊(cè)到NameServer，(其底層通信是基于Netty實(shí)現(xiàn)的)

· Broker負(fù)責(zé)消息存儲(chǔ)，以Topic為維度支持輕量級(jí)的隊(duì)列，單機(jī)可以支撐上萬(wàn)隊(duì)列規(guī)模，支持消息推拉模型。

· 具有上億級(jí)消息堆積能力，同時(shí)可嚴(yán)格保證消息的有序性

Producer (生產(chǎn)者)：

· 負(fù)責(zé)產(chǎn)生消息，生產(chǎn)者向消息服務(wù)器發(fā)送由業(yè)務(wù)應(yīng)用程序系統(tǒng)生成的消息

· 生產(chǎn)者支持分布式部署。分布式生產(chǎn)者通過(guò)多種負(fù)載平衡模式將消息發(fā)送到Broker集群。發(fā)送過(guò)程支持快速失敗并且延遲低

· 三種方式發(fā)送消息：同步、異步和單向

Consumer(消費(fèi)者)：

· 負(fù)責(zé)消費(fèi)消息，消費(fèi)者從消息服務(wù)器拉取信息并將其輸入用戶應(yīng)用程序

· 也支持“推和拉”模型中的分布式部署。

· 它還支持集群使用和消息廣播。它提供了實(shí)時(shí)消息訂閱機(jī)制，可以滿足大多數(shù)消費(fèi)者的需求。

Broker Server

Broker Server負(fù)責(zé)消息的存儲(chǔ)和傳遞，消息查詢，HA高可用等，Broker Server幾個(gè)主要模塊組成：



Remoting Module(遠(yuǎn)程模塊)：broker入口，處理來(lái)自客戶端的請(qǐng)求

Client Manager(客戶端管理)：管理client(生產(chǎn)者/消費(fèi)者)并維護(hù)消費(fèi)者的主題訂閱

Store Service(存儲(chǔ)服務(wù))：提供簡(jiǎn)單的API供給數(shù)據(jù)庫(kù)存儲(chǔ)或查詢消息

HA Service(高可用服務(wù))：提供master broker和slave broker之間的數(shù)據(jù)同步功能

Index Service(索引服務(wù))：將message建立索引來(lái)提供快速的查詢能力

RocketMQ 整體流程


1.啟動(dòng) NameServer，NameServer啟動(dòng)后進(jìn)行端口監(jiān)聽(tīng)，等待 Broker、Producer、Consumer 連上來(lái)，相當(dāng)于一個(gè)路由控制中心

2.Broker 啟動(dòng)，跟所有的 Namesrv 保持長(zhǎng)連接，定時(shí)發(fā)送心跳包

· 心跳包中，包含當(dāng)前 Broker 信息(IP+端口等)以及存儲(chǔ)所有 Topic 信息

· 注冊(cè)成功后，Namesrv 集群中就有 Topic 跟 Broker 的映射關(guān)系

3.收發(fā)消息前，先創(chuàng)建 Topic 。創(chuàng)建 Topic 時(shí)，需要指定該 Topic 要存儲(chǔ)在哪些 Broker上。也可以在發(fā)送消息時(shí)自動(dòng)創(chuàng)建Topic

4.Producer 發(fā)送消息

· 啟動(dòng)時(shí)，先跟 Namesrv 集群中的其中一臺(tái)建立長(zhǎng)連接，并從Namesrv 中獲取當(dāng)前發(fā)送的 Topic 存在哪些 Broker 上

· 然后跟對(duì)應(yīng)的 Broker 建立長(zhǎng)連接，直接向 Broker 發(fā)消息

5.Consumer 消費(fèi)消息

· 跟其中一臺(tái) Namesrv 建立長(zhǎng)連接，獲取當(dāng)前訂閱 Topic 存在哪些 Broker 上

· 然后直接跟 Broker 建立連接通道，開(kāi)始消費(fèi)消息*RocketMQ的消息領(lǐng)域模型

RocketMQ Message


Topic(主題)： 表示消息的第一級(jí)類型，是最細(xì)粒度的訂閱單位(生產(chǎn)者傳遞消息和消費(fèi)者提取消息標(biāo)識(shí))

· 一條消息必須有一個(gè)Topic

· 一個(gè)Group可以訂閱多個(gè)Topic的消息

· Topic一般為領(lǐng)域范圍，比如交易消息

Tag(標(biāo)簽)： 表示消息的第二級(jí)類型，可以是使用相同的Topic不同的Tag來(lái)表示同一業(yè)務(wù)模塊的不同任務(wù)的消息，比如交易消息又可以分為：交易創(chuàng)建消息，交易完成消息等

· 助于保持代碼整潔和一致

· 簡(jiǎn)化RocketMQ提供的查詢系統(tǒng)

Message(消息體)： 消息是要傳遞的信息。Message中必須包含一個(gè)Topic，可選Tag和key-vaule鍵值對(duì)

Message Queue(消息隊(duì)列)： 所有消息隊(duì)列都是持久化

· 一個(gè)Topic下可以有多個(gè)Queue

· Queue的引入使得消息的存儲(chǔ)可以分布式集群化，具有了水平擴(kuò)展能力

Group(組)： 分為Producer Group(生產(chǎn)者組)和Consumer Group(消費(fèi)者組)，具有相同角色組成Group

· 原生產(chǎn)者在交易后崩潰，broker可以聯(lián)系同一生產(chǎn)者組的不同生產(chǎn)者實(shí)例以進(jìn)行提交或回退交易。

· 消費(fèi)者組的消費(fèi)者實(shí)例必須具有完全相同的主題訂閱

RocketMQ 特性

Message Model(消息模式)：

· Clustering(集群式)：當(dāng)使用集群消費(fèi)模式時(shí)，MQ 認(rèn)為任意一條消息只需要被集群內(nèi)的任意一個(gè)消費(fèi)者處理即可

· Broadcasting(廣播式)：當(dāng)使用廣播消費(fèi)模式時(shí)，MQ 會(huì)將每條消息推送給集群內(nèi)所有注冊(cè)過(guò)的客戶端，保證消息至少被每臺(tái)機(jī)器消費(fèi)一次

Message Order(消息順序)

· 使用DefaultMQPushConsumer時(shí)，可以決定按順序或同時(shí)使用消息

如果指定按順序使用，則消息使用的最大并發(fā)度是使用者組訂閱的消息隊(duì)列數(shù)

在此模式下不再保證消息順序

· Concurrently：同時(shí)使用消息時(shí)，消息使用的最大并發(fā)性僅受為每個(gè)使用方客戶端指定的線程池限制

· Orderly：有序地使用消息意味著消息的消費(fèi)順序與生產(chǎn)者為每個(gè)消息隊(duì)列發(fā)送消息的順序相同。( 如果要處理必須強(qiáng)制執(zhí)行全局順序的情況，請(qǐng)確保您使用的主題只有一個(gè)消息隊(duì)列)

Message Types(消息類型)

· 事務(wù)消息

· 順序消息

· 延遲消息

RocketMQ單機(jī)版安裝

1.下載編譯源碼

1. # 下載$ 

2. > wget wget http://mirror.bit.edu.cn/apache/rocketmq/4.6.0/rocketmq-all-4.6.0-source- > 

3. # 解壓$ 

4. >unzip rocketmq-all-4.7.0-source-release.zip 

5. > cd rocketmq-all-4.7.0/ 

6. # 編譯$ 

7. > mvn -Prelease-all -DskipTests clean install -U 

8. > cd distribution/target/rocketmq-4.7.0/rocketmq-4.7.0 

2.啟動(dòng) Name Server

1. # 啟動(dòng) Name Server 服務(wù) 

2. > nohup sh bin/mqnamesrv & 

3. # 啟動(dòng)完成后，查看日志$ 

4. > tail -f ~/logs/rocketmqlogs/namesrv.log 

5. The Name Server boot success... 

3.啟動(dòng) Broker

在 conf 目錄下，RocketMQ 提供了多種 Broker 的配置文件：

1. # 啟動(dòng) Broker服務(wù) 

2. > nohup sh bin/mqbroker -n localhost:9876 & 

3. # 啟動(dòng)完成后，查看日志$ 

4. > tail -f ~/logs/rocketmqlogs/broker.log 

5. The broker[%s, 172.30.30.233:10911] boot success... 

其中，參數(shù)：

· 通過(guò) -c 參數(shù)，配置讀取的主 Broker 配置

· 通過(guò) -n 參數(shù)，設(shè)置 RocketMQ Namesrv 地址

· broker.conf ：?jiǎn)沃?，異步刷盤。

· 2m/ ：雙主，異步刷盤。

· 2m-2s-async/ ：兩主兩從，異步復(fù)制，異步刷盤。

· 2m-2s-sync/ ：兩主兩從，同步復(fù)制，異步刷盤。

· dledger/ ：Dledger 集群，至少三節(jié)點(diǎn)

4.Send & Receive Messages(消息發(fā)送與接收)

在發(fā)送/接收消息之前，我們需要告知client(生產(chǎn)者/消費(fèi)者)Name Servers的地址。RocketMQ提供了多種方法來(lái)實(shí)現(xiàn)：

為簡(jiǎn)單起見(jiàn)，我們使用環(huán)境變量：NAMESRV_ADDR，如下所示：

1. # 設(shè)置 Name Servers的地址$ 

2. > export NAMESRV_ADDR=localhost:9876 

3. # 生產(chǎn)消息$ 

4. > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer 

5. SendResult [sendStatus=SEND_OK, msgId= ... 

6. # 消費(fèi)消息$ 

7. > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer 

8. ConsumeMessageThread_%d Receive New Messages: [MessageExt... 

· 在代碼中設(shè)置：producer.setNamesrvAddr("ip:port")

· java屬性配置：rocketmq.namesrv.addr

· 環(huán)境變量配置：NAMESRV_ADDR

· HTTP Endpoint

以上文章由北京艾銻無(wú)限科技發(fā)展有限公司整理