RocketMQ保姆級教程

来源:https://www.cnblogs.com/zzyang/archive/2022/08/16/16591213.html
-Advertisement-
Play Games

大家好，我是三友~~ 上周花了一點時間從頭到尾、從無到有地搭建了一套RocketMQ的環境，覺得還挺easy的，所以就寫篇文章分享給大家。 整篇文章可以大致分為三個部分，第一部分屬於一些核心概念和工作流程的講解；第二部分就是純手動搭建了一套環境；第三部分是基於環境進行測試和集成到SpringBoot ...


大家好，我是三友~~

上周花了一點時間從頭到尾、從無到有地搭建了一套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的名稱啊，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包可以從這獲取

鏈接：
提取碼：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

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

然後在瀏覽器中輸入伺服器的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有個入門，有興趣的小伙伴可以手動搭起來，整個過程順利的話可能就十幾二十分鐘這樣子。

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

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

往期熱門文章推薦

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

RocketMQ的push消費方式實現的太聰明瞭

一網打盡非同步神器CompletableFuture

@Async註解的坑，小心

7000字+24張圖帶你徹底弄懂線程池

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


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • Java多線程開發系列之五：Springboot 中非同步請求方法的使用
    Springboot 中非同步線程的使用在過往的後臺開發中，我們往往使用java自帶的線程或線程池，來進行非同步的調用。這對於效果來說沒什麼，甚至可以讓開發人員對底層的狀況更清晰，但是對於代碼的易讀性和可維護性卻非常的差。開發人員在實際使用過程中，應該更多的將精力放置在業務代碼的書寫過程中，而不是系統代 ...
  • 對於Java迴圈中的For和For-each，哪個更快
    Which is Faster For Loop or For-each in Java 對於Java迴圈中的For和For-each，哪個更快 通過本文，您可以瞭解一些集合遍歷技巧。 Java遍歷集合有兩種方法。一個是最基本的for迴圈，另一個是jdk5引入的for each。通過這種方法，我們可 ...
  • 在使用amoeba連接資料庫時，報錯java.lang.Exception: poolName=slaves, no valid pools
    項目場景：Mysql 實現資料庫讀寫分離 搭建3台MySQL伺服器，完成主從複製，搭建一臺amoeba伺服器，完成MySQL的讀寫分離 問題描述： 問題1、 在服務搭建完畢後，利用客戶機連接amoeba伺服器登錄資料庫，無法查看資料庫內容客戶端報錯的數據代碼： mysql> show databas ...
  • MultipartFile上傳文件非同步處理時的java.io.FileNotFoundException
    參考：https://javajgs.com/archives/26157 一.背景 1-1 需求 前端上傳Word文檔，後端將接收到的Word文檔①上傳到文件伺服器②將Word轉為Pdf。 1-2 方案 因為Word轉Pdf的耗時較長，為了及時給到前端返回信息，在將文件上傳到文件伺服器後，非同步將W ...
  • python-map()函數基本用法
    最近經常遇到一個問題：輸入端在同一行輸入兩個整型數字，並用空格間隔，問如何方便快捷的將這兩個變數分別賦予給x1,x2？ 新手小白，由於不知道map（）函數的用法，便想要用僅有的知識去解決它： 1 list1=[int(i) for i in input().split()] 2 x1=list1[0 ...
  • 基本數據類型概述
    常用類 筆記目錄：(https://www.cnblogs.com/wenjie2000/p/16378441.html) 包裝類 包裝類的分類 針對八種基本數據類型相應的引用類型—包裝類 有了類的特點，就可以調用類中的方法。 | 基本數據類型 | 包裝類 | | | | | boolean | B ...
  • 光說不練假把式，一起Kafka業務實戰。
    7.1 順序性場景 7.1.1 場景概述 假設我們要傳輸一批訂單到另一個系統，那麼訂單對應狀態的演變是有順序性要求的。 已下單 → 已支付 → 已確認 不允許錯亂！ 7.1.2 順序級別 1）全局有序： 串列化。每條經過kafka的消息必須嚴格保障有序性。 這就要求kafka單通道，每個groupi ...
  • Dubbo源碼(八) - 負載均衡
    前言 本文基於Dubbo2.6.x版本，中文註釋版源碼已上傳github：xiaoguyu/dubbo 負載均衡，英文名稱為Load Balance，其含義就是指將負載（工作任務）進行平衡、分攤到多個操作單元上進行運行。 例如：在Dubbo中，同一個服務有多個服務提供者，每個服務提供者所在的機器性能 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類