RocketMQ 是一款分散式訊息中介軟體，最初是由阿里巴巴訊息中介軟體團隊研發並大規模應用於生產系統，滿足線上海量訊息堆積的需求， 在 2016 年底捐贈給 Apache 開源基金會成為孵化專案，經過不到一年時間正式成為了 Apache 頂級專案。<br />早期阿里曾經基於 ActiveMQ 研發訊息系統， 隨著業務訊息的規模增大，瓶頸逐漸顯現，後來也考慮過Kafka，但因為在低延遲和高可靠性方面沒有選擇，最後才自主研發了 RocketMQ， 各方面的效能都比目前已有的訊息佇列要好，RocketMQ 和 Kafka 在概念和原理上都非常相似，所以也經常被拿來對比；RocketMQ 預設採用長輪詢的拉模式， 單機支援千萬級別的訊息堆積，可以非常好的應用在海量訊息系統中。<br />本文分為三部分，如下圖所示：<br />

<a name="1xvvk"></a>

1 安裝 RocketMQ—Windows 版本

<a name="RdEk4"></a>

（1）下載 Windows 安裝包

Windows 版本下載地址：http://rocketmq.apache.org/release_notes/<br /><br />下載並解壓 rocketmq 安裝包。 <a name="LHt4n"></a>

（2）配置系統環境變數

配置系統變數 ROCKETMQ_HOME=“D:\soft\rocketmq-all-4.5.1-bin-release”，如下圖所示：<br />注意：每個人 rocketmq 存放目錄不一樣，我的在 D:\soft 下，使用者根據自己的環境配置相應的系統變數。<br />

因為接下來啟動 mqnamesrv.cmd 中使用到了環境變數 %ROCKETMQ_HOME%，所以這裡需要配置此係統變數。

<a name="2S7UG"></a>

（3）啟動 namesrv

進入 rocketmq 的 bin 目錄，執行 start mqnamesrv.cmd ，執行成功如下圖所示：<br /><br />注意：啟動之後，不能關閉此視窗。 <a name="28zct"></a>

（4）啟動 broker

還是在 bin 目錄下執行 start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

 ，執行成功如下圖所示：<br /><br />同樣不要關閉以上執行視窗。<br />完成以下步驟，說明你的 RocketMQ 已經按照成功了。 <a name="0ItMT"></a>

2 安裝視覺化外掛

<a name="5iVLy"></a>

（1）下載外掛

開啟連線 https://github.com/apache/rocketmq-externals.git 下載視覺化外掛 rocketmq-externals，如下圖所示：<br /><br />點選 Download ZIP 進行下載。

我為大家準備了國內百度雲的下載連結，方便大家使用。 百度連結：https://pan.baidu.com/s/1sMO6W-562IFJF1uUBQFXYg   提取碼：fuzy 

<a name="mNB2M"></a>

（2）配置外掛

下載完成之後，進入 rocketmq-externals\rocketmq-console\src\main\resources\application.properties 進行配置，如下圖所示：<br /><br />其中主要的欄位說明如下：

• server.port=8066：此視覺化外掛的執行埠。
• rocketmq.config.namesrvAddr=127.0.0.1:9876：rocketmq 的連結資訊。 <a name="lQyXw"></a>

（3）編譯外掛

進入 rocketmq-externals\rocketmq-console 資料夾，執行 mvn clean package -Dmaven.test.skip=true<br /> 編譯專案。<br />編譯成功如下圖所示：<br /><br />編譯階段有可能出現以下兩個問題，沒有找到 mvn 命令，或編譯超級慢的問題，以下提供解決方案。 <a name="umxKe"></a>

問題一：mvn 非可以執行的命令

解決方案：這是因為沒有安裝 Maven 或者沒有配置 Maven 的環境變數導致的，下載 Maven 安裝包，增加環境變數 MAVEN_HOME=maven安裝目錄  ，給 path 中新增 %MAVEN_HOME%\bin ，重新啟動命令列工具（CMD）重新執行命令。 <a name="dLAy7"></a>

問題二：編譯超慢的問題

解決方案：這是因為使用 Maven 資料來源為國外源的問題導致的，只需要配置阿里的 Maven 源即可。<br />開啟 Maven 目錄下的 conf/setting.xml 給 mirrors 節點下新增如下內容：

<mirror>
 <id>alimaven</id>
 <name>aliyun maven</name>
 <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
 <mirrorOf>central</mirrorOf>        
</mirror>

<a name="rypWH"></a>

（4）執行外掛

編譯成功之後，進入 target 資料夾，執行 java -jar rocketmq-console-ng-1.0.1.jar 啟動程式。<br />啟動成功之後，在瀏覽器輸入地址 http://127.0.0.1:8066 進行訪問，效果如下圖：<br /> <a name="8IIeb"></a>

3 基礎使用

<a name="EEp8i"></a>

（1）新增引用 jar 包

pom.xml 新增以下程式碼：

<!-- https://mvnrepository.com/artifact/com.alibaba.rocketmq/rocketmq-client -->
<dependency>
    <groupId>com.alibaba.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>3.6.2.Final</version>
</dependency>

<a name="gb2C2"></a>

（2）新增生產者和消費者程式碼

public class RocketMQDemo {
    static final String MQ_NAMESRVADDR = "localhost:9876";
    public static void main(String[] args) {
        // 分組名
        String groupName = "myGroup-1";
        // 主題名
        String topicName = "myTopic-1";
        // 標籤名
        String tagName = "myTag-1";
        new Thread(() -> {
            try {
                producer(groupName, topicName, tagName);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (RemotingException e) {
                e.printStackTrace();
            } catch (MQClientException e) {
                e.printStackTrace();
            } catch (MQBrokerException e) {
                e.printStackTrace();
            }
        }).start();
        new Thread(() -> {
            try {
                consumer(groupName, topicName, tagName);
            } catch (MQClientException e) {
                e.printStackTrace();
            }
        }).start();
    }

    /**
     * @Description 生產者
     * @Author wanglei
     * @Param [groupName 分組名, topicName 主題名, tagName 標籤名]
     **/
    public static void producer(String groupName, String topicName, String tagName) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer(groupName);
        producer.setNamesrvAddr(MQ_NAMESRVADDR);
        producer.start();
        String body = "Hello, 老王";
        Message message = new Message(topicName, tagName, body.getBytes());
        producer.send(message);
        producer.shutdown();
    }

    /**
     * @Description 消費者
     * @Author wanglei
     * @Param [groupName 分組名, topicName 主題名, tagName 標籤名]
     **/
    public static void consumer(String groupName, String topicName, String tagName) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(MQ_NAMESRVADDR);
        consumer.subscribe(topicName, tagName);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}

以上程式執行結果如下：

Hello, 老王

相關推薦