RocketMQ安装

下载地址

Apache 上开源官方地址：https://rocketmq.apache.org/
GitHub 托管地址：https://github.com/apache/rocketmq
阿里官方的介绍文档：http://jm.taobao.org/2017/01/12/rocketmq-quick-start-in-10-minutes/
Apache 官方提供的 4.3.0 版本的 “快速入门” 文档：https://rocketmq.apache.org/docs/quick-start/

Linux安装

环境说明

对于 RocketMQ 4.3.0版本，官方要求环境如下，其中 Git 用于从 GitHub 获取源码，没有安装也没关系，可以直接下载
官方推荐的流程是：Linux 系统上安装 Git 工具、Maven、Java JDK
Git 工具用于直接从 GitHub 获取 RocketMQ 项目源码下载到 Linux 系统上
然后 Maven 将 RocketMQ 源码进行编译成二进制文件
安装了 Java JDK 就可以运行 RocketMQ 了

linux 版本

实际操作的 Linux 系统的是 CentOS 7.2.1511 版本


[root@localhost ~]# lsb_release -a
LSB Version:	:core-4.1-amd64:core-4.1-noarch
Distributor ID:	CentOS
Description:	CentOS Linux release 7.2.1511 (Core) 
Release:	7.2.1511
Codename:	Core
[root@localhost ~]#

JDK版本

实际使用的 Java JDK 是 1.8.0_172 版本


[root@localhost ~]# java -version
java version "1.8.0_172"
Java(TM) SE Runtime Environment (build 1.8.0_172-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.172-b11, mixed mode)
[root@localhost ~]#

JDK和RocketMQ 版本说明

Version	Client	Broker	NameServer
4.0.0-incubating	>=1.7	>=1.8	>=1.8
4.1.0-incubating	>=1.6	>=1.8	>=1.8
4.2.0	>=1.6	>=1.8	>=1.8
4.3.x	>=1.6	>=1.8	>=1.8
4.4.x	>=1.6	>=1.8	>=1.8
4.5.x	>=1.6	>=1.8	>=1.8
4.6.x	>=1.6	>=1.8	>=1.8
4.7.x	>=1.6	>=1.8	>=1.8

RocketMQ 下载

可以从apache的文件库下载 https://archive.apache.org/dist/rocketmq/这里我们选择最新版

1	wget https://archive.apache.org/dist/rocketmq/4.7.1/rocketmq-all-4.7.1-bin-release.zip

RocketMQ 安装

项目解压

1	unzip rocketmq-all-4.7.1-bin-release.zip -d /usr/local/rocketmq

修改RocketMQ启动配置

修改目录 /usr/local/rocketmq/rocketmq-all-4.7.1-bin-release/bin 下的 3 个配置文件： runserver.sh、runbroker.sh 、tools.sh不然会报insufficient memory

runserver.sh

设置 runserver.sh 中此项配置

1	JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn512m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

runbroker.sh

设置 runbroker.sh 中此项配置

1	JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"

tools.sh

设置 tools.sh 中此项配置

1	JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn256m -XX:PermSize=128m -XX:MaxPermSize=128m"

RocketMQ 启动

启动 NameServer

进入 RocketMQ 安装目录下的 /usr/local/rocketmq/rocketmq-all-4.7.1-bin-release 目录进行操作

执行命令

执行命令启动NameServer

1	nohup sh bin/mqnamesrv &

nohup sh mqnamesrv & ：属于后台启动

sh mqnamesrv ：属于终端启动，直接输出日志信息，按 ctrl+c 可直接关闭退出

查看启动状态

如下所示，看到 The Name Server boot success. serializeType=JSON 表示启动成功

在当前目录下会有一个 nohup.out 的日志文件，可以打开查看 namesrv 的启动情况

启动 Broker

同样进入 RocketMQ 安装目录下的 /usr/local/rocketmq/rocketmq-all-4.7.1-bin-release 目录进行操作

执行命令

1 2	# 启动命令，并且常驻内存:注意ip地址要配置成为服务的ip地址，保证地址以及端口能够访问 > nohup sh bin/mqbroker -n 192.168.64.144:9876 &

nohup sh bin/mqbroker -n 192.168.64.144:9876 & ：属于后台启动

sh bin/mqbroker -n 192.168.64.144:9876 ：属于终端启动，直接输出日志信息，按 ctrl+c 可直接关闭退出

查看启动状态

启动之后同样提示将日志信息追加到了当前目录下的 nohup.out 文件中

输入 tail -f ~/logs/rocketmqlogs/broker.log 查看日志，如果看到 The broker[localhost, 172.17.0.1:10911] boot success....,则表示启动成功

发送消息

发送/接收消息之前，需要告诉客户端(Producer、Consumer)名称服务器的位置，RocketMQ 提供了多种方法来实现这一点:

编程方式，如：producer.setNamesrvAddr("ip:port")
Java 选项，如：rocketmq.namesrv.addr
环境变量，如：NAMESRV_ADDR
HTTP 端点

官方指导命令

1
2
3

> export NAMESRV_ADDR=localhost:9876
 > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
 SendResult [sendStatus=SEND_OK, msgId= ...

示例

如下所示官方提供这个例子属于生产者，用于发送消息，运行之后会发送大量的消息，之后就会退出

[root@localhost apache-rocketmq]# export NAMESRV_ADDR=localhost:9876
[root@localhost apache-rocketmq]# sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option PermSize=128m; support was removed in 8.0
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=128m; support was removed in 8.0
03:05:13.152 [main] DEBUG i.n.u.i.l.InternalLoggerFactory - Using SLF4J as the default logging framework
SendResult [sendStatus=SEND_OK, msgId=C0A83A8120207D4991AD2F02D2280000, offsetMsgId=C0A83A8100002A9F00000000000AFAC8, messageQueue=MessageQueue [topic=TopicTest, brokerName=localhost.localdomain, queueId=0], queueOffset=1000]
SendResult [sendStatus=SEND_OK, msgId=C0A83A8120207D4991AD2F02D2740001, offsetMsgId=C0A83A8100002A9F00000000000AFB7A, messageQueue=MessageQueue [topic=TopicTest, brokerName=localhost.localdomain, queueId=1], queueOffset=1000]
SendResult [sendStatus=SEND_OK, msgId=C0A83A8120207D4991AD2F02D27A0002, offsetMsgId=C0A83A8100002A9F00000000000AFC2C, messageQueue=MessageQueue [topic=TopicTest, brokerName=localhost.localdomain, queueId=2], queueOffset=1000]
.........................

接收消息

官方指导命令

1 2	> sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer ConsumeMessageThread_%d Receive New Messages: [MessageExt...

示例

上面的消息发送完毕之后就会退出，在同一窗口中可以使用消费者类来进行接收消息，如下所示，显然是多线程的

[root@localhost apache-rocketmq]# sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
ConsumeMessageThread_3 Receive New Messages: [MessageExt [queueId=0, storeSize=180, queueOffset=1031, sysFlag=0, bornTimestamp=1533841514766, bornHost=/192.168.58.129:46816, storeTimestamp=1533841514767, storeHost=/192.168.58.129:10911, msgId=C0A83A8100002A9F00000000000B518A, commitLogOffset=741770, bodyCRC=895437781, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1250, CONSUME_START_TIME=1533841719674, UNIQ_KEY=C0A83A8120207D4991AD2F02D50E007C, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 49, 50, 52], transactionId='null'}]] 
ConsumeMessageThread_2 Receive New Messages: [MessageExt [queueId=0, storeSize=180, queueOffset=1030, sysFlag=0, bornTimestamp=1533841514746, bornHost=/192.168.58.129:46816, storeTimestamp=1533841514747, storeHost=/192.168.58.129:10911, msgId=C0A83A8100002A9F00000000000B4EBA, commitLogOffset=741050, bodyCRC=842174412, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1250, CONSUME_START_TIME=1533841719669, UNIQ_KEY=C0A83A8120207D4991AD2F02D4FA0078, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 49, 50, 48], transactionId='null'}]] 
ConsumeMessageThread_20 Receive New Messages: [MessageExt [queueId=0, storeSize=179, queueOffset=1019, sysFlag=0, bornTimestamp=1533841514511, bornHost=/192.168.58.129:46816, storeTimestamp=1533841514515, storeHost=/192.168.58.129:10911, msgId=C0A83A8100002A9F00000000000B2FE2, commitLogOffset=733154, bodyCRC=884882597, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1250, CONSUME_START_TIME=1533841719680, UNIQ_KEY=C0A83A8120207D4991AD2F02D40F004C, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 55, 54], transactionId='null'}]] 
...............

关闭服务器

同样都是在 RocketMQ 安装目录下的 distribution/target/apache-rocketmq 目录下执行命令

与启动顺序相反进行关闭，先关闭 broker、在关闭 nameserv

> sh bin/mqshutdown broker
The mqbroker(36695) is running...
Send shutdown request to mqbroker(36695) OK
 
> sh bin/mqshutdown namesrv
The mqnamesrv(36664) is running...
Send shutdown request to mqnamesrv(36664) OK