一、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-releasenohup 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-releasenohup 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的都是表示 SlavebrokerId=0#nameServer地址,分号分割namesrvAddr=worker1:9876;worker2:9876;worker3:9876#是否允许 Broker 自动创建Topic,建议线下开启,线上关闭autoCreateTopicEnable=truedeleteWhen=04fileReservedTime=120#存储路径storePathRootDir=/app/rocketmq/storestorePathCommitLog=/app/rocketmq/store/commitlogstorePathConsumeQueue=/app/rocketmq/store/consumequeuestorePathIndex=/app/rocketmq/store/indexstoreCheckpoint=/app/rocketmq/store/checkpointabortFile=/app/rocketmq/store/abort#Broker 的角色brokerRole=ASYNC_MASTERflushDiskType=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的都是表示 SlavebrokerId=1#nameServer地址,分号分割namesrvAddr=worker1:9876;worker2:9876;worker3:9876#是否允许 Broker 自动创建Topic,建议线下开启,线上关闭autoCreateTopicEnable=truedeleteWhen=04fileReservedTime=120#存储路径storePathRootDir=/app/rocketmq/storeSlavestorePathCommitLog=/app/rocketmq/storeSlave/commitlogstorePathConsumeQueue=/app/rocketmq/storeSlave/consumequeuestorePathIndex=/app/rocketmq/storeSlave/indexstoreCheckpoint=/app/rocketmq/storeSlave/checkpointabortFile=/app/rocketmq/storeSlave/abort#Broker 的角色brokerRole=SLAVEflushDiskType=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的都是表示 SlavebrokerId=0#nameServer地址,分号分割namesrvAddr=worker1:9876;worker2:9876;worker3:9876#是否允许 Broker 自动创建Topic,建议线下开启,线上关闭autoCreateTopicEnable=truedeleteWhen=04fileReservedTime=120#存储路径storePathRootDir=/app/rocketmq/storestorePathCommitLog=/app/rocketmq/store/commitlogstorePathConsumeQueue=/app/rocketmq/store/consumequeuestorePathIndex=/app/rocketmq/store/indexstoreCheckpoint=/app/rocketmq/store/checkpointabortFile=/app/rocketmq/store/abort#Broker 的角色brokerRole=ASYNC_MASTERflushDiskType=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的都是表示 SlavebrokerId=1#nameServer地址,分号分割namesrvAddr=worker1:9876;worker2:9876;worker3:9876#是否允许 Broker 自动创建Topic,建议线下开启,线上关闭autoCreateTopicEnable=truedeleteWhen=04fileReservedTime=120#存储路径storePathRootDir=/app/rocketmq/storeSlavestorePathCommitLog=/app/rocketmq/storeSlave/commitlogstorePathConsumeQueue=/app/rocketmq/storeSlave/consumequeuestorePathIndex=/app/rocketmq/storeSlave/indexstoreCheckpoint=/app/rocketmq/storeSlave/checkpointabortFile=/app/rocketmq/storeSlave/abort#Broker 的角色brokerRole=SLAVEflushDiskType=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-releasenohup 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-releasenohup 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 = RaftClusterbrokerName=RaftNode00listenPort=30911namesrvAddr=master:9876;node1:9876;node2:9876storePathRootDir=/app/rocketmq/storeDledger/storePathCommitLog=/app/rocketmq/storeDledger/commitlogstorePathConsumeQueue=/app/rocketmq/storeDledger/consumequeuestorePathIndex=/app/rocketmq/storeDledger/indexstoreCheckpoint=/app/rocketmq/storeDledger/checkpointabortFile=/app/rocketmq/storeDledger/abortenableDLegerCommitLog=truedLegerGroup=RaftNode00dLegerPeers=n0-master:40911;n1-node1:40911;n2-node2:40911## must be uniquedLegerSelfId=n0sendMessageThreadPoolNums=16

node1的配置

brokerClusterName = RaftClusterbrokerName=RaftNode00listenPort=30911namesrvAddr=master:9876;node1:9876;node2:9876storePathRootDir=/app/rocketmq/storeDledger/storePathCommitLog=/app/rocketmq/storeDledger/commitlogstorePathConsumeQueue=/app/rocketmq/storeDledger/consumequeuestorePathIndex=/app/rocketmq/storeDledger/indexstoreCheckpoint=/app/rocketmq/storeDledger/checkpointabortFile=/app/rocketmq/storeDledger/abortenableDLegerCommitLog=truedLegerGroup=RaftNode00dLegerPeers=n0-master:40911;n1-node1:40911;n2-node2:40911## must be uniquedLegerSelfId=n1sendMessageThreadPoolNums=16

node2的配置

brokerClusterName = RaftClusterbrokerName=RaftNode00listenPort=30911namesrvAddr=master:9876;node1:9876;node2:9876storePathRootDir=/app/rocketmq/storeDledger/storePathCommitLog=/app/rocketmq/storeDledger/commitlogstorePathConsumeQueue=/app/rocketmq/storeDledger/consumequeuestorePathIndex=/app/rocketmq/storeDledger/indexstoreCheckpoint=/app/rocketmq/storeDledger/checkpointabortFile=/app/rocketmq/storeDledger/abortenableDLegerCommitLog=truedLegerGroup=RaftNode00dLegerPeers=n0-master:40911;n1-node1:40911;n2-node2:40911## must be uniquedLegerSelfId=n2sendMessageThreadPoolNums=16

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