一、安装
1.1 确定环境
确定java 与maven 环境
官网：http://rocketmq.apache.org/dowloading/releases/
Github：https://github.com/apache/rocketmq/
下载并解压后。配置环境变量
配置环境变量
变量名：ROCKETMQ_HOME
变量值：MQ 解压路径\MQ 文件夹名
例如：
启动
在 RocketMQ 的架构中，都是需要先启动 NameServer 再启动 Broker 的。所以先启动
NameServer。
启动 NameServer ，一定到bin目录下
使用 cmd 命令框执行进入至‘MQ 文件夹\bin’下，然后执行‘start mqnamesrv.cmd’，启动
NameServer。成功后会弹出提示框，此框勿关闭。
启动 Broker
使用 cmd 命令框执行进入至‘MQ 文件夹\bin’下，然后执行‘start mqbroker.cmd -n
127.0.0.1:9876 autoCreateTopicEnable=true’，启动 Broker。成功后会弹出提示框，此框勿
关闭。
若弹出提示框‘错误: 找不到或无法加载主类 xxxxxx’的处理
打开‘MQ 文件夹\bin’下的 runbroker.cmd，然后将‘%CLASSPATH%’加上英文双引号。保存
并重新执行 start 语句。
若内存不足：
修改rocketmq的存储路径
rocketmq-consolo控制台插件
环境要求
运行前确保：已经有 jdk1.8，Maven(打包需要安装 Maven 3.2.x)
下载
老版本地址下载：https://codeload.github.com/apache/rocketmq-externals/zip/master
新版本地址：https://github.com/apache/rocketmq-dashboard
这里本人安装的为新版本：
下载完成之后，进入‘\rocketmq-console\src\main\resources’文件夹，打开后可修改端口号和对应rocketmq服务地址；
maven编译并启动spring boot
进入D:\rocketmq-dashboard-master 目录，执行 mvn spring-boot:run，正常则出现spring，若error，github443【win10下 Failed to connect to github.com port 443: Timed out】，则配置hosts；
20.205.243.166 github.com
104.244.43.57 github.global.ssl.fastly.net
之后重新执行 mvn spring-boot:run 则启动正常
打开http://localhost:8089/ 则可看到rocketmq-consolo
【看自己配置的端口号】
简单使用：
文档介绍：
控制台使用文档
运维页面
你可以修改这个服务使用的 namesrv 的地址
你可以修改这个服务是否使用 VIPChannel(如果你的 mq server 版本小于 3.5.8，请设置不使用)
驾驶舱
查看 broker 的消息量（总量/5 分钟图）
查看单一主题的消息量（总量/趋势图）
集群
查看集群的分布情况：cluster 与 broker 关系、broker
查看 broker 具体信息/运行信息
查看 broker 配置信息
主题页面
展示所有的主题，可以通过搜索框进行过滤
筛选 普通/重试/死信 主题
添加/更新主题
clusterName 创建在哪几个 cluster 上
brokerName 创建在哪几个 broker 上
topicName 主题名
writeQueueNums 写队列数量
readQueueNums 读队列数量
perm //2 是写 4 是读 6 是读写
状态 查询消息投递状态（投递到哪些 broker/哪些 queue/多少量等）
路由 查看消息的路由（现在你发这个主题的消息会发往哪些 broker，对应 broker 的queue 信息）
CONSUMER 管理（这个 topic 都被哪些 group 消费了，消费情况何如）
topic 配置（查看变更当前的配置）
发送消息（向这个主题发送一个测试消息）
重置消费位点(分为在线和不在线两种情况，不过都需要检查重置是否成功)
删除主题 （会删除掉所有 broker 以及 namesrv 上的主题配置和路由信息）
消费者页面
展示所有的消费组，可以通过搜索框进行过滤
刷新页面/每隔五秒定时刷新页面
按照订阅组/数量/TPS/延迟 进行排序
添加/更新消费组
clusterName 创建在哪几个集群上
brokerName 创建在哪几个 broker 上
groupName 消费组名字
consumeEnable //是否可以消费 FALSE 的话将无法进行消费
consumeBroadcastEnable //是否可以广播消费
retryQueueNums //重试队列的大小
brokerId //正常情况从哪消费
whichBrokerWhenConsumeSlowly//出问题了从哪消费
终端 在线的消费客户端查看，包括版本订阅信息和消费模式
消费详情 对应消费组的消费明细查看，这个消费组订阅的所有 Topic 的消费情况，每个queue 对应的消费 client 查看（包括 Retry 消息）
配置 查看变更消费组的配置
删除 在指定的 broker 上删除消费组
生产者页面
通过 Topic 和 Group 查询在线的消息生产者客户端
信息包含客户端主机 版本
消息查询页面
根据 Topic 和时间区间查询 *由于数据量大 最多只会展示 2000 条，多的会被忽略
根据 Topic 和 Key 进行查询
最多只会展示 64 条
根据消息主题和消息 Id 进行消息的查询
消息详情可以展示这条消息的详细信息，查看消息对应到具体消费组的消费情况（如果异常，可以查看具体的异常信息）。可以向指定的消费组重发消息
可使用官方简单样例来验证：https://github.com/apache/rocketmq/blob/master/docs/cn/Example_Simple_cn.md。
消息发送，生产者
1 @Test
2 public void sandMessageTest() throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
3
4 // 实例化消息生产者Producer
5 DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
6 // 设置NameServer的地址
7 producer.setNamesrvAddr("localhost:9876");
8 // 启动Producer实例
9 producer.start();
10 for (int i = 0; i < 10; i++) {
11 // 创建消息，并指定Topic，Tag和消息体
12 Message msg = new Message("TopicTest" , "TagA" ,""+i,
13 ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
14 );
15 // 发送消息到一个Broker
16 SendResult sendResult = producer.send(msg);
17 // 通过sendResult返回消息是否成功送达
18 System.out.printf("%s%n", sendResult);
19 }
20 // 如果不再发送消息，关闭Producer实例。
21 producer.shutdown();
22 }如上图，
1)先启动producer
2)同步发送10条消息，topic=‘TopicTest’, tag=“TagA”,keys=“”+i, 消息体=Hello RocketMQ +i。
3)关闭生产者。
如上图所示，topic、tag、key、消息体都是正确的。
消费；
1 @Test
2 public void consumerMessageTest() throws MQClientException, InterruptedException {
3 // 实例化消费者
4 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
5
6 // 设置NameServer的地址
7 consumer.setNamesrvAddr("localhost:9876");
8
9 // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
10 consumer.subscribe("TopicTest", "*");
11 // 注册回调实现类来处理从broker拉取回来的消息
12 consumer.registerMessageListener(new MessageListenerConcurrently() {
13 @Override
14 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
15 System.out.printf("%s Receive New Messages: %s ,body= %s %n ", Thread.currentThread().getName(), msgs,new String(msgs.get(0).getBody()));
16 // 标记该消息已经被成功消费
17 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
18 }
19 });
20 // 启动消费者实例
21 consumer.start();
22 System.out.printf("Consumer Started.%n");
23 // 史前巨坑，这里官方样例不带这个，加上才能来得及消费。
24 Thread.sleep(10000L);
25 }如上图，
1)构造了一个消费者实例，并设置了nameserver、订阅条件(topic+tag表达式)。
- 注册了一个并发消息监听器，定义如何消费消息。
3)启动消费者，执行消费。这里最后一行Thread.sleep(10000L);要记得加上，不加还没消费线程就被关闭了…
执行后，消费日志如下：
Consumer Started.
ConsumeMessageThread_1 Receive New Messages: [MessageExt [queueId=8, storeSize=208, queueOffset=0, sysFlag=0, bornTimestamp=1631790768772, bornHost=/172.19.60.44:59863, storeTimestamp=1631790768795, storeHost=/172.19.60.44:10911, msgId=AC133C2C00002A9F0000000000000000, commitLogOffset=0, bodyCRC=613185359, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, KEYS=0, CONSUME_START_TIME=1631839990074, UNIQ_KEY=AC133C2C0E0018B4AAC2515ED2840000, CLUSTER=DefaultCluster, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 48], transactionId='null'}]]
ConsumeMessageThread_3 Receive New Messages: [MessageExt [queueId=14, storeSize=208, queueOffset=0, sysFlag=0, bornTimestamp=1631790768819, bornHost=/172.19.60.44:59863, storeTimestamp=1631790768819, storeHost=/172.19.60.44:10911, msgId=AC133C2C00002A9F00000000000004E0, commitLogOffset=1248, bodyCRC=1307562618, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, KEYS=6, CONSUME_START_TIME=1631839990074, UNIQ_KEY=AC133C2C0E0018B4AAC2515ED2B30006, CLUSTER=DefaultCluster, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 54], transactionId='null'}]]
ConsumeMessageThread_5 Receive New Messages: [MessageExt [queueId=1, storeSize=208, queueOffset=0, sysFlag=0, bornTimestamp=1631790768823, bornHost=/172.19.60.44:59863, storeTimestamp=1631790768824, storeHost=/172.19.60.44:10911, msgId=AC133C2C00002A9F0000000000000750, commitLogOffset=1872, bodyCRC=1565577195, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, KEYS=9, CONSUME_START_TIME=1631839990074, UNIQ_KEY=AC133C2C0E0018B4AAC2515ED2B70009, CLUSTER=DefaultCluster, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, ...省略其它