RocketMQ之三：RocketMQ集群环境搭建

1、初步理解Producer/Consumer Group

在安装RocketMQ之前我们先来理解Group概念，在RocketMQ中Group是很重要的。通过Group机制，让RocketMQ天然的支持消息负载均衡！比如某个Topic有9条消息，其中一个Consumer Group有3个实例（3个进程 OR 3台机器），那么每个实例将均摊3条消息！（注意RocketMQ只有一种模式，即发布订阅模式。）如下图所示：


2、RocketMQ的集群部署模式：

在动手操作之前我们先来了解一下RocketMQ的集群部署模式都有哪些。RocketMQ在众多消息队列产品中高可用性是其亮点之一，其集群部署方式有很多种。比如单Master模式、多Master模式、多Master多Slave模式（异步复制）、多Master多Slave模式（同步双写）等。明确个概念，RocketMQ Slave不可以写，可以读，类似于MySQL的主从机制。

• 单Master模式：无需多言，一旦单个broker重启或宕机，一切都结束了！很显然，线上不可以使用。
• 多Master模式：全是Master，没有Slave。当然，一个broker宕机了，应用是无影响的，缺点在于宕机的Master上未被消费的消息在Master没有恢复之前不可以订阅。
• 多Master多Slave模式（异步复制）：多对Master-Slave，高可用！采用异步复制的方式，主备之间短暂延迟，MS级别。Master宕机，消费者可以从Slave上进行消费，不受影响，但是Master的宕机，会导致丢失掉极少量的消息。
• 多Master多Slave模式（同步双写）：和上面的区别点在于采用的是同步方式，也就是在Master/Slave都写成功的前提下，向应用返回成功，可见不论是数据，还是服务都没有单点，都非常可靠！缺点在于同步的性能比异步稍低。

3、RocketMQ 1个Nameserver 2个Master安装部署：

目标：熟悉RocketMQ的各模块搭建流程，部署一个测试集群。1个nameserver，两个master。
环境：两台虚拟机，java环境，mvn
虚拟机1启动一个nameserver + 1个broker。虚拟机2启动一个broker。
主要参考：https://rocketmq.apache.org/docs/quick-start/
主要操作命令上面文档中都有

环境准备

64位操作系统，建议使用Linux / Unix /

• CentOs 7.3 +
• 64bit JDK 1.8 +
• Maven 3.2.x +
• RocketMQ4.4
• RocketMQ-console

1.1安装rocketmq

可以在操作系统用wget直接下载，也可以在本机下载完后上传到服务器。
wget http://mirrors.hust.edu.cn/apache/rocketmq/4.5.2/rocketmq-all-4.5.2-bin-release.zip
RocketMQ已归属Apache开源项目，可以去apache官网下载。
解压unzip rocketmq-all-4.5.2-bin-release.zip

1.2 Download & Build from Release

Click here to download the 4.3.0 source release. Also you could download a binary release from here .

Build from source release:

> unzip rocketmq-all-4.5.2-bin-release.zip
> cd rocketmq-all-4.5.2-bin-release/
> mvn -Prelease-all -DskipTests clean install -U
> cd distribution/rocketmq-4.5.2

1.3 Start Name Server

Name Server 默认配置Xmx=4G，调整以适合部署：

vi /bin/runserver.sh

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

调整为

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

启动Name Server

nohup sh bin/mqnamesrv -n "10.200.110.91:9876" > /dev/null 2>&1 &

tail -f ~/logs/rocketmqlogs/namesrv.log

1.4 Start Broker

调整java opt：

vi /bin/runbroker.sh

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

调整为：

JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g -Xmn1g"

启动Broker

虚拟机1启动broker命令：nohup sh bin/mqbroker -n 10.200.110.91:9876 -c ./conf/2m-noslave/broker-a.properties &

虚拟机2启动broker命令：nohup sh bin/mqbroker -n 10.200.110.91:9876 -c ./conf/2m-noslave/broker-b.properties & tail -f ~/logs/rocketmqlogs/broker.log
-n ip:9876，这里的ip是nameserver的

broker-a.properties和broker-b.properties的区别：brokerName不同

1.5 查看进程

[root@node1 apache-rocketmq]# jps
2374 BrokerStartup
2350 NamesrvStartup

[root@node1 apache-rocketmq]# netstat -ntlp
Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name          
tcp       0    0127.0.0.1:25            0.0.0.0:*               LISTEN      1508/master         
tcp6      0    0:::9876                 :::*                    LISTEN      2350/java                    
tcp6      0    0::1:25                  :::*                    LISTEN      1508/master         
tcp6      0    0:::10909                :::*                    LISTEN      2374/java           
tcp6      0    0:::10911                :::*                    LISTEN      2374/java           
tcp6      0    0:::10912                :::*                    LISTEN      2374/java

1.6、控制台的使用rocketmq-console

Rocket-console做为rocketmq社区维护的产品需要从GitHub上下载，下载地址：https://github.com/apache/rocketmq-externals。下载下来后我们会发现有好多应用工程，这些全是rocketmq的外围支持系统，感兴趣的可以查看。我们需要的工程名是rocketmq-console。

打开src/main/resources/application.properties文件，修改配置：

rocketmq.config.namesrvAddr=10.200.110.91:9876 (rocketmq安装namesrv地址和端口)

进入rocketmq-console根目录，打开README.md文件，按照提示操作。Rocketmq-console的安装方式分为基于docker的安装和非docker安装。在测试环境中采用了非docker安装方式。

系统默认安装maven3以上版本，执行命令：mvn clean package -Dmaven.test.skip=true

进入target,将rocketmq-console-ng-1.0.0.jar上传到服务器执行：java -jar rocketmq-console-ng-1.0.0.jar&

在目录下生成out.log日志文件，查看启动成功。在浏览器中输入访问地址：http://10.200.110.91:8080

1.7、Send & Receive Messages

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

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

1.8、Shutdown Servers

> 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

问题

1、RemotingTooMuchRequestException: sendDefaultImpl call timeout

异常说明

• 在客户端的 Producer 运行起来准备发送消息时抛异常如下
• 因为使用的是虚拟机，设置内存也只有1G，所以从 Windows 上开发连接 虚拟机中的 nameServer 时要经过 Linux 系统的防火墙，而防火墙一般都会有超时的机制，在网络连接长时间不传输数据时，会关闭这个 TCP 的会话，关闭后再读写，就有可能导致这个异常。

解决办法

1. 因为现在主要是熟练 RocketMQ 的 API，所以直接关闭了 Linux 的防火墙，实际中如果买的是如 阿里云的云主机，应该会好一些（关闭防火墙：systemctl stop firewalld，查看：systemctl status firewalld）
2. 启动nameserver和broker时，指定IP，不指定时，console显示是这样的

　　　　　　

　　　　1）nameserver启动时指定ip：[root@localhost rocketmq-all-4.4.0-bin-release]# sh bin/mqnamesrv -n "10.200.110.91:9876"

　　　　2）broker启动时指定broker的ip及nameserver的ip：

　　　　　　A、broker的ip在配置文件中指定：[root@localhost ~]# vi /distribut/rocketmq-all-4.4.0-bin-release/conf/2m-noslave/broker-a.properties

　　　　　　　　加一行ip：brokerIP1=10.200.110.91

　　　　　　B、nameserver的ip指定，在broker的启动命令中指定：[root@localhost rocketmq-all-4.4.0-bin-release]# sh bin/mqbroker -n 10.200.110.91:9876 -c ./conf/2m-noslave/broker-a.properties

 　　　　重新启动nameserver、broker后console的显示为：

　　　　

这是我的springboot的客户端producter可以正常发送消息了。

附上常用的命令行：

2 常用命令

cd  /ROCKETMQ/bin

创建topic：

sh mqadmin updateTopic -c DefaultCluster -n localhost:9876 -t testtopic

查看topic列表：

sh mqadmin topicList -n 11.4.74.48:9876

查看指定topic路由信息：

sh mqadmin topicRoute -n 11.4.74.48:9876 -t testtopic1

查看指定topic状态：

sh mqadmin topicStatus -n 11.4.74.48:9876 -t testtopic1

查看指定topic的消息：

sh mqadmin queryMsgByOffset -b broker-b -i 1  -o 0 -t NewUpt_PsgEventTopic -n localhost:9876

创建消费者组：

sh mqadmin updateSubGroup -n localhost:9876 -c DefaultCluster -g push-consumer-group1

查看消费者组：

sh mqadmin consumerProgress -n localhost:9876

查看集群信息：

sh mqadmin clusterList -n 11.4.74.48:9876

查看broker状态

sh mqadmin brokerStatus -b 11.4.74.44:10911 -n 11.4.74.48:9876

创建生产者组：

创建订阅组：

sh mqadmin updateSubGroup -c DefaultCluster -g subgroup1 -n 11.4.74.48:9876果是对集群扩容，则可以通过指定新的broker地址在扩容的机器上创建一份新的订阅组信息

根据key查消息：

sh mqadmin queryMsgByKey -k test100 -t testtopic1  -n 11.4.74.48:9876

根据unique key查：(非msgid )

sh mqadmin queryMsgByUniqueKey  -i  0B0D308C16C418B4AAC236B77FFB03E3  -n 11.4.74.48:9876 -t testtopic1

附上我的broker的配置

[root@localhost conf]# cat 2m-noslave/broker-demo.properties 
#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-c
#0表示Master，>0表示Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=10.200.110.46:9876;10.200.110.101:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/root/store/store
#commitLog 存储路径
storePathCommitLog=/root/store/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/root/store/store/consumequeue
#消息索引存储路径
storePathIndex=/root/store/store/index
#checkpoint 文件存储路径
storeCheckpoint=/root/store/store/checkpoint
#abort 文件存储路径
abortFile=/root/store/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=ASYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128
#这里设置IP后，在web控制台显示ip了，同时java客户端可以连上
brokerIP1=10.200.110.48