RocketMQ第一课-快速实战以及集群架构搭建

一、RocketMQ产品特点

1、RocketMQ介绍

​ RocketMQ是阿里巴巴开源的一个消息中间件，在阿里内部历经了双十一等很多高并发场景的考验，能够处理亿万级别的消息。2016年开源后捐赠给Apache，现在是Apache的一个顶级项目。

​ 早期阿里使用ActiveMQ，但是，当消息开始逐渐增多后，ActiveMQ的IO性能很快达到了瓶颈。于是，阿里开始关注Kafka。但是Kafka是针对日志收集场景设计的，他的高级功能并不是很贴合阿里的业务场景。尤其当他的Topic过多时，由于Partition文件也会过多，这就会加大文件索引的耗时，会严重影响IO性能。于是阿里才决定自研中间件，最早叫做MetaQ，后来改名成为RocketMQ。最早他所希望解决的最大问题就是多Topic下的IO性能压力。但是产品在阿里内部的不断改进，RocketMQ开始体现出一些不一样的优势。

2、RocketMQ特点

当今互联网MQ产品众多，其中，影响力和使用范围最大的当数Apache Kafka、RabbitMQ、Apache RocketMQ以及Apache Plusar。这几大产品虽然都是典型的MQ产品，但是由于设计和实现上的一些差异，造成他们适合于不同的细分场景。

二、RocketMQ快速实战

1、快速搭建RocketMQ服务

​ RocketMQ的官网地址： http://rocketmq.apache.org 。在下载页面可以获取RocketMQ的源码包以及运行包。下载页面地址： https://rocketmq.apache.org/download

​ 运行只需要下载Binary运行版本就可以了。 当然，源码包也建议下载下来，后续会进行解读。运行包下载下来后，就可以直接解压，上传到服务器上。我们这里会上传到/app/rocketmq目录。解压后几个重要的目录如下:

接下来，RocketMQ建议的运行环境需要至少12G的内存，这是生产环境比较理想的资源配置。但是，学习阶段，如果你的服务器没有这么大的内存空间，那么就需要做一下调整。进入bin目录，对其中的runserver.sh和runbroker.sh两个脚本进行一下修改。

​ 使用vi runserver.sh指令，编辑这个脚本，找到下面的一行配置，调整Java进程的内存大小。

​ 接下来，同样调整runbroker.sh中的内存大小。

​ RocketMQ的后端服务分为nameserver和broker两个服务

• 第一步：启动nameserver服务。

cd /app/rocketmq/rocketmq-all-4.9.5-bin-release
nohup bin/mqnamesrv &

​ 接下来，可以通过jsp指令进行验证。使用jps指令后，可以看到有一个NamesrvStartup的进程运行，也表示nameserver服务启动完成。

• 第二步：启动broker服务。
  ​ 启动broker服务之前，要做一个小小的配置。进入RocketMQ安装目录下的conf目录，修改broker.conf文件，在文件最后面加入一个配置：

#自动创建主题
autoCreateTopicEnable=true

然后也可以用之前的方式启动broker服务。启动broker服务的指令是mqbroker

cd /app/rocketmq/rocketmq-all-4.9.5-bin-release
nohup bin/mqbroker &

注：1、在实际服务部署时，通常会将RocketMQ的部署地址添加到环境变量当中。例如使用vi ~/.bash_profile指令，添加以下内容。

export ROCKETMQ_HOME=/app/rocketmq/rocketmq-all-4.9.5-bin-releasePATH=$ROCKETMQ_HOME/bin:$PATHexport PATH

这样就不必每次进入RocketMQ的安装目录了。直接可以使用mqnamesrv 和mqbroker指令。

2、停止RocketMQ服务可以通过mqshutdown指令进行

mqshutdown namesrv # 关闭nameserver服务

mqshutdown broker # 关闭broker服务

2、快速实现消息收发

​ RocketMQ后端服务启动完成后，就可以启动客户端的消息生产者和消息消费者进行消息转发了。接下来，我们会先通过RocketMQ提供的命令行工具快速体验一下RocketMQ消息收发的功能。然后，再动手搭建一个Maven项目，在项目中使用RocketMQ进行消息收发。

1、命令行快速实现消息收发

第一步：需要配置一个环境变量NAMESRV_ADDR，只想我们之前启动的nameserver服务。
通过vi ~/.bash_profile添加以下配置。然后使用source ~/.bash_profile让配置生效。

export NAMESRV_ADDR='localhost:9876' 

第二步通过指令启动RocketMQ的消息生产者发送消息。

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

这个指令会默认往RocketMQ中发送1000条消息。在命令行窗口可以看到发送消息的日志：

第三步：可以启动消息消费者接收之前发送的消息

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

​ 每一条这样的日志信息就表示消费者接收到了一条消息。

​ 这个Consumer消费者的指令并不会主动结束，他会继续挂起，等待消费新的消息。我们可以使用CTRL+C停止该进程。

3、搭建RocketMQ可视化管理服务

​ 在之前的简单实验中，RocketMQ都是以后台服务的方式在运行，我们并不很清楚RocketMQ是如何运行的。RocketMQ的社区就提供了一个图形化的管理控制台Dashboard，可以用可视化的方式直接观测并管理RocketMQ的运行过程。
备注：直接放了打包好的jar包：
链接：https://pan.baidu.com/s/15_w9Q0t7dFVGHiTLJcHuWg
提取码：bzg2
​ Dashboard服务并不在RocketMQ的运行包中，需要到RocketMQ的官网下载页面单独下载。
​ 这里只提供了源码，并没有提供直接运行的jar包。将源码下载下来后，需要解压并进入对应的目录，使用maven进行编译。(需要提前安装maven客户端)

mvn clean package -Dmaven.test.skip=true

编译完成后，在源码的target目录下会生成可运行的jar包rocketmq-dashboard-1.0.1-SNAPSHOT.jar。接下来可以将这个jar包上传到服务器上。我们上传到/app/rocketmq/rocketmq-dashboard目录下
接下来我们需要在jar包所在的目录下创建一个application.yml配置文件，在配置文件中做如下配置：

rocketmq: 
  config: 
    namesrvAddrs: 
      - 192.168.232.128:9876 

主要是要指定nameserver的地址。

接下来就可以通过java指令执行这个jar包，启动管理控制台服务。

java -jar rocketmq-dashboard-1.0.1-SNAPSHOT.jar

4、升级分布式集群

​ 之前我们用一台Linux服务器，快速搭建起了一整套RocketMQ的服务。但是很明显，这样搭建的服务是无法放到生产环境上去用的。一旦nameserver服务或者broker服务出现了问题，整个RocketMQ就无法正常工作。而且更严重的是，如果服务器出现了问题，比如磁盘坏了，那么存储在磁盘上的数据就会丢失。这时RocketMQ暂存到磁盘上的消息也会跟着丢失，这个问题就非常严重了。因此，我们需要搭建一个分布式的RocketMQ服务集群，来防止单点故障问题。

​ RocketMQ的分布式集群基于主从架构搭建。在多个服务器组成的集群中，指定一部分节点作为Master节点，负责响应客户端的请求。指令另一部分节点作为Slave节点，负责备份Master节点上的数据，这样，当Master节点出现故障时，在Slave节点上可以保留有数据备份，至少保证数据不会丢失。
整个集群方案如下图所示：

​ 接下来我们准备三台相同的Linux服务器，搭建一下RocketMQ的分布式集群。为了更清晰的描述这三台服务器上的操作，我们给每个服务器指定一个机器名。

​ 为了便于观察，我们这次搭建一个2主2从的RocketMQ集群，并将主节点和节点都分别部署在不同的服务器上。预备的集群规划情况如下：
主机 nameServer服务部署 broker服务部署
master nameServer
node1 nameServer broker-a,broker-b-s
node2 nameServer broker-a-s,broker-b

解释：即master主机只部署nameServer 注册中心，node1部署主节点a和从节点b ，node2部署从节点a和主节点b

第一步：部署nameServer服务。
​ nameServer服务不需要做特别的配置，按照之前的步骤，在三台服务器上都分别部署nameServer服务即可。

第二步：对Broker服务进行集群配置。
​ 这里需要修改RocketMQ的配置文件，对broker服务做一些集群相关的参数部署。这些配置文件并不需要我们手动进行创建，在RocketMQ运行包的conf目录下，提供了多种集群的部署配置文件模板。

• 2m-noslave: 2主无从的集群参考配置。这种集群存在单点故障。
• 2m-2s-async和2m-2s-sync: 2主2从的集群参考配置。其中async和sync表示主节点与从节点之间是同步同步还是异步同步
• dledger: 具备主从切换功能的高可用集群。集群中的节点会基于Raft协议随机选举出一个Leader，其作用类似于Master节点。其他的节点都是follower，其作用类似于Slave节点。

我们这次采用2m-2s-async的方式搭建集群，需要在node1和node2上修改这个文件夹下的配置文件。
1> 配置第一组broker-a服务
在worker2机器上配置broker-a的MASTER服务，需要修改conf/2m-2s-async/broker-a.properties。示例配置如下：

#所属集群名字，名字一样的节点就在同一个集群内
brokerClusterName=rocketmq-cluster
#broker名字，名字一样的节点就是一组主从节点。
brokerName=broker-a
#brokerid,0就表示是Master，>0的都是表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=worker1:9876;worker2:9876;worker3:9876
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
deleteWhen=04
fileReservedTime=120
#存储路径
storePathRootDir=/app/rocketmq/store
storePathCommitLog=/app/rocketmq/store/commitlog
storePathConsumeQueue=/app/rocketmq/store/consumequeue
storePathIndex=/app/rocketmq/store/index
storeCheckpoint=/app/rocketmq/store/checkpoint
abortFile=/app/rocketmq/store/abort
#Broker 的角色
brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH
#Broker 对外服务的监听端口
listenPort=10911

这里对几个需要重点关注的属性，做下简单介绍：

• brokerClusterName:
  集群名。RocketMQ会将同一个局域网下所有brokerClusterName相同的服务自动组成一个集群，这个集群可以作为一个整体对外提供服务
• brokerName:
  Broker服务名。同一个RocketMQ集群当中，brokerName相同的多个服务会有一套相同的数据副本。同一个RocketMQ集群中，是可以将消息分散存储到多个不同的brokerName服务上的。
• brokerId:
  RocketMQ中对每个服务的唯一标识。RocketMQ对brokerId定义了一套简单的规则，master节点需要固定配置为0，负责响应客户端的请求。slave节点配置成其他任意数字，负责备份master上的消息。
• brokerRole:
  服务的角色。这个属性有三个可选项：ASYNC_MASTER，SYNC_MASTER和SLAVE。其中，ASYNC_MASTER和SYNC_MASTER表示当前节点是master节点，目前暂时不用关心他们的区别。SLAVE则表示从节点。
• namesrvAddr:
  nameserver服务的地址。nameserver服务默认占用9876端口。多个nameserver地址用；隔开。

​ 接下来在worekr3上配置broker-a的SLAVE服务。需要修改conf/2m-2s-async/broker-a-s.properties。示例配置如下：

#所属集群名字，名字一样的节点就在同一个集群内
brokerClusterName=rocketmq-cluster
#broker名字，名字一样的节点就是一组主从节点。
brokerName=broker-a
#brokerid,0就表示是Master，>0的都是表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=worker1:9876;worker2:9876;worker3:9876
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
deleteWhen=04
fileReservedTime=120
#存储路径
storePathRootDir=/app/rocketmq/storeSlave
storePathCommitLog=/app/rocketmq/storeSlave/commitlog
storePathConsumeQueue=/app/rocketmq/storeSlave/consumequeue
storePathIndex=/app/rocketmq/storeSlave/index
storeCheckpoint=/app/rocketmq/storeSlave/checkpoint
abortFile=/app/rocketmq/storeSlave/abort
#Broker 的角色
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
#Broker 对外服务的监听端口
listenPort=11011

2> 配置第二组borker-b服务
​ 与第一组broker-a服务的配置方式类似，在worker3上配置broker-b的MASTER服务。需要修改conf/2m-2s-async/broker-b.properties文件

#所属集群名字，名字一样的节点就在同一个集群内
brokerClusterName=rocketmq-cluster
#broker名字，名字一样的节点就是一组主从节点。
brokerName=broker-b
#brokerid,0就表示是Master，>0的都是表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=worker1:9876;worker2:9876;worker3:9876
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
deleteWhen=04
fileReservedTime=120
#存储路径
storePathRootDir=/app/rocketmq/store
storePathCommitLog=/app/rocketmq/store/commitlog
storePathConsumeQueue=/app/rocketmq/store/consumequeue
storePathIndex=/app/rocketmq/store/index
storeCheckpoint=/app/rocketmq/store/checkpoint
abortFile=/app/rocketmq/store/abort
#Broker 的角色
brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH
#Broker 对外服务的监听端口
listenPort=10911

​ 在worker2上配置broker-b的SLAVE服务。需要修改conf/2m-2s-async/broker-b-s.properties文件，配置示例如下：

#所属集群名字，名字一样的节点就在同一个集群内
brokerClusterName=rocketmq-cluster
#broker名字，名字一样的节点就是一组主从节点。
brokerName=broker-b
#brokerid,0就表示是Master，>0的都是表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=worker1:9876;worker2:9876;worker3:9876
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
deleteWhen=04
fileReservedTime=120
#存储路径
storePathRootDir=/app/rocketmq/storeSlave
storePathCommitLog=/app/rocketmq/storeSlave/commitlog
storePathConsumeQueue=/app/rocketmq/storeSlave/consumequeue
storePathIndex=/app/rocketmq/storeSlave/index
storeCheckpoint=/app/rocketmq/storeSlave/checkpoint
abortFile=/app/rocketmq/storeSlave/abort
#Broker 的角色
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
#Broker 对外服务的监听端口
listenPort=11011

第三步：启动Broker服务
​ 集群配置完成后，需要启动Broker服务。与之前启动broker服务稍有不同，启动时需要增加-c参数，指向我们修改的配置文件。

​ 在node1上启动broker-a的master服务和broker-b的slave服务：

cd /app/rocketmq/rocketmq-all-4.9.5-bin-release
nohup bin/mqbroker -c ./conf/2m-2s-async/broker-a.properties &
nohup bin/mqbroker -c ./conf/2m-2s-async/broker-b-s.properties &

​ 在node2上启动broker-b的master服务和broker-a的slave服务：

cd /app/rocketmq/rocketmq-all-4.9.5-bin-release
nohup bin/mqbroker -c ./conf/2m-2s-async/broker-b.properties &
nohup bin/mqbroker -c ./conf/2m-2s-async/broker-a-s.properties &

第四步：检查集群服务状态

在RocketMQ的这种主从架构的集群下，客户端发送的消息会分散保存到broker-a和broker-b两个服务上，然后每个服务都配有slave服务，可以备份对应master服务上的消息，这样就可以防止单点故障造成的消息丢失问题。

5、升级高可用集群

​ 主从架构的RocketMQ集群，由于给每个broker服务配置了一个或多个slave备份服务，可以保证当broker服务出现问题时，broker上的消息不会丢失。但是，这种主从架构的集群却也有一个不足的地方，那就是不具备服务高可用。

​ 这里所说的服务高可用，并不是并不是指整个RocketMQ集群就不能对外提供服务了，而是指集群中的消息就不完整了。实际上，当RocketMQ集群中的broker宕机后，整个集群会自动进行broker状态感知。后续客户端的各种请求，依然可以转发到其他正常的broker上。只不过，原本保存在当前broker上的消息，就无法正常读取了，需要等到当前broker服务重启后，才能重新被消息消费者读取。

​ 当一个broker上的服务宕机后，我们可以从对应的slave服务上找到broker上所有的消息。但是很可惜，主从架构中各个服务的角色都是固定了的，slave服务虽然拥有全部的数据，但是它没办法升级成为master服务去响应客户端的请求，依然只是傻傻等待master服务重启后，继续做它的数据备份工作。

​ 这时，我们自然就希望这个slave服务可以升级成为master服务，继续响应客户端的各种请求，这样整个集群的消息服务就不会有任何中断。而RocketMQ提供的Dledger集群，就是具备角色自动转换功能的高可用集群。

在Dledger集群中，就不再单独指定各个broker的服务，而是由这些broker服务自行进行选举，产生一个Leader角色的服务，响应客户端的各种请求。而其他的broker服务，就作为Follower角色，负责对Leader上的数据进行备份。当然，Follower所要负责的事情，比主从架构中的SLAVE角色会要复杂一点，因为这种节点选举是在后端不断进行的，他们需要随时做好升级成Leader的准备。

​ Dledger集群的选举是通过Raft协议进行的，Raft协议是一种多数同意机制。也就是每次选举需要有集群中超过半数的节点确认，才能形成整个集群的共同决定。同时，这也意味着在Dledger集群中，只要有超过半数的节点能够正常工作，那么整个集群就能正常工作。因此，在部署Dledger集群时，通常都是部署奇数台服务，这样可以让集群的容错性达到最大。

​ 接下来，我们就用之前准备的3台服务器，搭建一个3个节点的Dledger集群。在这个集群中，只需要有2台Broker服务正常运行，这个集群就能正常工作。

第一步：部署nameserver
这一步和之前部署主从集群没有区别，不需要做过多的配置，直接在三台服务器上启动nameserver服务即可。

​ 实际上，如果你是从上一个主从架构开始搭建起来的话，那么nameserver集群都不需要重新启动，nameserver会自动感知到broker的变化。

第二步：对Broker服务进行集群配置。
​ 对于Dledger集群的配置，RocketMQ依然贴心的给出了完整的示例，不需要强行记忆。

​ 在conf/dledger目录下，RocketMQ默认给出了三个配置文件，这三个配置文件可以在单机情况下直接部署成一个具有三个broker服务的Dledger集群，我们只需要按照这个配置进行修改即可。
​ 接下来我们可以在三台机器的conf/dledger目录下，都创建一个broker.conf文件，对每个broker服务进行配置。
master的配置

brokerClusterName = RaftCluster
brokerName=RaftNode00
listenPort=30911
namesrvAddr=master:9876;node1:9876;node2:9876
storePathRootDir=/app/rocketmq/storeDledger/
storePathCommitLog=/app/rocketmq/storeDledger/commitlog
storePathConsumeQueue=/app/rocketmq/storeDledger/consumequeue
storePathIndex=/app/rocketmq/storeDledger/index
storeCheckpoint=/app/rocketmq/storeDledger/checkpoint
abortFile=/app/rocketmq/storeDledger/abort
enableDLegerCommitLog=true
dLegerGroup=RaftNode00
dLegerPeers=n0-master:40911;n1-node1:40911;n2-node2:40911
## must be unique
dLegerSelfId=n0
sendMessageThreadPoolNums=16

node1的配置

brokerClusterName = RaftCluster
brokerName=RaftNode00
listenPort=30911
namesrvAddr=master:9876;node1:9876;node2:9876
storePathRootDir=/app/rocketmq/storeDledger/
storePathCommitLog=/app/rocketmq/storeDledger/commitlog
storePathConsumeQueue=/app/rocketmq/storeDledger/consumequeue
storePathIndex=/app/rocketmq/storeDledger/index
storeCheckpoint=/app/rocketmq/storeDledger/checkpoint
abortFile=/app/rocketmq/storeDledger/abort
enableDLegerCommitLog=true
dLegerGroup=RaftNode00
dLegerPeers=n0-master:40911;n1-node1:40911;n2-node2:40911
## must be unique
dLegerSelfId=n1
sendMessageThreadPoolNums=16

node2的配置

brokerClusterName = RaftCluster
brokerName=RaftNode00
listenPort=30911
namesrvAddr=master:9876;node1:9876;node2:9876
storePathRootDir=/app/rocketmq/storeDledger/
storePathCommitLog=/app/rocketmq/storeDledger/commitlog
storePathConsumeQueue=/app/rocketmq/storeDledger/consumequeue
storePathIndex=/app/rocketmq/storeDledger/index
storeCheckpoint=/app/rocketmq/storeDledger/checkpoint
abortFile=/app/rocketmq/storeDledger/abort
enableDLegerCommitLog=true
dLegerGroup=RaftNode00
dLegerPeers=n0-master:40911;n1-node1:40911;n2-node2:40911
## must be unique
dLegerSelfId=n2
sendMessageThreadPoolNums=16

第四步：检查集群服务状态