RocketMQ 可視化環境搭建和基礎代碼使用

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

1 安裝 RocketMQ—Windows 版本

（1）下載 Windows 安裝包

Windows 版本下載地址：http://rocketmq.apache.org/release_notes/

下載並解壓 rocketmq 安裝包。

（2）配置系統環境變量

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

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

（3）啟動 namesrv

進入 rocketmq 的 bin 目錄，執行 start mqnamesrv.cmd ，執行成功如下圖所示：

注意：啟動之后，不能關閉此窗口。

（4）啟動 broker

還是在 bin 目錄下執行 start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true ，執行成功如下圖所示：

同樣不要關閉以上運行窗口。
完成以下步驟，說明你的 RocketMQ 已經按照成功了。

2 安裝可視化插件

（1）下載插件

打開連接 https://github.com/apache/rocketmq-externals.git 下載可視化插件 rocketmq-externals，如下圖所示：

點擊 Download ZIP 進行下載。

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

（2）配置插件

下載完成之后，進入 rocketmq-externals\rocketmq-console\src\main\resources\application.properties 進行配置，如下圖所示：

其中主要的字段說明如下：

• server.port=8066：此可視化插件的運行端口。
• rocketmq.config.namesrvAddr=127.0.0.1:9876：rocketmq 的鏈接信息。
  

（3）編譯插件

進入 rocketmq-externals\rocketmq-console 文件夾，執行 mvn clean package -Dmaven.test.skip=true
 編譯項目。
編譯成功如下圖所示：

編譯階段有可能出現以下兩個問題，沒有找到 mvn 命令，或編譯超級慢的問題，以下提供解決方案。

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

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

問題二：編譯超慢的問題

解決方案：這是因為使用 Maven 數據源為國外源的問題導致的，只需要配置阿里的 Maven 源即可。
打開 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>

（4）運行插件

編譯成功之后，進入 target 文件夾，執行 java -jar rocketmq-console-ng-1.0.1.jar 啟動程序。
啟動成功之后，在瀏覽器輸入地址 http://127.0.0.1:8066 進行訪問，效果如下圖：

3 基礎使用

（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>

（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, 老王