RocketMQ 詳解系列

什么是RocketMQ

RocketMQ作為一款純java、分布式、隊列模型的開源消息中間件,支持事務(wù)消息、順序消息、批量消息、定時消息、消息回溯等。主要功能是異步解耦和流量削峰:。

常見的MQ主要有:ActiveMQ、RabbitMQ、Kafka、RocketMQ

四種消息中間件的基本介紹:

特性ActiveMQRabbitMQRocketMQKafka
單機吞吐量萬級,比RocketMQ和Kafka第一個級別同ActiveMQ10萬級,支撐高吞吐10萬級,高吞吐,一般配合大數(shù)據(jù)類的系統(tǒng)進行實時數(shù)據(jù)計算、日志采集等場景
topic數(shù)量對吞吐量的影響topic可以達(dá)到幾百/幾千級別,吞吐量會有較小幅度的下降,這是RocketMQ的一大優(yōu)勢,在同等機器下,可以支撐大量的topictopic從幾十到幾百時,吞吐量會大幅度下降,在同等機器下,kafka盡量保證topic數(shù)量不要過多,如果要支撐大規(guī)模的topic,需要增加更多的機器資源
時效性ms級微秒級別,RabbitMQ的特性,延遲最低ms級別延遲在ms級別以內(nèi)
可用性高,基于主從架構(gòu)實現(xiàn)高可用同ActiveMQ非常高,分布式架構(gòu)非常高,分布式一個數(shù)據(jù)多個副本,少數(shù)機器宕機,不會丟失數(shù)據(jù),不會導(dǎo)致不可用
消息可靠性有較低的概率丟失數(shù)據(jù)基本不丟經(jīng)過參數(shù)優(yōu)化配置,可以做到0丟失經(jīng)過參數(shù)優(yōu)化配置,可以做到0丟失
功能支持MQ領(lǐng)域的功能機器完備基于erlang開發(fā),并發(fā)能力很強,性能極好,延時很低MQ功能較為完善,基本分布式,擴展性好功能較簡單,主要支持簡單的MQ功能,在大數(shù)據(jù)領(lǐng)域的實時計算以及日志采集被大規(guī)模使用
其他Apache開發(fā),起步早,沒有經(jīng)過高吞吐場景驗證,社區(qū)不活躍開源、穩(wěn)定、社區(qū)活躍度高阿里開源,交給Apache,社區(qū)活躍度低Apache開發(fā),開源、高吞吐量、社區(qū)活躍度高

消息中間件的使用場景:

異步與解耦:

當(dāng)我們下了一個訂單之后,訂單系統(tǒng)會進行RPC同步調(diào)用 支付系統(tǒng)、庫存系統(tǒng)、物流系統(tǒng)等,那么系統(tǒng)之間就會有耦合性,耦合性越高的話,容錯性就越低,比如我們的支付系統(tǒng)如果宕機了,就會導(dǎo)致我們整個交易的異常,從而影響用戶的體驗。

如果我們中間加入了消息中間件,不管是支付還是庫存等系統(tǒng),都是通過異步的方式進行調(diào)用的,如果其中一個系統(tǒng)宕機了,不會影響我們用戶下單的使用。

本質(zhì)上MQ第一步完成了 異步 ,第二步完成了 解耦 。那么系統(tǒng)的容錯性就越高。

流量削峰:

流量削峰也可以叫削峰填谷,比如一些互聯(lián)網(wǎng)公司大促場景,雙十一、店慶或者秒殺活動,都會使用到消息中間件。

如果在不使用消息中間件或者沒有流量削峰,每秒是很高的并發(fā),這個時候如果我們的A系統(tǒng),如果要將數(shù)據(jù)寫入到我們的MYSQL中,受限于MYSQL本身服務(wù)的上限,最大我們只能每秒處理200請求,這個時候會有大量的消息進行堆積,從而導(dǎo)致A系統(tǒng)的奔潰。

這個時候我們可以將用戶的請求消息通過MQ進行寫入,因為消息中間件本身是對數(shù)據(jù)量處理比較高的一個系統(tǒng),所以對于每秒2000請求,消息中間件可以處理,然后A系統(tǒng)作為消息中間件的一個消費者,以固定的速度從MQ中拉取200個消息,完成我們的業(yè)務(wù)操作,用時間換空間 從而確保我們A系統(tǒng)的穩(wěn)定性。

數(shù)據(jù)分發(fā):

如果S系統(tǒng),在對系統(tǒng)進行開發(fā)的時候,需要對接多個(A、B、C、D)系統(tǒng),使用傳統(tǒng)的接口調(diào)用,中間有改動就需要修改我們的代碼,當(dāng)新增了A系統(tǒng),我們需要去修改代碼去調(diào)用A系統(tǒng)來完成對應(yīng)的業(yè)務(wù)邏輯,如果我們當(dāng)中的D系統(tǒng)需要移除, 同樣也需要修改代碼刪除對應(yīng)的接口調(diào)用。

如果S系統(tǒng)使用了消息中間件,我們S系統(tǒng)只要將消息交給MQ,剩下的不論是新增還是移除,還是原有的,他們都只是消息中間件的一個消費者,這個時候我們就便于數(shù)據(jù)的分發(fā)。

比如我們新增一個系統(tǒng),我們只需要新增一個MQ的消費者,直接從MQ里面拿消息就可以,當(dāng)我們需要移除一個系統(tǒng)的時候,只需要取消對MQ消息的監(jiān)聽即可。對于我們原有的S系統(tǒng)不需要進行額外的修改。如果使用MQ作為數(shù)據(jù)分發(fā),減少數(shù)據(jù)的修改,提高開發(fā)的效率。

RocketMQ 基本概念

RocketMQ主要有四大核心組成部分:NameServer、Broker、Producer以及Consumer四部分。這些角色通常以集群的方式存在,RocketMQ 基于純Java開發(fā),具有高吞吐量、高可用性、適合大規(guī)模分布式系統(tǒng)應(yīng)用的特點。

對于 RockerMQ 而言,我們想要啟動,必須首先啟動 NameServer,在啟動 Brober 主機, Brober 會向 NameServer 注冊對應(yīng)的路由和服務(wù)(Broker 地址、主體和),Producer會進行路由的發(fā)現(xiàn),向NameServer請求Broker路由信息,進行消息的發(fā)送。

作為Consumer要連通NameServer,獲取到相關(guān)的路由信息,方便我們進行消息的訂閱。

Broker 也是一個很重要的角色,主要負(fù)責(zé)消息的存儲,不管是生產(chǎn)消息還是訂閱消息,消息的來源都是 Broker,一般來說消息的發(fā)送(Producer)只會發(fā)到主節(jié)點,然后Broker會進行消息的同步,同步到從節(jié)點,作為消費者(Consumer)也只會優(yōu)先從Master節(jié)點,獲取消息,進行消費,除非主節(jié)點不可用或者非常繁忙,才會從從節(jié)點進行消費,Broker除了消息的中轉(zhuǎn),還負(fù)責(zé)消息的持久化以及主從數(shù)據(jù)之間的復(fù)制

NameServer:

NameServer 是一個服務(wù)與注冊的發(fā)現(xiàn)中心。也是整個 RocketMQ 的“大腦”,所以 RocketMQ 需要先啟動 NameServer 再啟動 RocketMQ 中的 Broker

NameServer 是一個幾乎無狀態(tài)節(jié)點,可集群部署,節(jié)點之間無任何信息同步。NameServer底層由 Netty 實現(xiàn),是內(nèi)存式存儲,所以 NameServer 中的 broker、topic不會持久化。

NameServer 其角色類似Dubbo和zookeeper,主要負(fù)責(zé)Broker的動態(tài)注冊與發(fā)現(xiàn)。為什么不使用zookeeper?rocketmq主要是在分布式情況下使用追求性能,因為zookeeper最求最終一致性,所以在性能上會有所折扣。






Broker:

消息服務(wù)器(Broker)是消息存儲中心,主要作用是接收來自 Producer 的消息并存儲,Consumer 從這里取得消息。存儲與消息相關(guān)的元數(shù)據(jù),包括用戶組、消費進度偏移量、隊列信息等。從部署結(jié)構(gòu)圖中可以看出 BrokerMasterSlave 兩種類型, Master 既可以寫又可以讀,Slave 不可以寫只可以讀。

Producer:

Producer 也稱為消息發(fā)布者(生產(chǎn)者),負(fù)責(zé)生產(chǎn)并發(fā)送消息至 Topic。生產(chǎn)者向 broker 發(fā)送由業(yè)務(wù)應(yīng)用程序系統(tǒng)生成的消息。RocketMQ 提供了發(fā)送:同步、異步和單向(one-way)的多種范例。

Consumer:

也稱為消息訂閱者,負(fù)責(zé)從 Topic 接收并消費消息。消費者從 brokers 那里拉取信息并將其輸入應(yīng)用程序。從Master拿到消息,執(zhí)行完成后,會發(fā)送一個消息給Broker進行確認(rèn),這個就是ACK確認(rèn)

RocketMQ 基本概念

分組(Group)

Group 分為兩個部分 生產(chǎn)者和消費者

  • 生產(chǎn)者: 表示發(fā)送同一類消息的 Producer,通常情況下發(fā)送邏輯是一致的。發(fā)送普通消息時,用于標(biāo)識使用,沒有特別的用處。

    主要用來作用于事務(wù)消息,當(dāng)事務(wù)消息中某條消息一直處于等待狀態(tài)并超時,Broker會回查同一個Group下的其他producer,確定該消息是 commit 還是 rollback

  • 消費者: 消費者的分組就非常有意義了,消費者是標(biāo)識一類 Consumer 的集合名稱,這類 Consumer 通常消費一類消息,且消費邏輯一致。同一個 Consumer Group 下的各個實例將共同消費 topic 的消息,起到負(fù)載均衡的作用。

    消費進度以 Consumer Group 為粒度管理,不同 Consumer Group 之間消費進度彼此不受影響,即消息 A 被 Consumer Group1 消費過,也會再給 Consumer Group2 消費。

主體(Topic)

用來區(qū)分消息的種類,表示一類消息的邏輯名字,消息的邏輯管理單位,無論生產(chǎn)還是消費消息,都需要執(zhí)行Topic。

一個發(fā)送者可以發(fā)送消息給一個或者多個Topic;

一個消息接受者可以訂閱一個或多個Topic消息;

消息隊列(Message Queue)

消息隊列 簡稱 Queue ,消息物理管理單位。用來并行發(fā)送和接收消息,相當(dāng)于是Topic的分區(qū)。

一個Topic會有若干個Queue,消息的生產(chǎn)一般會比消息消費的速度要快,消息進行消費的時會有對應(yīng)的業(yè)務(wù)邏輯進行處理,這個時候就會降低消息消費的速度。所有一般Topic會有若干個Queue。主要用來解決生產(chǎn)很快,消費很慢。

如果同一個Topic創(chuàng)建在不同的Broker,那么不同的Broker有不同的Queue,將物理存儲在不同的Broker節(jié)點之上,具有水平擴展的能力。無論是生產(chǎn)者還是消費者,實際的操作都是針對Queue級別。

標(biāo)簽(Tag)

RocketMQ 支持在發(fā)送時給 topic 的消息設(shè)置 tag,用于同一主題下區(qū)分不同類型的消息。

來自同一業(yè)務(wù)單元的消息,可以根據(jù)不同業(yè)務(wù)目的在同一主題下設(shè)置不同標(biāo)簽。比如有一個 Topic 消息為水果,那么水果可以有其他的標(biāo)簽 可以是 香蕉、西瓜、草莓等等,我們可以把對應(yīng)的消息,打上對應(yīng)的標(biāo)簽(Tag),這個就是方便我們在消費的時候做對應(yīng)的篩選。

標(biāo)簽?zāi)軌蛴行У乇3执a的清晰度和連貫性,并優(yōu)化 RocketMQ 提供的查詢系統(tǒng)。消費者可以根據(jù)Tag實現(xiàn)對不同子主題的不同消費邏輯,實現(xiàn)更好的擴展性。

偏移量(Offset)

在 RocketMQ 中,有很多 offset 的概念。一般我們只關(guān)心暴露到客戶端的 offset。不指定的話,一般指的是消費者消息的偏移量(ConsumerOffset)

Message queue 是無限長的數(shù)組。一條消息進來下標(biāo)就會漲 1,而這個數(shù)組的下標(biāo)就是 offset。

Message queue 中的 max offset 表示消息的最大 offset,Consumer offset 可以理解為標(biāo)記 Consumer Group 在一條邏輯 Message Queue 上,消息消費到哪里即消費進度。

RocketMQ 下載安裝

下載地址:https://rocketmq.apache.org/dowloading/releases/

環(huán)境要求:

  • Windows/Linux 64位系統(tǒng)
  • JDK1.8(64位)
  • 源碼安裝需要安裝Maven 3.2.x

這里我們用 rocketmq-4.9.2 來做演示案例。
設(shè)置環(huán)境變量:

變量名: ROCKETMQ_HOME

變量值: MQ解壓路徑\MQ文件夾名

啟動

rocketmq-4.9.2\bin目錄下,打開cmd窗口

先啟動 nameServer,啟動命令:start mqnamesrv.cmd

然后在啟動 Broker,啟動命令:start mqbroker.cmd -n 127.0.0.1:7906 autoCreateTopicEnable=true

管理端插件安裝:

老版本地址下載:https://codeload.github.com/apache/rocketmq-externals/zip/master

新版本地址:https://github.com/apache/rocketmq-dashboard

啟動完成之后,瀏覽器中輸入‘127.0.0.1:8089’,成功后即可進行管理端查看。

消息發(fā)送

RocketMQ提供的原生客戶端的API,當(dāng)然除了原生客戶端外,SpringBoot、SpringCloudStream也進行了集成,但本質(zhì)上這些也是基于原生API的封裝,所以只需掌握原生API,其他的也會水到渠成。

導(dǎo)入MQ客戶端依賴

<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-client</artifactId>
  <version>4.9.2</version>
</dependency>

消息發(fā)送:

/**
 * 同步發(fā)送
 */
public class SyncProducer {
    public static void main(String[] args) throws Exception{
        // 實例化消息生產(chǎn)者Producer
        DefaultMQProducer producer = new DefaultMQProducer("group_test");

        // 設(shè)置NameServer的地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        //producer.setSendLatencyFaultEnable(true);

        // 啟動Producer實例
        producer.start();


        for (int i = 0; i < 10; i++) {
            // 創(chuàng)建消息,并指定Topic,Tag和消息體
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            // 發(fā)送消息到一個Broker
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        //如果不再發(fā)送消息,關(guān)閉Producer實例。
        producer.shutdown();
    }
}

總結(jié)

這篇主要是帶大家了解RocketMQ的基本原理和介紹,在后面的章節(jié)中,會帶大家深入了解和使用RocketMQ,如果覺得文章有幫助的,記得點贊關(guān)注,您的支持是我創(chuàng)作的最大動力。


作者:牧小農(nóng)


公眾號:牧小農(nóng),微信掃碼關(guān)注或搜索公眾號名稱