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

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

1 安裝 RocketMQ—Windows 版本

（1）下載 Windows 安裝包

Windows 版本下載地址：rocketmq.apache.org/release_not…

下載並解壓 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）下載插件

打開鏈接 github.com/apache/rock… 下載可視化插件 rocketmq-externals，以下圖所示：

點擊 Download ZIP 進行下載。

我爲你們準備了國內百度雲的下載連接，方便你們使用。 百度連接：pan.baidu.com/s/1sMO6W-56…   提取碼：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, 老王