【RocketMQ】4、Windows10单机环境部署

• 【RocketMQ】4、Windows10单机环境部署
• 一、RocketMQ下载
• 二、环境搭建
• 三、启动RocketMQ
• 四、图形管理控制台Console
• • 1、下载
  • 2、更改配置
  • 3、Maven打包
  • 4、启动
• 四、测试
• • 1、按前面步骤启动NameServer、Broker、Console
  • 2、代码测试

官网下载地址：https://rocketmq.apache.org/dowloading/releases/

下载解压后文件夹截图

需提前配好Java环境

• 变量名：ROCKETMQ_HOME
  变量值：D:EnvsRocketMQrocketmq-all-4.9.4-bin-release
  

进入到bin目录下

• 启动NameServer
  双击mqnamesrv.cmd
  

• 启动Broker
  修改confbroker.conf配置文件，添加Name Server地址：namesrvAddr = 127.0.0.1:9876
  
  start binmqbroker.cmd -c confbroker.conf 启动Broker
  

踩坑：没有添加name server地址控制台启动Broker时，控制台没有Name Server地址显示

且会导致后面生产者发送消息失败，显示Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: No route info of this topic: TopicTest

Console项目在develop分支下
地址：https://github.com/apache/rocketmq-externals/tree/develop
Clone下来后，用idea打开

修改console项目中的application.properties文件

• Console端口号server.port=9999
• NameServer地址rocketmq.config.namesrvAddr=0.0.0.0:9876
• 数据存放路径rocketmq.config.dataPath=D:/Envs/RocketMQ/rocketmq-console
  

踩坑：rocketmq.config.dataPath中的路径分隔符为反斜杠“/”，否则后面执行包时会报错找不到文件夹。

进入console目录后用maven打包
命令：mvn clean package ‘-Dmaven.test.skip=true’

踩坑：提示Unknown lifecycle phase ".test.skip=true". You must specify a valid lifecycle phase or a goal in the format

原因： IntelliJ IDEA 的终端默认使用 PowerShell 来运行命令，而在 PowerShell 下，参数 -Dmaven.test.skip=true 没有被正确地识别。

解决方法：

• 在此参数外加单引号即可，即：mvn package '-Dmaven.test.skip=true'
• 也可以选择将 IntelliJ IDEA 的终端设置成 CMD 模式（Command Prompt）

踩坑：提示Could not find artifact org.apache.rocketmq:rocketmq-tools:jar:4.4.0-SNAPSHOT

解决方法：

• 将pom文件中的4.4.0-SNAPSHOT
  修改为 4.4.0

进入target文件夹，控制台命令java -jar .rocketmq-console-ng-1.0.0.jar。

浏览器输入刚刚设置的Console地址和端口127.0.0.1:9999，控制台启动成功。

测试代码使用官方的demo，链接：https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md

• Maven依赖

    org.apache.rocketmq
    rocketmq-client
    4.9.1

• 先运行生产者（同步消息）

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("pg1");
        // 设置NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动Producer实例
        producer.start();
        for (int i = 0; i < 100; i++) {
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message("TopicTest" ,
                    "TagA" ,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) 
            );
            // 发送消息到一个Broker
            SendResult sendResult = producer.send(msg);
            // 通过sendResult返回消息是否成功送达
            System.out.printf("%s%n", sendResult);
        }
        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();
    }
}

• 后运行消费者

public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg1");
        // 设置NameServer的地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("TopicTest", "*");
        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}