RocketMQ保姆級教程-有解無憂

大家好，我是三友~~

上周花了一點時間從頭到尾、從無到有地搭建了一套RocketMQ的環境，覺得還挺easy的，所以就寫篇文章分享給大家，

整篇文章可以大致分為三個部分，第一部分屬于一些核心概念和作業流程的講解；第二部分就是純手動搭建了一套環境；第三部分是基于環境進行測驗和集成到SpringBoot，因為整個程序講的比較細，所以我稱之為“保姆級教程”，

好了，廢話補多少，直接進入主題，

前言

RocketMQ是阿里巴巴旗下一款開源的MQ框架，經歷過雙十一考驗、Java編程語言實作，有非常好完整生態系統，RocketMQ作為一款純java、分布式、佇列模型的開源訊息中間件，支持事務訊息、順序訊息、批量訊息、定時訊息、訊息回溯等，總之就是葛大爺的一句話

核心概念

NameServer：可以理解為是一個注冊中心，主要是用來保存topic路由資訊，管理Broker，在NameServer的集群中，NameServer與NameServer之間是沒有任何通信的，
Broker：核心的一個角色，主要是用來保存topic的資訊，接受生產者產生的訊息，持久化訊息，在一個Broker集群中，相同的BrokerName可以稱為一個Broker組，一個Broker組中,BrokerId為0的為主節點，其它的為從節點，BrokerName和BrokerId是可以在Broker啟動時通過組態檔配置的，每個Broker組只存放一部分訊息，
生產者：生產訊息的一方就是生產者
生產者組：一個生產者組可以有很多生產者，只需要在創建生產者的時候指定生產者組，那么這個生產者就在那個生產者組
消費者：用來消費生產者訊息的一方
消費者組：跟生產者一樣，每個消費者都有所在的消費者組，一個消費者組可以有很多的消費者，不同的消費者組消費訊息是互不影響的，
topic（主題）：可以理解為一個訊息的集合的名字，生產者在發送訊息的時候需要指定發到哪個topic下，消費者消費訊息的時候也需要知道自己消費的是哪些topic底下的訊息，
Tag（子主題）：比topic低一級，可以用來區分同一topic下的不同業務型別的訊息，發送訊息的時候也需要指定，

這里有組的概念是因為可以用來做到不同的生產者組或者消費者組有不同的配置，這樣就可以使得生產者或者消費者更加靈活，

作業流程

說完核心概念，再來說一下核心的作業流程，這里我先畫了一張圖，

通過這張圖就可以很清楚的知道，RocketMQ大致的作業流程：

Broker啟動的時候，會往每臺NameServer（因為NameServer之間不通信，所以每臺都得注冊）注冊自己的資訊，這些資訊包括自己的ip和埠號，自己這臺Broker有哪些topic等資訊，
Producer在啟動之后會跟會NameServer建立連接，定期從NameServer中獲取Broker的資訊，當發送訊息的時候，會根據訊息需要發送到哪個topic去找對應的Broker地址，如果有的話，就向這臺Broker發送請求；沒有找到的話，就看根據是否允許自動創建topic來決定是否發送訊息，
Broker在接收到Producer的訊息之后，會將訊息存起來，持久化，如果有從節點的話，也會主動同步給從節點，實作資料的備份
Consumer啟動之后也會跟會NameServer建立連接，定期從NameServer中獲取Broker和對應topic的資訊，然后根據自己需要訂閱的topic資訊找到對應的Broker的地址，然后跟Broker建立連接，獲取訊息，進行消費

就跟上面的圖一樣，整體的作業流程還是比較簡單的，這里我簡化了很多概念，主要是為了好理解，

環境搭建

終于講完了一些簡單的概念，接下來就來搭建一套RocketMQ的環境，

通過上面分析，我們知道，在RocketMQ中有NameServer、Broker、生產者、消費者四種角色，而生產者和消費者實際上就是業務系統，所以這里不需要搭建，真正要搭建的就是NameServer和Broker，但是為了方便RocketMQ資料的可視化，這里我多搭建一套可視化的服務，

搭建程序比較簡單，按照步驟一步一步來就可以完成，如果提示一些命令不存在，那么直接通過yum安裝這些命令就行，

一、準備

需要準備一個linux服務器，需要先安裝好JDK

關閉防火墻

systemctl stop firewalld
systemctl disable firewalld

下載并解壓RocketMQ

1、創建一個目錄，用來存放rocketmq相關的東西

mkdir /usr/rocketmq
cd /usr/rocketmq

2、下載并解壓rocketmq

下載

wget https://archive.apache.org/dist/rocketmq/4.7.1/rocketmq-all-4.7.1-bin-release.zip

解壓

unzip rocketmq-all-4.7.1-bin-release.zip

看到這一個檔案夾就完成了

然后進入rocketmq-all-4.7.1-bin-release檔案夾

cd rocketmq-all-4.7.1-bin-release

RocketMQ的東西都在這了

二、搭建NameServer

修改jvm引數

在啟動NameServer之前，強烈建議修改一下啟動時的jvm引數，因為默認的引數都比較大，為了避免記憶體不夠，建議修改小，當然，如果你的記憶體足夠大，可以忽略，

vi bin/runserver.sh

修改畫圈的這一行

這里你可以直接修改成跟我一樣的

-server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=32m -XX:MaxMetaspaceSize=50m

啟動NameServer

修改完之后，執行如下命令就可以啟動NameServer了

nohup sh bin/mqnamesrv &

查看NameServer日志

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

如果看到如下的日志，就說明啟動成功了

NameServer日志

三、搭建Broker

這里啟動單機版的Broker

修改jvm引數

跟啟動NameServer一樣，也建議去修改jvm引數

vi bin/runbroker.sh

將畫圈的地方設定小點，當然也別太小啊

當然你還是可以跟我設定的一樣

-server -Xms1g -Xmx1g -Xmn512m

修改Broker組態檔broker.conf

這里需要改一下Broker組態檔，需要指定NameServer的地址，因為需要Broker需要往NameServer注冊

vi conf/broker.conf

Broker組態檔

這里就能看出Broker的配置了，什么Broker集群的名稱啊，Broker的名稱啊，Broker的id啊，都跟前面說的對上了，

在檔案末尾追加地址

namesrvAddr = localhost:9876

因為NameServer跟Broker在同一臺機器，所以是localhost，NameServer埠默認的是9876，

不過這里我還建議再修改一處資訊，因為Broker向NameServer進行注冊的時候，帶過去的ip如果不指定就會自動獲取，但是自動獲取的有個坑，就是有可能你的電腦無法訪問到這個自動獲取的ip，所以我建議手動指定你的電腦可以訪問到的服務器ip，

我的虛擬機的ip是192.168.200.143，所以就指定為192.168.200.143，如下

brokerIP1 = 192.168.200.143
brokerIP2 = 192.168.200.143

如果以上都配置的話，最終的組態檔應該如下，紅圈的為新加的

啟動Broker

nohup sh bin/mqbroker -c conf/broker.conf &

-c 引數就是指定組態檔

查看日志

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

當看到如下日志就說明啟動成功了

四、搭建可視化控制臺

其實前面NameServer和Broker搭建完成之后，就可以用來收發訊息了，但是為了更加直觀，可以搭一套可視化的服務，

可視化服務其實就是一個jar包，啟動就行了，

jar包可以從這獲取

鏈接：https://pan.baidu.com/s/16s1qwY2qzE2bxR81t5Wm6w
提取碼：s0sd

將jar包上傳到服務器，放到/usr/rocketmq的目錄底下，當然放哪都無所謂，這里只是為了方便，因為rocketmq的東西都在這里

然后進入/usr/rocketmq下，執行如下命名

nohup java -jar -server -Xms256m -Xmx256m -Drocketmq.config.namesrvAddr=localhost:9876 -Dserver.port=8088 rocketmq-console-ng-1.0.1.jar &

rocketmq.config.namesrvAddr就是用來指定NameServer的地址的

查看日志

tail -f ~/logs/consolelogs/rocketmq-console.log

當看到如下日志，就說明啟動成功了

然后在瀏覽器中輸入http://linux服務器的ip:8088/就可以看到控制臺了，如果無法訪問，可以看看防火墻有沒有關閉

右上角可以把語言切換成中文

Broker集群資訊

topic資訊

通過控制臺可以查看生產者、消費者、Broker集群等資訊，非常直觀，

功能很多，這里就不一一介紹了，

測驗

環境搭好之后，就可以進行測驗了，

引入依賴

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

生產者發送訊息

public class Producer {
    public static void main(String[] args) throws Exception {
        //創建一個生產者，指定生產者組為sanyouProducer
        DefaultMQProducer producer = new DefaultMQProducer("sanyouProducer");

        // 指定NameServer的地址
        producer.setNamesrvAddr("192.168.200.143:9876");
        // 第一次發送可能會超時，我設定的比較大
        producer.setSendMsgTimeout(60000);

        // 啟動生產者
        producer.start();

        // 創建一條訊息
        // topic為 sanyouTopic
        // 訊息內容為 三友的java日記
        // tags 為 TagA
        Message msg = new Message("sanyouTopic", "TagA", "三友的java日記 ".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 發送訊息并得到訊息的發送結果，然后列印
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);

        // 關閉生產者
        producer.shutdown();
    }

}

構建一個訊息生產者DefaultMQProducer實體，然后指定生產者組為sanyouProducer；
指定NameServer的地址：服務器的ip:9876，因為需要從NameServer拉取Broker的資訊
producer.start() 啟動生產者
構建一個內容為三友的java日記的訊息，然后指定這個訊息往sanyouTopic這個topic發送
producer.send(msg)：發送訊息，列印結果
關閉生產者

運行結果如下

SendResult [sendStatus=SEND_OK, msgId=C0A81FAF54F818B4AAC2475FD2010000, offsetMsgId=C0A8C88F00002A9F000000000009AE55, messageQueue=MessageQueue [topic=sanyouTopic, brokerName=broker-a, queueId=0], queueOffset=0]

sendStatus=SEND_OK 說明發送成功了，此時就能后控制臺看到未消費的訊息了，

到控制臺看到訊息那塊，然后選定發送的topic，查詢的時間范圍手動再選一下，不選就查不出來(我懷疑這是個bug)，然后查詢就能看到了一條訊息，

然后點擊一下MESSAGE DETAIL就能夠看到詳情，

這里就能看到發送訊息的詳細資訊，

左下角訊息的消費的消費，因為我們還沒有消費者訂閱這個topic，所以左下角沒資料，

消費者消費訊息

public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {

        // 通過push模式消費訊息，指定消費者組
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sanyouConsumer");

        // 指定NameServer的地址
        consumer.setNamesrvAddr("192.168.200.143:9876");

        // 訂閱這個topic下的所有的訊息
        consumer.subscribe("sanyouTopic", "*");

        // 注冊一個消費的監聽器，當有訊息的時候，會回呼這個監聽器來消費訊息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("消費訊息:%s", new String(msg.getBody()) + "\n");
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 啟動消費者
        consumer.start();

        System.out.printf("Consumer Started.%n");
    }
}

創建一個消費者實體物件，指定消費者組為sanyouConsumer
指定NameServer的地址：服務器的ip:9876
訂閱 sanyouTopic 這個topic的所有資訊
consumer.registerMessageListener ，這個很重要，是注冊一個監聽器，這個監聽器是當有訊息的時候就會回呼這個監聽器，處理訊息，所以需要用戶實作這個介面，然后處理訊息，
啟動消費者

啟動之后，消費者就會消費剛才生產者發送的訊息，于是控制臺就列印出如下資訊

Consumer Started.
消費訊息:三友的java日記

此時再去看控制臺

發現被sanyouConsumer這個消費者組給消費了，

SpringBoot環境下集成RocketMQ

集成

在實際專案中肯定不會像上面測驗那樣用，都是集成SpringBoot的，

1、引入依賴

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.1</version>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <version>2.1.1.RELEASE</version>
</dependency>

2、yml配置

rocketmq:
  producer:
    group: sanyouProducer
  name-server: 192.168.200.143:9876

3、創建消費者

SpringBoot底下只需要實作RocketMQListener介面，然后加上@RocketMQMessageListener注解即可

@Component
@RocketMQMessageListener(consumerGroup = "sanyouConsumer", topic = "sanyouTopic")
public class SanYouTopicListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String msg) {
        System.out.println("處理訊息:" + msg);
    }

}

@RocketMQMessageListener需要指定消費者屬于哪個消費者組，消費哪個topic，NameServer的地址已經通過yml組態檔配置類

4、測驗

@SpringBootTest(classes = RocketMQApplication.class)
@RunWith(SpringRunner.class)
public class RocketMQTest {

    @Autowired
    private RocketMQTemplate template;

    @Test
    public void send() throws InterruptedException {
        template.convertAndSend("sanyouTopic", "三友的java日記");
        Thread.sleep(60000);
    }

}

直接注入一個RocketMQTemplate，然后通過RocketMQTemplate發送訊息，

運行結果如下：

處理訊息:三友的java日記

的確消費到訊息了，

原理

其實原理是一樣的，只不過在SpringBoot中給封裝了一層，讓使用起來更加簡單，

1、RocketMQTemplate構造代碼

所以從這可以看出，最終在構造RocketMQTemplate的時候，傳入了一個DefaultMQProducer，所以可想而知，最終RocketMQTemplate發送訊息也是通過DefaultMQProducer發送的，

2、@RocketMQMessageListener 注解處理

從這可以看出，會為每一個加了@RocketMQMessageListener注解的物件創建一個DefaultMQPushConsumer，所以最終也是通過DefaultMQPushConsumer消費訊息的，

至于監聽器，是在這

遍歷每條訊息，然后呼叫handleMessage，最侄訓呼叫實作了RocketMQListener的物件處理訊息，

最后

通過上面的理論介紹和實際的環境搭建再加上代碼的測驗，相信應該可以對RocketMQ有個入門，有興趣的小伙伴可以手動搭起來，整個程序順利的話可能就十幾二十分鐘這樣子，

最后我再說一句，從文章整體也可以看出本文沒有涉及太深入的一些機制和原理的講解，比如訊息是如何存盤的，事務和延遲訊息是如何實作的，主從是如何同步的等等，甚至壓根就沒提到佇列這個詞，主要是因為我打算后面再寫一篇文章，來單獨剖析這些機制和原理，

最后的最后，本文所有的代碼地址:

https://github.com/sanyou3/rocketmq-demo.git

往期熱門文章推薦

三萬字盤點Spring/Boot的那些常用擴展點

RocketMQ的push消費方式實作的太聰明了

一網打盡異步神器CompletableFuture

@Async注解的坑，小心

7000字+24張圖帶你徹底弄懂執行緒池

掃碼或者搜索關注公眾號三友的java日記，及時干貨不錯過，公眾號致力于通過畫圖加上通俗易懂的語言講解技術，讓技術更加容易學習，回復面試即可獲得一套精美的面試真題，

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/501975.html

標籤：Java

上一篇：Java注解最全詳解(超級詳細)

下一篇：Dubbo原始碼(八) - 負載均衡