高性能分布式消息中间件—RocketMQ

Posted 2021-04-02 云客时间

我是这样规划的，想从以下的技术栈记录：

1：分布式系统消息中间件。

2：分布式搜索引擎。

3：分布式JOB。

4：分布式缓存。

5：分库分表。

6：手把手教你打造一个第三方公众平台。

7：分布式系统性能监控技术。

8：Spring生态源码解读。

9：Spring生态实战技术。

10：并发技术汇总。

11：jvm调优。

12：netty。

13：架构实战。

14：java基础。

15：mysql & TIDB。

简介：

RocketMQ作为一款纯java、是由阿里巴巴公司开发的分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。

由开源社区killme2008维护，开源社区非常活跃。https://github.com/killme2008/Metamorphosis。Metaq 2.x。于2012年10月份上线，在淘宝内部被广泛使用 。Metaq 3.0发布时，产品名称改为RocketMQ。基于公司内部开源共建原则，RocketMQ项目只维护核心功能，且去除了所有其他运行时的依赖，核心功能最简化。每个BU的个性化需求都在RocketMQ项目之上进行深度定制。RocketMQ向其他BU提供的仅仅是jar包，例如要定制一个Broker，那么只需要依赖rocketmq-broker这个jar包即可，可通过API进行交互，如果定制client,则依赖rocketmq-client这个jar包，对其提供的api进行再封装。如今的rocketmq孵化成了Apache的顶级开源项目。

特点：

• 支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型
• 在一个队列中可靠的先进先出（FIFO）和严格的顺序传递 （RocketMQ可以保证严格的消息顺序，而ActiveMQ无法保证）
• 支持拉（pull）和推（push）两种消息模式 （Push好理解，比如在消费者端设置Listener回调；而Pull，控制权在于应用，即应用需要主动的调用拉消息方法从Broker获取消息，这里面存在一个消费位置记录的问题（如果不记录，会导致消息重复消费））
• 单一队列百万消息的堆积能力       （RocketMQ提供亿级消息的堆积能力，这不是重点，重点是堆积了亿级的消息后，依然保持写入低延迟）
• 支持多种消息协议，如 JMS、MQTT 等
• 分布式高可用的部署架构,满足至少一次消息传递语义    （RocketMQ原生就是支持分布式的，而ActiveMQ原生存在单点性）
• 提供 docker 镜像用于隔离测试和云集群部署
• 提供配置、指标和监控等功能丰富的 Dashboard

在Metaq1.x/2.x的版本中，分布式协调采用的是Zookeeper，而RocketMQ自己实现了一个NameServer，更加轻量级，性能更好！

• **组（Group）**有Producer/Consumer Group。ActiveMQ中并没有Group这个概念，而在RocketMQ中理解Group的机制很重要。通过Group机制，让RocketMQ天然的支持消息负载均衡！比如某个Topic有9条消息，其中一个Consumer Group有3个实例（3个进程 OR 3台机器），那么每个实例将均摊3条消息！（注意RocketMQ只有一种模式，即发布订阅模式。）
• 消息失败重试机制、高效的订阅者水平扩展能力、强大的API、事务机制等等。

专业术语

Producer

消息生产者，生产者的作用就是将消息发送到 MQ，生产者本身既可以产生消息，如读取文本信息等。也可以对外提供接口，由外部应用来调用接口，再由生产者将收到的消息发送到 MQ。

Producer Group

生产者组，简单来说就是多个发送同一类消息的生产者称之为一个生产者组。在这里可以不用关心，只要知道有这么一个概念即可。

Consumer

消息消费者，简单来说，消费 MQ 上的消息的应用程序就是消费者，至于消息是否进行逻辑处理，还是直接存储到数据库等取决于业务需要。

Consumer Group

消费者组，和生产者类似，消费同一类消息的多个 consumer 实例组成一个消费者组。

Topic

Topic 是一种消息的逻辑分类，比如说你有订单类的消息，也有库存类的消息，那么就需要进行分类，一个是订单 Topic 存放订单相关的消息，一个是库存 Topic 存储库存相关的消息。

Message

Tag

标签可以被认为是对 Topic 进一步细化。一般在相同业务模块中通过引入标签来标记不同用途的消息。

Broker

Broker 是 RocketMQ 系统的主要角色，其实就是前面一直说的 MQ。Broker 接收来自生产者的消息，储存以及为消费者拉取消息的请求做好准备。

Name Server

Name Server 为 producer 和 consumer 提供路由信息。

rocketmq基于Dledger构建高可用

构建集群需要至少3台服务器

本人的RocketMq集群是基于阿里云的centos7.9版本。网上很多教程写的基于虚拟机的本地环境。我认为要玩儿就玩儿的正式一点，真正可以接近生产环境的玩法。

构建安装DLedger

git clone https://github.com/openmessaging/openmessaging-storage-dledger.git

cd openmessaging-storage-dledger

mvn clean install -DskipTests

另外系统中还要安装一个git，用作拉取rocketmq控制台的源码工程。

下载RocketMQ

wget https://github.com/apache/rocketmq/releases

unzip rocketmq-all-4.7.1-source-release.zip

cd rocketmq-all-4.7.1/

mvn -Prelease-all -DskipTests clean install -U

cd distribution/target/rocketmq-4.7.1/rocketmq-4.7.1

这里注意，rocketmq只有4.5以上版本支持dledger集群构建

修改配置文件

服务器说明：(生产中应该将 NameServer 部署到其他服务器中，在这为了方便，与Broker部署在一起)

安装Maven

cd /opt
wget http://mirrors.cnnic.cn/apache/maven/maven-3/3.5.4/binaries/apache-maven-3.5.4-bin.tar.gz
tar -zxvf apache-maven-3.5.4-bin.tar.gz

# 修改环境变量
vim /etc/profile
# 最下面添加
export MAVEN_HOME=/opt/apache-maven-3.5.4
export PATH=$MAVEN_HOME/bin:$PATH
# 保存退出
source /etc/profile
# 建立软连接
ln -s /opt/apache-maven-3.5.4/bin/mvn /usr/bin/mvn

安装JDK

这里需要自己前往oracle官方下载

配置java环境变量

环境变量配置	配置
""	export JAVA_HOME=/opt/jdk1.8.0_261
vi /etc/profile	export CLASSPATH=${JAVA_HOME}/lib
""	export PATH= {JAVA_HOME}/bin

source /etc/profile 刷新配置

修改配置文件

服务器	ip	安装的服务
服务器1-主	172.30.134.189	DLedger，Broker，NameServer
服务器2-从	172.30.134.190	DLedger，Broker，NameServer
服务器3-从	172.30.134.191	DLedger，Broker，NameServer

服务器1配置-Master

进入rocketmq安装目录找到

vim conf/dledger/broker-n0.conf

修改Broker配置

## 集群名
brokerClusterName = RaftCluster
### broker组名，同一个RaftClusterGroup内，brokerName名要一样
brokerName=RaftNode00
### 监听的端口
listenPort=10911
### 你设置的NameServer地址和端口
namesrvAddr=172.30.134.189:9876;172.30.134.190:9876;172.30.134.191:9876
storePathRootDir=/tmp/rmqstore/node00
storePathCommitLog=/tmp/rmqstore/node00/commitlog
enableDLegerCommitLog=true
dLegerGroup=RaftNode00
### n0 n1 n2 分别是broker1,broker2,broker3 的 dLegerSelfId
### 例如：dLegerPeers=n0-服务器1的IP:40911;n1-服务器2的IP:40912;n2-服务器3的IP:40913
dLegerPeers=n0-172.30.134.189:40911;n1-172.30.134.190:40912;n2-172.30.134.191:40913
### must be unique
autoCreateTopicEnable=true
### 这个值必须是在同一个RaftClusterGroup内唯一的
dLegerSelfId=n0
sendMessageThreadPoolNums=4
### 由于我的虚拟机配置了多个网卡，所以会绑定ip错误，因此我配置了这项，
brokerIP1=master主机的外网IP地址

服务器2配置-Slave

vim conf/dledger/broker-n1.conf

修改Broker配置

brokerClusterName = RaftCluster
brokerName=RaftNode00
listenPort=10921
namesrvAddr=172.30.134.189:9876;172.30.134.190:9876;172.30.134.191:9876
storePathRootDir=/tmp/rmqstore/node01
storePathCommitLog=/tmp/rmqstore/node01/commitlog
enableDLegerCommitLog=true
dLegerGroup=RaftNode00
dLegerPeers=n0-172.30.134.189:40911;n1-172.30.134.190:40912;n2-172.30.134.191:40913
## must be unique
dLegerSelfId=n1
autoCreateTopicEnable=true
sendMessageThreadPoolNums=4
brokerIP1=内网IP

这里有个坑需要注意一下，如果说，你的master主机的brokerIP1设置了外网的ip那么你的slave节点的brokerIP1就要设置为内网的IP

服务器3配置-Slave

vim conf/dledger/broker-n2.conf 修改Broker配置

brokerClusterName = RaftCluster
brokerName=RaftNode00
listenPort=10931
namesrvAddr=172.30.134.189:9876;172.30.134.190:9876;172.30.134.191:9876
storePathRootDir=/tmp/rmqstore/node02
storePathCommitLog=/tmp/rmqstore/node02/commitlog
enableDLegerCommitLog=true
dLegerGroup=RaftNode00
dLegerPeers=n0-172.30.134.189:40911;n1-172.30.134.190:40912;n2-172.30.134.191:40913
## must be unique
autoCreateTopicEnable=true
dLegerSelfId=n2
sendMessageThreadPoolNums=4
brokerIP1=172.30.134.191

启动集群

在服务器1 执行

nohup sh bin/mqnamesrv -n 172.30.134.189 & > nohubNameserv
nohup sh bin/mqbroker  > nohubBroker -c conf/dledger/broker-n0.conf autoCreateTopicEnable=true &

启动集群

在服务器2 执行

nohup sh bin/mqnamesrv -n 172.30.134.190 & > nohubNameserv
nohup sh bin/mqbroker  > nohubBroker -c conf/dledger/broker-n1.conf autoCreateTopicEnable=true &

启动集群

在服务器3 执行

nohup sh bin/mqnamesrv -n 172.30.134.191 & > nohubNameserv
nohup sh bin/mqbroker  > nohubBroker -c conf/dledger/broker-n2.conf autoCreateTopicEnable=true &

查看集群情况

sh bin/mqadmin clusterList -n 127.0.0.1:9876

备注

Broker 配置

Broker 配置

参考文档：

参数名	默认值	说明
listenPort	10911	接受客户端连接的监听端口
namesrvAddr	null	nameServer 地址
brokerIP1	网卡的 InetAddress	当前 broker 监听的 IP
brokerIP2	跟 brokerIP1 一样	存在主从 broker 时，如果在 broker 主节点上配置了 brokerIP2 属性，broker 从节点会连接主节点配置的 brokerIP2 进行同步
brokerName	null	broker 的名称
brokerClusterName	DefaultCluster	本 broker 所属的 Cluser 名称
brokerId	0	broker id, 0 表示 master, 其他的正整数表示 slave
storePathCommitLog	$HOME/store/commitlog/	存储 commit log 的路径
storePathConsumerQueue	$HOME/store/consumequeue/	存储 consume queue 的路径
mappedFileSizeCommitLog	1024 * 1024 * 1024(1G)	commit log 的映射文件大小
deleteWhen	04	在每天的什么时间删除已经超过文件保留时间的 commit log
fileReservedTime	72	以小时计算的文件保留时间
brokerRole	ASYNC_MASTER	SYNC_MASTER/ASYNC_MASTER/SLAVE
flushDiskType	ASYNC_FLUSH	SYNC_FLUSH/ASYNC_FLUSH SYNC_FLUSH 模式下的 broker 保证在收到确认生产者之前将消息刷盘。ASYNC_FLUSH 模式下的 broker 则利用刷盘一组消息的模式，可以取得更好的性能。
enableDLegerCommitLog	是否启动 DLedger	true
dLegerGroup	DLedger Raft Group的名字，建议和 brokerName 保持一致	RaftNode00
dLegerPeers	DLedger Group 内各节点的端口信息，同一个 Group 内的各个节点配置必须要保证一致	n0-172.30.134.189:40911;n1-172.30.134.190:40912;n2-172.30.134.191:40913
dLegerSelfId	节点 id, 必须属于 dLegerPeers 中的一个；同 Group 内各个节点要唯一	n0
sendMessageThreadPoolNums	发送线程个数，建议配置成 Cpu 核数	16

启动内存不够

修改 bin/runbroker.sh 和 bin/runserver.sh 中的
JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m"

报连接超时

查看防火墙服务状态 systemctl status firewalld

将防火墙关闭 systemctl stop firewalld

安装控制台

1.下载文件(/usr/local目录下)

2.解压(/usr/local目录下)

yum install -y unzip zip 前提是：unzip解压文件无法使用 unzip rocketmq-externals-master.zip 解压文件

3.修改配置文件(usr/local/rocketmq-externals-master/目录下)

find -name application.properties 可以查看到两个文件都在rocketmq-console文件目录下

vi application.properties

4.编译(usr/local/rocketmq-externals-master/rocketmq-console/目录下)

mvn clean package -Dmaven.test.skip=true 如果失败多编译几次--可能是网络问题

编译成功后，在rocketmq-console目录下会生成一个目录：target目录，该目录下有启动rocketmq界面的jar文件

5.启动web(usr/local/rocketmq-externals-master/rocketmq-console/target目录下)

java -jar rocketmq-console-ng-1.0.0.jar 启动 ---当终端断了该服务就会停止 nohup java -jar rocketmq-console-ng-1.0.0.jar >>/usr/logs/log.out 2>&1 & 后台启动 --当终端断了也不会停止服务

6.添加端口并在阿里云服务器上开通端口

firewall-cmd --zone=public --add-port=8080/tcp --permanen 永久添加端口并重启防火墙

在阿里云服务器添加出入规则。

查看web监控：

消费者组：

本篇文章就介绍到这里，读者们首先要学会安装rocketmq、以及rocketmq中的基础概念。后续的文章会对rocketmq做具体的介绍以及使用，以及如何集成Springboot和spring cloud stream。