RocketMQ快速入门

简介

Apache RocketMQ 是阿里开源的一款高性能、高吞吐量的分布式消息中间件，在由阿里捐赠给Apache软件基金会之后孵化成了Apache的一个顶级项目（Top-Level Project，TLP）。RocketMQ 是基于阿里闭源的 MetaQ 内核实现的，它服务于阿里的各个体系，尤其是在电商领域提供了极高的并发支撑。RocketMQ 是使用 Java 语言开发的，因此相较于业界流行的其它消息中间件比如 RabbitMQ（使用 Erlang 开发）、Kafka（使用 Scala 开发），它对于 Java 程序员来说更加友好，更加利于开发者在其底层的基础上进行封装和扩展。
RocketMQ 的官网是http://rocketmq.apache.org/，根据官网的介绍，Apache RocketMQ™  是一个标准化的消息引擎，轻量级的数据处理平台。它具备以下优势：

• 低延迟：在高并发压力下，超过99.6%的响应延迟在1毫秒之内
• 面向金融的：系统支撑的追踪和审计功能具备高可用性
• 行业可发展性：保证了万亿级别的消息容量
• 厂商中立的：在最新的4.1版本开放了一个新的分布式消息和流标准
• 对大数据友好的：具备通用集成功能的批量传输支撑了海量吞吐
• 大量的积累：只要给与足够的磁盘空间，就可以累积消息而不会造成性能损失

RocketMQ 诞生背景

在早期，阿里曾基于 ActiveMQ 5.x（低于5.3）构建了分布式消息中间件。 并将其用于他们的跨国公司的异步通信，搜索，社交网络活动流，数据管道，甚至交易过程中。 随着贸易业务吞吐量的增长，来自消息集群的压力也就变成了亟待解决的问题。

为什么设计 RocketMQ？

随着队列的增长和虚拟主题的使用，ActiveMQ IO 模块遇到了瓶颈。 尽管阿里尽力通过限流，熔断或降级来解决此问题，但效果不佳。 因此，那时他们开始关注流行的消息传递解决方案Kafka。 不幸的是，Kafka不能满足阿里的要求，特别是在低延迟和高可靠性方面，请参阅此处以了解详细信息。
在这种情况下，他们决定发明一个新的消息传递引擎来处理更广泛的用例集——从传统的发布/订阅方案到大批量实时零损失容忍的交易系统。

RocketMQ vs. ActiveMQ vs. Kafka

下表展示了RocketMQ，ActiveMQ和Kafka（依照awesome-java 来说，Apache RocketMQ是最流行的消息传递解决方案）之间的比较：

Quick Start

本部分内容将会介绍如何在本地的机器上快速安装一个 RocketMQ 用于收发消息，更多的细节可以访问 https://github.com/apache/rocketmq/tree/master/docs/cn进行查看。

前置要求

在安装 RokcetMQ 之前需要先安装如下软件：

1. 64位操作系统，推荐使用 Linux/Unix/Mac 操作系统
2. 64位 JDK 1.8 及以上的版本
3. Maven 3.2.x
4. Git
5. 为 Broker server 至少预留 4G 的磁盘空间

下载、安装及使用

Linux

1. 安装jdk

1. 首先在 oracle 官网下载 Linux jdk 8 的压缩包

2. 然后使用 ftp 工具将下载好的压缩包上传到 Linux 服务器上
3. 使用 tar -zxvf jdk-8u261-linux-x64.tar.gz 命令解压文件
4. 配置系统环境变量

[root@node01 opt]# vi /etc/profile

在文件末尾添加以下内容：

# jdk放置目录JAVA_HOME=/opt/jdk1.8.0_226
# jre放置目录JRE_HOME=/opt/jdk1.8.0_226/jre
# 配置 path 环境变量，以 : 分隔PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin
# 配置 classpath 环境变量，以 : 分隔CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib
# 设置环境变量export JAVA_HOME JRE_HOME PATH CLASSPATH

5. 使配置文件的变更生效

[root@node01 opt]# source /etc/profile

6. 查看 jdk 版本，至此，jdk 就已经安装完毕

[root@node01 opt]# java -versionjava version "1.8.0_226"Java(TM) SE Runtime Environment (build 1.8.0_226-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.226-b11, mixed mode)

2. 安装maven

1. 使用 wget 下载 maven

[root@node01 opt]# wget https://mirrors.bfsu.edu.cn/apache/maven/maven-3/3.6.3/binaries/apache-maven-3.6.3-bin.tar.gz

2. 解压

[root@node01 opt]# tar -zxvf apache-maven-3.6.3-bin.tar.gz

3. 添加阿里云镜像

[root@node01 opt]# cd apache-maven-3.6.3/conf/[root@node01 conf]# ll总用量 20drwxr-xr-x. 2 root root  409611月  72019 logging
-rw-r--r--. 1 root root 1046811月  72019 settings.xml
-rw-r--r--. 1 root root  374711月  72019 toolchains.xml
[root@node01 conf]# vi settings.xml

在配置文件 settings.xml 中加入如下代码：

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

4. 配置环境变量

[root@node01 conf]# vi /etc/profile

在文件中添加如下配置：

# maven 安装目录M2_HOME=/opt/apache-maven-3.6.3
PATH=$PATH:$M2_HOME/bin
export M2_HOME

[root@node01 opt]# source /etc/profile

5. 查看 maven 版本，至此，maven 就安装完毕了

[root@node01 conf]# mvn -vApache Maven 3.6.3 (cecedd343002696d0abb50b32b541b8a6ba2883f)
Maven home: /opt/apache-maven-3.6.3
Java version: 1.8.0_226, vendor: Oracle Corporation, runtime: /opt/jdk1.8.0_226/jre
Default locale: zh_CN, platform encoding: UTF-8
OS name: "linux", version: "3.10.0-1127.el7.x86_64", arch: "amd64", family: "unix"

3. 安装rocketmq

1. 下载 rocketmq 源码包

[root@node01 opt]# wget https://mirror.bit.edu.cn/apache/rocketmq/4.7.1/rocketmq-all-4.7.1-source-release.zip

2. 解压

[root@node01 opt]# yum install -y unzip[root@node01 opt]# unzip rocketmq-all-4.7.1-source-release.zip

3. 去到解压后的文件目录中进行编译

[root@node01 rocketmq-all-4.7.1-source-release]# mvn -Prelease-all -DskipTests clean install -U

4. 将编译好的文件移个位置

[root@node01 rocketmq-all-4.7.1-source-release]# cd distribution/target/rocketmq-4.7.1[root@node01 rocketmq-4.7.1]# ll总用量 4drwxr-xr-x. 6 root root 40968月  2915:42 rocketmq-4.7.1
[root@node01 rocketmq-4.7.1]# mv rocketmq-4.7.1/ /opt

5. 启动 Name Server，如果在终端显示如下信息，则说明启动成功了：

[root@node01 opt]# cd rocketmq-4.7.1/bin/[root@node01 bin]# lscachedog.sh       dledger      mqbroker            mqbroker.numanode1  mqnamesrv      mqshutdown.cmd  play.sh        runbroker.sh   setcache.sh   tools.sh
cleancache.sh     mqadmin      mqbroker.cmd        mqbroker.numanode2  mqnamesrv.cmd  os.sh           README.md      runserver.cmd  startfsrv.sh
cleancache.v1.sh  mqadmin.cmd  mqbroker.numanode0  mqbroker.numanode3  mqshutdown     play.cmd        runbroker.cmd  runserver.sh   tools.cmd
[root@node01 bin]# ./mqnamesrvJava HotSpot(TM) 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release
Java HotSpot(TM) 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release.
The Name Server boot success. serializeType=JSON

6. 启动 broker
   使用命令启动 broker，会发现报如下的错误：

[root@node01 bin]# ./mqbroker -n localhost:9876Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000005c0000000, 8589934592, 0) failed; error='Cannot allocate memory'(errno=12)
## There is insufficient memory for the Java Runtime Environment to continue.# Native memory allocation (mmap) failed to map 8589934592 bytes for committing reserved memory.# An error report file with more information is saved as:# /opt/rocketmq-4.7.1/bin/hs_err_pid9514.log

从错误信息我们可以得知原因是 jvm 启动初始化内存分配大于物理内存。
因此，我们可以修改启动脚本中的 jvm 参数来解决这个问题，首先，让我们来修改nameserver的启动脚本：

[root@node01 bin]# vi runserver.sh

找到分配jvm内存的配置行：

JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

将它修改为如下配置（具体配置数值根据实际生产场景进行调整）：

JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

然后修改broker的启动脚本：

[root@node01 bin]# vi runbroker.sh

同样是找到分配jvm内存的配置行：

JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"

将它修改为如下配置（具体配置数值根据实际生产场景进行调整）：

JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m"

修改完之后重启 nameserver 和 broker，当出现如下信息，说明 broker 已经启动成功：

[root@node01 bin]# ./mqbroker -n localhost:9876The broker[node01, 192.168.114.60:10911] boot success. serializeType=JSON and name server is localhost:9876

7. 执行测试程序测试消息发送和接收
   在发送/接收消息之前，我们需要告诉客户端 Name Server 的位置。RocketMQ提供了多种方法来实现这一目标。为了简单起见，我们通过设置环境变量 NAMESRV_ADDR 来实现，在 /etc/profile 中加入如下配置：

exportNAMESRV_ADDR=localhost:9876

使用测试程序发送消息：

./tools.sh org.apache.rocketmq.example.quickstart.Producer

显示如下信息则表示消息发送成功：

另起一个会话，然后用测试程序来接收消息：

./tools.sh org.apache.rocketmq.example.quickstart.Consumer

显示如下信息则表示消息接收成功：

8. 关闭

关闭 broker：

[root@node01 bin]# ./mqshutdown brokerThe mqbroker(1502) is running...
Send shutdown request to mqbroker(1502) OK

关闭 name server：

[root@node01 bin]# ./mqshutdown namesrvThe mqnamesrv(1444) is running...
Send shutdown request to mqnamesrv(1444) OK

Windows

本文主要讲述的是 windows 10 操作系统下 RocketMQ 的安装，请确保你的操作系统中已经安装了 PowerShell。和 Linux 一样，在 Windows 安装 RocketMQ 需要先安装 jdk 和 maven ，本文就不再细述如何安装 jdk 和 maven 了，请自行查阅资料安装。

1. 安装 RocketMQ

1. 首先，在官网下载 RocketMQ 的二进制压缩包，然后选择一个本地目录进行解压缩
2. 配置环境变量，添加如下两个环境变量：

打开 runbroker.cmd，修改如下配置行：

set"JAVA_OPT=%JAVA_OPT% -cp %CLASSPATH%"

修改后的配置行如下：

rem set"JAVA_OPT=%JAVA_OPT% -cp "%CLASSPATH%"

3. 启动 name server：

4. 启动 broker：

当控制台显示如下信息时，则说明启动成功了：

5. 使用测试程序发送/接收消息

发送消息：

接收消息：

6. 关闭
   直接关闭 cmd/powershell 即可（不要在生产环境这样做）

安装启动时可能会遇到的问题

1. 编译时包无法在mirror上找到 提示502错误

原因：网络不好或maven仓库服务器出错
重试即可，或者换一个镜像仓库

2. 启动broker失败，报 Cannot allocate memory 错误

此问题在上文中已经叙述过了，在此就不再赘述了

3. 启动broker成功但提示：Failed to obtain the host name

原因：无法解析当前的主机名
在/etc/hosts里添加映射即可

# 配置 ip 地址到主机名的映射192.168.114.60 node-01

4. linux日期校准

安装ntpdate

yum install ntpdate
ntpdate ntp1.aliyun.com

控制台rocketmq-console

编译安装

1. 下载

编译源码包下载地址：https://github.com/apache/rocketmq-externals
中文指南
https://github.com/apache/rocketmq-externals/blob/master/rocketmq-console/doc/1_0_0/UserGuide_CN.md

2. 将源码包上传到服务器并解压缩

[root@node01 opt]# unzip rocketmq-externals.zip

3. 编译

[root@node01 opt]# cd rocketmq-externals/rocketmq-console/[root@node01 rocketmq-console]# mvn clean package -Dmaven.test.skip=true

4. 启动

编译成功后在rocketmq-console/target目录下执行rocketmq-console-ng-2.0.0.jar
启动（需要确保你的 name server 已经启动了），直接动态添加nameserver地址即可，或者你也可以通过编辑rocketmq-console/src/main/resources目录下的application.properties配置文件添加rocketmq.config.namesrvAddr属性来配置nameserver地址。

[root@node01 rocketmq-console]# ll总用量 68drwxr-xr-x. 3 root root  40968月  3010:50 doc
-rw-r--r--. 1 root root 304228月  3010:50 LICENSE
-rw-r--r--. 1 root root   1808月  3010:50 NOTICE
-rw-r--r--. 1 root root 105938月  3010:50 pom.xml
-rw-r--r--. 1 root root  23908月  3010:50 README.md
drwxr-xr-x. 4 root root  40968月  3010:50 src
drwxr-xr-x. 3 root root  40968月  3010:50 style
drwxr-xr-x. 7 root root  40968月  3010:59 target
[root@node01 rocketmq-console]# cd target/[root@node01 target]# java -jar rocketmq-console-ng-2.0.0.jar --rocketmq.config.namesrvAddr=localhost:9876

当出现如下界面时就说明启动成功了：

启动成功后访问服务器 8080 端口即可。

界面组成

RocketMQ 是面向集群而生的，这一点从 rocketmq-console 的界面上就能体现出来。

运维

• NameSvrAddrList：配置 name server 的地址，它是个列表，因此可以配置多个 name server 的地址，这说明 name server 是可以集群部署的
• IsUseVIPChannel：是否使用VIPChannel

驾驶舱

• Broker TOP 10：查看消息量最多的10个broker的消息量（总量）
• Broker 5min trend：查看broker消息量5分钟的趋势
• 主题 TOP 10：查看消息量最多的10个单一主题的消息量（总量）
• 主题 5min trend：查看主题消息量5分钟的趋势

集群

• 查看集群的分布情况

• cluster与broker关系
• cluster中包含的broker

• 查看broker具体信息/运行信息/状态信息
• 查看broker配置信息

主题

• 展示所有的主题，可以通过主题名称进行过滤
• 筛选 普通/重试/死信/系统 主题
• 新增/更新主题

• clusterName：集群名
• brokerName：主机名
• topicName：主题名
• writeQueueNums：写队列数量
• readQueueNums：读队列数量
• perm：2是写 4是读 6是读写

• 状态 查询消息投递状态（投递到哪些broker/哪些queue/多少量等）
• 路由 查看消息的路由（现在你发这个主题的消息会发往哪些broker，对应broker的queue信息）
• CONSUMER管理（这个topic都被哪些group订阅了、消费了，消费情况何如）
• topic配置（查看/变更当前的topic的配置）
• 发送消息（向这个主题发送一个测试消息）
• 重置消费位点(分为在线和离线两种情况，不过都需要检查重置是否成功)
• 删除主题 （会删除掉所有broker以及namesrv上的主题配置和路由信息，在生产环境上请慎重进行这项操作）

消费者

• 展示所有的消费组，可以通过组名进行过滤
• 刷新页面/每隔五秒定时刷新页面
• 按照订阅组/数量/TPS/延迟 进行排序

• 新增/更新消费组
• clusterName ：集群名
• brokerName：主机名
• groupName：消费组名字
• consumeEnable：是否可以消费，设置为FALSE的话将无法进行消费
• consumeBroadcastEnable：是否可以广播消费
• retryQueueNums：重试队列的大小
• brokerId：正常情况从哪消费
• whichBrokerWhenConsumeSlowly：出问题了从哪消费

• 终端：在线的消费客户端查看，包括版本订阅信息和消费模式
• 消费详情：对应消费组的消费明细查看，这个消费组订阅的所有Topic的消费情况，每个queue对应的消费client查看（包括Retry消息）
• 配置：查看/变更消费组的配置
• 删除：在指定的broker上删除消费组（谨慎操作）

生产者

• 展示在线的消息生产者客户端（主机、版本、地址等信息），可以通过组名进行过滤
• 通过主题进行筛选

消息

• 根据Topic和时间区间查询（由于数据量大 最多只会展示2000条，多的会被忽略）
• 根据Message Key和Topic进行查询

• 最多只会展示64条

• 根据Message Id和Topic进行消息的查询
• 消息详情：展示这条消息的详细信息，查看消息对应到具体消费组的消费情况（如果异常，可以查看具体的异常信息）。可以向指定的消费组重发消息。

消息轨迹

• 根据Message Key和Topic进行查询

• 最多只会展示64条

• 根据Message Id和Topic进行消息的查询
• 消息轨迹详情：展示展示这条消息的消息轨迹的详细信息

参考资料

• RocketMQ 官网
• RocketMQ 官方文档