一、RocketMQ简介
RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件，前身是MetaQ，是阿里参考Kafka特点研发的一个队列模型的消息中间件，后开源给apache基金会成为了apache的顶级开源项目，具有高性能、高可靠、高实时、分布式特点。
RocketMQ单机也可以支持亿级的消息堆积能力。单机写入TPS单实例约7万条/秒，单机部署3个Broker，可以跑到最高12万条/秒，消息大小10个字节。
1.1 RocketMQ发展历程
2007年：淘宝实施了“五彩石”项目，“五彩石”用于将交易系统从单机变成分布式，也是在这个过程中产生了阿里巴巴第一代消息引擎——Notify。
2010年：阿里巴巴B2B部门基于ActiveMQ的5.1版本也开发了自己的一款消息引擎，称为Napoli，这款消息引擎在B2B里面广泛地被使用，不仅仅是在交易领域，在很多的后台异步解耦等方面也得到了广泛的应用。
2011年：业界出现了现在被很多大数据领域所推崇的Kafka消息引擎，阿里巴巴在研究了Kafka的整体机制和架构设计之后，基于Kafka的设计使用Java进行了完全重写并推出了MetaQ 1.0版本，主要是用于解决顺序消息和海量堆积的问题。
2012年：阿里巴巴开源其自研的第三代分布式消息中间件——RocketMQ。
经过几年的技术打磨，阿里称基于RocketMQ技术，目前双十一当天消息容量可达到万亿级。
2016年11月：阿里将RocketMQ捐献给Apache软件基金会，正式成为孵化项目。
阿里称会将其打造成顶级项目。这是阿里迈出的一大步，因为加入到开源软件基金会需要经过评审方的考核与观察。
2017年2月20日：RocketMQ正式发布4.0版本，专家称新版本适用于电商领域，金融领域，大数据领域，兼有物联网领域的编程模型。
以上就是RocketMQ的整体发展历史，其实在阿里巴巴内部围绕着RocketMQ内核打造了三款产品，分别是MetaQ、Notify和Aliware MQ。
这三者分别采用了不同的模型，MetaQ主要使用了拉模型，解决了顺序消息和海量堆积问题；Notify主要使用了推模型，解决了事务消息；而云产品Aliware MQ则是提供了商业化的版本。
1.2 RocketMQ的功能
消息传递模式包括发布/订阅、请求/回复和流式传输
财务级事务消息
基于DLedger的内置容错和高可用性配置选项
各种跨语言客户端，如java、C/C++、Python、Go
可插拔传输协议，如TCP、SSL、AIO
内置消息跟踪功能，还支持opentracing
多功能大数据与流媒体生态系统集成
按时间或偏移量的邮件追溯
同一队列中可靠的FIFO和严格有序的消息传递
高效的推拉消费模式
单队列百万级消息累积容量
多种消息传递协议，如JMS和OpenMessageing
灵活的分布式部署体系结构
闪电式批量信息交换系统
各种消息过滤机制，如SQL和Tag
用于独立测试和云隔离集群的Docker映像
功能丰富的管理仪表板，用于配置、度量和监视
身份验证和授权
免费开放源代码连接器，用于源和接收器
二、RocketMQ架构组成
2.1 集群组成
RocketMQ部署结构如下图所示：
从中间件服务角度来看整个RocketMQ消息系统（服务端）主要分为：NameServer和Broker两个部分。
2.1.1 NameServer
提供服务发现和注册，这里主要是管理Broker，NameServer接受来自Broker的注册，并通过心跳机制来检测Broker服务的健康性；
提供路由功能，集群(这里是指以集群方式部署的NameServer)中的每个NameServer都保存了Broker集群(这里是指以集群方式部署的Broker)中整个的路由信息和队列信息。这里需要注意，在NameServer集群中，每个NameServer都是相互独立的，所以每个Broker需要连接所有的NameServer，每创建一个新的topic都要同步到所有的NameServer上。
NameServer是一个功能齐全的服务器，其角色类似Dubbo中的Zookeeper，但NameServer与Zookeeper相比更轻量。主要是因为每个NameServer节点互相之间是独立的，没有任何信息交互。
NameServer压力不会太大，平时主要开销是在维持心跳和提供Topic-Broker的关系数据。
但有一点需要注意，Broker向NameServer发心跳时， 会带上当前自己所负责的所有Topic信息，如果Topic个数太多（万级别），会导致一次心跳中，就Topic的数据就几十M，网络情况差的话， 网络传输失败，心跳失败，导致NameServer误认为Broker心跳失败。
NameServer 被设计成几乎无状态的，可以横向扩展，节点之间相互之间无通信，通过部署多台机器来标记自己是一个伪集群。
每个 Broker 在启动的时候会到 NameServer 注册，Producer 在发送消息前会根据 Topic 到 NameServer 获取到 Broker 的路由信息，Consumer 也会定时获取 Topic 的路由信息。
2.1.2 Broker
负责消息的存储、传递、查询以及高可用(HA)保证等。其由如下几个子模块（源码总体上也是这么拆分的）构成：
Remoting Module：整个Broker的实体，负责处理来自clients端的请求。
Client Manager：负责管理客户端(Producer/Consumer)和维护Consumer的Topic订阅信息
Store Service：提供方便简单的API接口处理消息存储到物理硬盘和查询功能。
HA Service：高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。
Index Service：根据特定的Message key对投递到Broker的消息进行索引服务，以提供消息的快速查询。
Broker是具体提供业务的服务器，单个Broker节点与所有的NameServer节点保持长连接及心跳，并会定时将Topic信息注册到NameServer，顺带一提底层的通信和连接都是基于Netty实现的。
Broker负责消息存储，以Topic为纬度支持轻量级的队列，单机可以支撑上万队列规模，支持消息推拉模型。
官网上有数据显示：具有上亿级消息堆积能力，同时可严格保证消息的有序性。
而从客户端的角度看主要有：Producer、Consumer两个部分。
2.1.3 Producer
消息生产者，负责产生消息，一般由业务系统负责产生消息。
Producer由用户进行分布式部署，消息由Producer通过多种负载均衡模式发送到Broker集群，发送低延时，支持快速失败。
RocketMQ 提供了三种方式发送消息：同步、异步和单向
- 同步发送：同步发送指消息发送方发出数据后会在收到接收方发回响应之后才发下一个数据包。一般用于重要通知消息，例如重要通知邮件、营销短信。
- 异步发送：异步发送指发送方发出数据后，不等接收方发回响应，接着发送下个数据包，一般用于可能链路耗时较长而对响应时间敏感的业务场景，例如用户视频上传后通知启动转码服务。
- 单向发送：单向发送是指只负责发送消息而不等待服务器回应且没有回调函数触发，适用于某些耗时非常短但对可靠性要求并不高的场景，例如日志收集。
2.1.4 Consumer
消息消费者，负责消费消息，一般是后台系统负责异步消费。
Consumer也由用户部署，支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。
Pull：拉取型消费者（Pull Consumer）主动从消息服务器拉取信息，只要批量拉取到消息，用户应用就会启动消费过程，所以 Pull 称为主动消费型。
Push：推送型消费者（Push Consumer）封装了消息的拉取、消费进度和其他的内部维护工作，将消息到达时执行的回调接口留给用户应用程序来实现。所以 Push 称为被动消费类型，但从实现上看还是从消息服务器中拉取消息，不同于 Pull 的是 Push 首先要注册消费监听器，当监听器处触发后才开始消费消息。
2.2 消息领域模型
2.2.1 Message
Message（消息）就是要传输的信息。
一条消息必须有一个主题（Topic），主题可以看做是你的信件要邮寄的地址。
一条消息也可以拥有一个可选的标签（Tag）和额处的键值对，它们可以用于设置一个业务 Key 并在 Broker 上查找此消息以便在开发期间查找问题。
2.2.2 Topic
Topic（主题）可以看做消息的规类，它是消息的第一级类型。比如一个电商系统可以分为：交易消息、物流消息等，一条消息必须有一个 Topic 。
Topic 与生产者和消费者的关系非常松散，一个 Topic 可以有0个、1个、多个生产者向其发送消息，一个生产者也可以同时向不同的 Topic 发送消息。一个 Topic 也可以被 0个、1个、多个消费者订阅。
2.2.3 Tag
Tag（标签）可以看作子主题，它是消息的第二级类型，用于为用户提供额外的灵活性。使用标签，同一业务模块不同目的的消息就可以用相同 Topic 而不同的 Tag 来标识。比如交易消息又可以分为：交易创建消息、交易完成消息等，一条消息可以没有 Tag 。
标签有助于保持您的代码干净和连贯，并且还可以为 RocketMQ 提供的查询系统提供帮助。
2.2.4 Group
分组，一个组可以订阅多个Topic。
分为ProducerGroup，ConsumerGroup，代表某一类的生产者和消费者，一般来说同一个服务可以作为Group，同一个Group一般来说发送和消费的消息都是一样的
2.2.5 Queue
在Kafka中叫Partition，每个Queue内部是有序的，在RocketMQ中分为读和写两种队列，一般来说读写队列数量一致，如果不一致就会出现很多问题。
Message Queue
Message Queue（消息队列），主题被划分为一个或多个子主题，即消息队列。
一个 Topic 下可以设置多个消息队列，发送消息时执行该消息的 Topic ，RocketMQ 会轮询该 Topic 下的所有队列将消息发出去。
一个Topic下可以有多个Queue，Queue的引入使得消息的存储可以分布式集群化，具有了水平扩展能力。
2.2.6 Offset
在RocketMQ 中，所有消息队列都是持久化，长度无限的数据结构，所谓长度无限是指队列中的每个存储单元都是定长，访问其中的存储单元使用Offset 来访问，Offset 为 java long 类型，64 位，理论上在 100年内不会溢出，所以认为是长度无限。
也可以认为 Message Queue 是一个长度无限的数组，Offset 就是下标。
2.3 消息消费模式
消息消费模式有两种：Clustering（集群消费）和Broadcasting（广播消费）。
默认情况下就是集群消费，该模式下一个消费者集群共同消费一个主题的多个队列，一个队列只会被一个消费者消费，如果某个消费者挂掉，分组内其它消费者会接替挂掉的消费者继续消费。
而广播消费消息会发给消费者组中的每一个消费者进行消费。
2.4 Message Order
Message Order（消息顺序）有两种：Orderly（顺序消费）和Concurrently（并行消费）。
顺序消费表示消息消费的顺序同生产者为每个消息队列发送的顺序一致，所以如果正在处理全局顺序是强制性的场景，需要确保使用的主题只有一个消息队列。
并行消费不再保证消息顺序，消费的最大并行数量受每个消费者客户端指定的线程池限制。
三、RocketMQ集群模式
编译源码包
1 | [admin@rocketmq1 ~]$ wget https://github.com/apache/rocketmq/archive/rocketmq-all-4.8.0.tar.gz |
编译后产物所在目录
1 | [admin@rocketmq1 rocketmq-rocketmq-all-4.8.0]$ ls distribution/target/rocketmq-4.8.0/rocketmq-4.8.0/ |
拷贝到工作目录下
1 | [admin@rocketmq2 rocketmq-4.8.0]$ cp -rf distribution/target/rocketmq-4.8.0/rocketmq-4.8.0/ ~/rocketmq |
安装jdk 1.8
1 | [admin@rocketmq1 ~]$ wget --no-cookies --no-check-certificate --header "Cookie: gpw_e24=http%3A%2F%2Fwww.oracle.com%2F; oraclelicense=accept-securebackup-cookie" "http://download.oracle.com/otn-pub/java/jdk/8u141-b15/336fa29ff2bb4ef291e347e091f7f4a7/jdk-8u141-linux-x64.tar.gz" |
3.1 单Master模式
这种方式风险较大，一旦Broker重启或者宕机时，会导致整个服务不可用。不建议线上环境使用,可以用于本地测试。
环境如下：
| 主机 | IP地址 | 组件划分 |
|---|---|---|
| rocketmq1 | 192.168.10.76 | Nameserver、Broker |
3.1.1 启动nameserver
1 | [admin@rocketmq1 ~]$ cd rocketmq-4.8.0/ |
3.1.2 启动broker
1 | [admin@rocketmq1 rocketmq-4.8.0]$ nohup bash bin/mqbroker -n 127.0.0.1:9876 autoCreateTopicEnable=true -c conf/broker.conf & |
内存不够的话可以修改bin/runserver.sh和bin/runbroker.sh适当降低内存限制。
3.1.3 看是否启动成功
通过jps命令查看是否启动成功
1 | [admin@rocketmq1 rocketmq-4.8.0]$ jps |
3.2 多Master模式
一个集群无Slave，全是Master，例如2个Master或者3个Master，这种模式的优缺点如下：
优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高；
缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。
环境如下：
| 主机 | IP地址 | 组件划分 |
|---|---|---|
| rocketmq1 | 192.168.10.76 | Nameserver1、Broker_Master-1 |
| rocketmq2 | 192.168.10.77 | Nameserver2、Broker_Master-2 |
3.2.1 创建目录修改配置
创建数据目录
1 | #在各服务器执行以下命令创建数据目录 |
修改日志路径
1 | #在各服务器执行以下命令修改日志存储路径 |
配置rocketmq
1 | #配置Broker_Master-1 |
1 | #配置Broker_Master-2 |
3.2.2 启动NameServer
NameServer需要先于Broker启动，且如果在生产环境使用，为了保证高可用，建议一般规模的集群启动3个NameServer，各节点的启动命令相同，如下：
1 | #启动Nameserver1 |
3.2.3 启动Broker
1 | #启动Broker_Master-1 |
3.2.4 关闭服务
关闭NameServer
1 | [admin@rocketmq1 rocketmq]$ bash bin/mqshutdown namesrv |
关闭Broker
1 | [admin@rocketmq1 rocketmq]$ bash bin/mqshutdown broker |
3.3 多Master多Slave模式-异步复制
每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式，主备有短暂消息延迟（毫秒级），这种模式的优缺点如下：
- 优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎一样；
- 缺点：Master宕机，磁盘损坏情况下会丢失少量消息。
环境如下：
| 主机 | IP地址 | 组件划分 |
|---|---|---|
| rocketmq1 | 192.168.10.76 | Nameserver1、Broker_Master-1 |
| rocketmq2 | 192.168.10.77 | Nameserver2、Broker_Master-2 |
| rocketmq3 | 192.168.10.78 | Nameserver3、Broker_Master_slave-1 |
| rocketmq4 | 192.168.10.79 | Nameserver4、Broker_Master_slave-2 |
3.3.1 创建目录修改配置
创建数据目录
1 | #在各服务器执行以下命令创建数据目录 |
修改日志路径
1 | #在各服务器执行以下命令修改日志存储路径 |
配置rocketmq
1 | #配置Broker_Master-1 |
1 | #配置Broker_Master-2 |
1 | #配置Broker_Master_slave-1 |
1 | #配置Broker_Master_slave-2 |
3.3.2 启动NameServer
1 | #启动Nameserver1 |
3.3.3 启动Broker
1 | #启动Broker_Master-1 |
3.3.4 关闭服务
关闭NameServer
1 | [admin@rocketmq1 rocketmq]$ bash bin/mqshutdown namesrv |
关闭Broker
1 | [admin@rocketmq1 rocketmq]$ bash bin/mqshutdown broker |
3.4 多Master多Slave模式-同步双写
每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功，这种模式的优缺点如下：
- 优点：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；
- 缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。
环境如下：
| 主机 | IP地址 | 组件划分 |
|---|---|---|
| rocketmq1 | 192.168.10.76 | Nameserver1、Broker_Master-1 |
| rocketmq2 | 192.168.10.77 | Nameserver2、Broker_Master-2 |
| rocketmq3 | 192.168.10.78 | Nameserver3、Broker_Master_slave-1 |
| rocketmq4 | 192.168.10.79 | Nameserver4、Broker_Master_slave-2 |
3.2.1 创建目录修改配置
创建数据目录
1 | #在各服务器执行以下命令创建数据目录 |
修改日志路径
1 | #在各服务器执行以下命令修改日志存储路径 |
配置rocketmq
1 | #配置Broker_Master-1 |
1 | #配置Broker_Master-2 |
1 | #配置Broker_Master_slave-1 |
1 | #配置Broker_Master_slave-2 |
3.4.2 启动NameServer
1 | #启动Nameserver1 |
3.4.3 启动Broker
1 | #启动Broker_Master-1 |
3.4.4 关闭服务
关闭NameServer
1 | [admin@rocketmq1 rocketmq]$ bash bin/mqshutdown namesrv |
关闭Broker
1 | [admin@rocketmq1 rocketmq]$ bash bin/mqshutdown broker |
3.5 Dledger 模式
RocketMQ 4.5 以前的版本大多都是采用 Master-Slave 架构来部署，能在一定程度上保证数据的不丢失，也能保证一定的可用性。
但是那种方式 的缺陷很明显，最大的问题就是当 Master Broker 挂了之后 ，没办法让 Slave Broker 自动 切换为新的 Master Broker，需要手动更改配置将 Slave Broker 设置为 Master Broker，以及重启机器，这个非常麻烦。
在手动运维的期间，可能会导致系统的不可用。
使用 Dledger 技术要求至少由三个 Broker 组成 ，一个 Master 和两个 Slave，这样三个 Broker 就可以组成一个 Group ，也就是三个 Broker 可以分组来运行。一但 Master 宕机，Dledger 就可以从剩下的两个 Broker 中选举一个 Master 继续对外提供服务。
3.5.1 DLedger 多副本即主从切换核心配置参数
| 配置名称 | 含义 |
|---|---|
| enableDLegerCommitLog | 是否启用 DLedger，即是否启用 RocketMQ 主从切换，默认值为 false。如果需要开启主从切换，则该值需要设置为 true。 |
| dLegerGroup | 节点所属的 raft 组，建议与 brokerName 保持一致，例如 broker-a。 |
| dLegerPeers | 集群节点信息，示例配置如下：n0-127.0.0.1:40911;n1-127.0.0.1:40912;n2-127.0.0.1:40913，多个节点用英文冒号隔开，单个条目遵循 legerSlefId-ip:端口，这里的端口用作 dledger 内部通信。 |
| dLegerSelfId | 当前节点id。取自 legerPeers 中条目的开头，即上述示例中的 n0，并且特别需要强调，只能第一个字符为英文，其他字符需要配置成数字。 |
| storePathRootDir | DLedger 日志文件的存储根目录，为了能够支持平滑升级，该值与 storePathCommitLog 设置为不同的目录 |
环境如下：
| 主机 | IP地址 | 组件划分 |
|---|---|---|
| rocketmq1 | 192.168.10.76 | Nameserver1、Broker_Master-1 |
| rocketmq2 | 192.168.10.77 | Nameserver2、Broker_Master_slave-1 |
| rocketmq3 | 192.168.10.78 | Nameserver3、Broker_Master_slave-2 |
3.5.2 创建目录修改配置
创建数据目录
1 | #在各服务器执行以下命令创建数据目录 |
修改日志路径
1 | #在各服务器执行以下命令修改日志存储路径 |
配置rocketmq
1 | #配置Broker_Master-1 |
1 | #配置Broker_Master_slave-1 |
1 | #配置Broker_Master_slave-1 |
3.5.3 启动NameServer
1 | #启动Nameserver1 |
3.5.4 启动Broker
1 | #启动Broker_Master-1 |
3.5.5 测试Master自动切换
通过3.6步安装rocketmq-console，在浏览器输入 http://IP:8080 访问控制台，显示集群状态如下：
通过上图可以看到，当前192.168.10.77是master节点。
接下来停止192.168.10.77上的broker服务
1 | [admin@rocketmq2 rocketmq]$ bash bin/mqshutdown broker |
再次查看群集状态
可以看到当前192.168.10.78被选举成了master
3.6 通过控制台连接rocketmq
3.6.1 下载rocketmq-console
1 | [admin@rocketmq1 ~]$ git clone https://github.com/apache/rocketmq-externals.git |
3.6.2 修改配置文件
修改pom.xml文件删掉如下plugins段内容，在234行附近
1 | [admin@rocketmq1 ~]$ cd rocketmq-externals/rocketmq-console/ |
修改连接rocketmq的地址，主要修改连接rocketmq的地址
1 | [admin@rocketmq1 rocketmq-console]$ vim src/main/resources/application.properties |
3.6.3 编译rocketmq-console
1 | [admin@rocketmq1 rocketmq-console]$ mvn clean package -Dmaven.test.skip=true |
3.6.4 启动rocketmq-console
1 | [admin@rocketmq1 rocketmq-console]$ java -jar target/rocketmq-console-ng-2.0.0.jar |
3.6.5 访问rocketmq-console控制台
在浏览器输入 http://IP:8080
3.7 rocketmq配置解析
| 配置名称 | 含义 |
|---|---|
| brokerClusterName | 所属集群名字 |
| brokerName | broker名字 |
| brokerId | 0 Master | >0 Slave |
| namesrvAddr | nameServer地址,分号分隔 |
| listenPort | Broker 对外服务的监听端口，默认10911 |
| defaultTopicQueueNums | 发送消息时,自动创建服务器不存在的topic,默认创建的队列数 |
| autoCreateTopicEnable | 是否允许 Broker 自动创建Topic, 通常线下开启，线上关闭 |
| autoCreateSubscriptionGroup | 是否允许 Broker 自动创建订阅组, 通常线下开启, 线上关闭 |
| listenPort | Broker 服务对外监听端口 |
| deleteWhen | 删除文件时间点，默认凌晨 4点 |
| fileReservedTime | 文件保留时间，默认 48 小时 |
| mapedFileSizeCommitLog | commitLog每个文件的大小默认1G |
| mapedFileSizeConsumeQueue | ConsumeQueue每个文件默认存30W条,根据业务情况调整 |
| destroyMapedFileIntervalForcibly | 强制删除文件间隔时间（单位毫秒） |
| redeleteHangedFileInterval | 定期检查Hanged文件间隔时间（单位毫秒） |
| diskMaxUsedSpaceRatio | 检测物理文件磁盘空间 |
| maxMessageSize | 限制的消息大小 |
| sendMessageThreadPoolNums | 发消息线程池数量 |
| pullMessageThreadPoolNums | 拉消息线程池数量 |
| brokerRole | Broker 的角色 [ASYNC_FLUSH 异步复制Master, SYNC_MASTER 同步双写Master],从节点为SLAVE |
| flushDiskType | 刷盘方式[ASYNC_FLUSH 异步刷盘, SYNC_FLUSH 同步刷盘] |
| storePathRootDir | 存储路径 |
| storePathCommitLog | 消费队列存储路径存储路径 |
| storePathConsumeQueue | 消费队列存储路径存储路径 |
| storePathIndex | 消息索引存储路径 |
| storeCheckpoint | checkpoint 文件存储路径 |
| abortFile | abort 文件存储路径 |