RocketMq之基础概念及架构解析

冷不防 2023-02-14 13:03 16阅读 0赞

# 一、RocketMQ的起源 #

通常，每个产品的诞生都源于一个具体的需求或问题，RocketMQ也不例外。起初，产品的原型像一个巨石，把所有需要实现的程序和接口都罗列到一起。但随着公司业务的发展，所有的系统和功能都在这个巨石上开发，当覆盖几百上千名开发人员的时候，瓶颈就出来了。这时候，就需要我们把系统进行分解。

![format_png][]

分解后，就出现了上图中的分布式架构，这类架构最大的特点就是解耦，而RocketMQ的异步解耦意味着底层的重构不会影响到上层应用的功能。RocketMQ另一个优势是削峰填谷，在面临流量的不确定性时，实现对流量的缓冲处理。此外，RocketMQ的顺序设计特性使得RocketMQ成为一个天然的排队引擎，例如，三个应用同时对一个后台引擎发起请求，排队引擎的特性可以确保不会引起“撞车”事故。

总结：RocketMQ的作用（消息中间件）：解耦、削峰填谷、高并发，或者说：异步消息处理、高性能（高并发读写）、高可用（主备）、可伸缩（削峰填谷）、最终一致性。

补充：

> 在2007年的时候，淘宝实施了“五彩石”项目，“五彩石”用于将交易系统从单机变成分布式，也是在这个过程中产生了阿里巴巴第一代消息引擎——Notify。

> 在2010年的时候，阿里巴巴B2B部门基于ActiveMQ的5.1版本也开发了自己的一款消息引擎，称为Napoli，这款消息引擎在B2B里面广泛地被使用，不仅仅是在交易领域，在很多的后台异步解耦等方面也得到了广泛的应用。

> 在2011年的时候，业界出现了现在被很多大数据领域所推崇的Kafka消息引擎，阿里巴巴在研究了Kafka的整体机制和架构设计之后，基于Kafka的设计使用Java进行了完全重写并推出了MetaQ 1.0版本，主要是用于解决顺序消息和海量堆积的问题。

> 在2012年，阿里巴巴对于MetaQ进行了架构重组升级，开发出了MetaQ 2.0，这时就发现MetaQ原本基于Kafka的架构在阿里巴巴如此庞大的体系下很难进行水平扩展，所以在2012年的时候就开发了RocketMQ 3.0版本。很多人会问到RocketMQ 3.0和MetaQ 3.0的区别，其实这两者是等价的版本，只不过阿里内部使用的称为MetaQ 3.0，外部开源称之为RocketMQ 3.0。

> 在2015年，又基于RocketMQ开发了阿里云上的Aliware MQ和Notify 3.0。

> 在2016年的时候，阿里巴巴将RocketMQ的内核引擎捐赠给了Apache基金会。

> 以上就是RocketMQ的整体发展历史，其实在阿里巴巴内部围绕着RocketMQ内核打造了三款产品，分别是MetaQ、Notify和Aliware MQ。这三者分别采用了不同的模型，MetaQ主要使用了**拉模型**，解决了顺序消息和海量堆积问题；Notify主要使用了**推模型**，解决了事务消息；而云产品Aliware MQ则是提供了商业化的版本。

# 二、RocketMQ的概念模型 #

对于任何一款中间件产品而言，清晰的概念模型是帮助开发者正确理解使用它的关键。从RocketMQ的概念模型来看：Topic是用于存储逻辑的地址的，Producer是信息的发送者，Consumer是信息的接收者。

![format_png 1][]

这只是一个基础的概念模型，在实际的生产中，结构会更复杂，例如我们需要对中间的Topic进行分区，出现多个有关联的Topic，再如同一个信息的发送方会有多个订阅者，同一个需求方会有多个发送方，出现一对多、多对一的情况。

![format_png 2][]

上图就是对Topic、Producer、Consumer扩展后的概念模型。RocketMQ中可以接触到的所有概念都可以在这个概念模型图中找到。

左边有两个Producer，中间就是两个分布式的Topic，用于存储逻辑地址的两个Topic中分别有两个用于存储物理存储地址的Message Queue，Broker是实际部署过程的对应的一台设备，右边则是两个Consumer，Consumer Group是代表两个Consumer可共享相互之间的订阅。不同的Consumer Group相互独立。

一句话总结就是不同的Group是广播订阅的，同一个Group则是负载订阅的。图中的连线表示各模块之间的关系，例如Consumer Group A中的Consumer1对应着Message Queue 0和Message Queue 1的两个队列，分布在BrokerA这一台设备上。

补充：

> Producer Group：  
> 用来表示一个发送消息应用，一个 Producer Group 下包含多个 Producer 实例，可以是多台机器，也可以 是一台机器的多个进程，或者一个进程的多个 Producer 对象。一个 Producer Group 可以发送多个 Topic 消息，Producer Group 作用如下：
> 
> 1.  标识一类 Producer
> 2.  可以通过运维工具查询返个发送消息应用下有多个 Producer 实例
> 3.  发送分布式事务消息时，如果 Producer 中途意外宕机，Broker 会主劢回调 Producer Group 内的任意一台机器来确认事务状态

> Consumer Group：  
> 用来表示一个消费消息应用，一个 Consumer Group 下包含多个 Consumer 实例，可以是多台机器，也可  
> 以是多个进程，或者是一个进程的多个 Consumer 对象。一个 Consumer Group 下的多个 Consumer 以均摊  
> 方式消费消息，如果设置为广播方式，那每个 Consumer Group 下的每个实例都消费全量数据。

> Topoic：消息的逻辑管理单位。

> Message Queue：消息的物理管理单位。一个Topic下可以有多个Queue，Queue的引入使得消息存储可以分布式集群化，具有了水平扩展的能力。

> 顺序消息：用户实现MessageQueueSelector为某一批消息（通常是有同样的唯一的标示ID），选择同一个Queue，则这一批消息的消费将是顺序消费（并由同一个consumer完成消费）。

> 事务消息：这样的消息有多个状态，并且其发送是两阶段的。第一个阶段发送PREPARED状态的消息，此时consumer是看不见这种状态的消息的，发送完毕后回调用户的TransactionExecutor接口，执行相应的事务操作（如数据库），当事务操作成功时，则对此条消息返回commit，让broker对该消息执行commit操作，成为commit状态的消息对consumer是可见的。

# 三、RocketMQ的存储模型 #

RocketMQ的消息的存储是由ConsumeQueue和CommitLog 配合来完成的，ConsumeQueue中只存储很少的数据，消息主体都是通过CommitLog来进行读写。

![format_png 3][]

CommitLog：

是消息主体以及元数据的存储主体，对CommitLog建立一个ConsumeQueue，每个ConsumeQueue对应一个（概念模型中的）MessageQueue，所以只要有Commit Log在，Consume Queue即使数据丢失，仍然可以恢复出来。消息存放的物理文件，每台broker上的commit log被本机所有的queue共享，不做任何区分。

文件的默认位置如下，仍然可通过配置文件修改：

${user.home}/store/${commitlog}/${fileName}

Commit Log的消息存储单元长度不固定，文件顺序写，随机读。消息的存储结构按照编号顺序以及编号对应的内容依次存储。

Consume Queue：

是一个消息的逻辑队列，存储了这个Queue在CommitLog中的起始offset，log大小和MessageTag的hashCode。每个Topic下的每个Queue都有一个对应的ConsumerQueue文件，例如Topic中有三个队列，每个队列中的消息索引都会有一个编号，编号从0开始，往上递增。并由此一个位点offset的概念，有了这个概念，就可以对Consumer端的消费情况进行队列定义。Consume queue是消息的逻辑队列，相当于字典的目录，用来指定消息在物理文件commit log上的位置。我们可以在配置中指定Consume queue与commit log存储的目录，每个topic下的每个queue都有一个对应的Consume queue文件，比如：

${rocketmq.home}/store/consumequeue/${topicName}/${queueId}/${fileName}

补充：

> RocketMQ的broker端，不负责推送消息，无论消费者是否消费消息，都将消息存储起来。谁要消费消息，就向broker发请求获取消息，消费记录由consumer来维护。RocketMQ提供了两种存储方式来保留消费记录：一种是保留在consumer所在的服务器上；另一种是保存在broker服务器上。用户还可以自己实现相应的消费进度存储接口。

> 默认情况下，采用集群消费（CLUSTERING），会将记录保存在broker端；而采用广播消费（BROADCASTING）则会将消费记录保存在本地。

> RocketMQ以Topic来管理不同应用的消息。对于生产者而言，发送消息是，需要指定消息的Topic，对于消费者而言，在启动后，需要订阅相应的Topic，然后可以消费相应的消息。Topic是逻辑上的概念，在物理实现上，一个Topic由多个Queue组成，采用多个Queue的好处是可以将Broker存储分布式化，提高系统性能。

> RocketMQ中，producer将消息发送给Broker时，需要制定发送到哪一个队列中，默认情况下，producer会轮询的将消息发送到每个队列中（所有broker下的Queue合并成一个List去轮询）。

> 对于consumer而言，会为每个consumer分配固定的队列（如果队列总数没有发生变化），consumer从固定的队列中去拉取没有消费的消息进行处理。

RocketMQ 存储特点：

> 零拷贝原理：Consumer 消费消息过程，使用了零拷贝，零拷贝包含以下两种方式：

> 1.  使用 mmap + write 方式

> 优点：即使频繁调用，使用小块文件传输，效率也很高

> 缺点：不能很好的利用 DMA 方式，会比 sendfile 多消耗CPU，内存安全性控制复杂，需要避免 JVM Crash 问题。

> 1.  使用 sendfile 方式

> 优点：可以利用 DMA 方式，消耗 CPU 较少，大块文件传输效率高，无内存安全新问题。

> 缺点：小块文件效率低亍 mmap 方式，只能是 BIO 方式传输，不能使用 NIO。

> RocketMQ 选择了第一种方式，mmap+write 方式，因为有小块数据传输的需求，效果会比 sendfile 更好。

RocketMQ 文件系统：

> RocketMQ 选择 Linux Ext4 文件系统。

> 原因如下：  
> Ext4 文件系统删除 1G 大小的文件通常耗时小亍 50ms，而 Ext3 文件系统耗时约 1s 左右，且删除文件时，磁盘 IO 压力极大，会导致 IO 写入超时。

> 文件系统局面需要做以下调优措施： 文件系统 IO 调度算法需要调整为 deadline，因为 deadline 算法在随机读情况下，可以合幵读请求为顺序跳跃  
> 方式，从而提高读 IO 吞吐量。

RocketMQ 数据存储结构：

![format_png 4][]

RocketMQ 存储目录结构

|-- abort 
    |-- checkpoint 
    |-- config
    | |-- consumerOffset.json
    | |-- consumerOffset.json.bak
    | |-- delayOffset.json
    | |-- delayOffset.json.bak
    | |-- subscriptionGroup.json.bak
    | |-- topics.json
    | |-- topics.json.bak 
    |-- commitlog
    | |-- 00000003384434229248
    | |-- 000000033855079710
    | |-- 0000000338658171289
    |-- consumequeue
      |-- %DLQ%ConsumerGroupA
      | |-- 0
      | | |-- 00000000000006000000
      |-- %RETRY%ConsumerGroupA
      | |-- 0
      | | |-- 00000000000000000000
      |-- %RETRY%ConsumerGroupB
      | |-- 0
      | | |-- 00000000000000000000
      |-- SCHEDULE_TOPIC_XXXX
      | |-- 2
      | | |-- 00000000000006000000
      | |-- 3
      | | |-- 00000000000006000000
      |-- TopicA
      | |-- 0
      | | |-- 00000000002604000000
      | | |-- 00000000002610000000
      | | |-- 00000000002616000000
      | |-- 1
      | | |-- 00000000002610000000
      | | |-- 00000000002610000000
      |-- TopicB
      | |-- 0
      | | |-- 00000000000732000000
      | |-- 1
      | | |-- 00000000000732000000
      | |-- 2
      | | |-- 00000000000732000000

# 四、RocketMQ的部署模型 #

在实际的部署过程中，Broker是实际存储消息的数据节点，Nameserver则是服务发现节点，Producer发送消息到某一个Topic，并给到某个Consumer用于消费的过程中，需要先请求Nameserver拿到这个Topic的路由信息，即Topic在哪些Broker上有，每个Broker上有哪些队列，拿到这些请求后再把消息发送到Broker中；相对的，Consumer在消费的时候，也会经历这个流程。

![format_png 5][]

补充：

NameServer：

> NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步（类似ZK）。

> NameServer用于存储Topic、Broker关系信息，功能简单，稳定性高。多个NameServer之间相互没有通信，单台NameServer宕机不影响其他NameServer与集群；即使整个NameServer集群宕机，已经正常工作的Producer，Consumer，Broker仍然能正常工作，但新起的Producer, Consumer，Broker就无法工作。

> NameServer压力不会太大，平时主要开销是在维持心跳和提供Topic-Broker的关系数据。但有一点需要注意，Broker向NameServer发心跳时，会带上当前自己所负责的所有Topic信息，如果Topic个数太多（万级别），会导致一次心跳中，就Topic的数据就几十M，网络情况差的话，网络传输失败，心跳失败，导致NameServer误认为Broker心跳失败。

Broker ：

> Broker 部署相对复杂，Broker分为Master 与 Slave，一个Master可以对应多个 Slave，但是一个Slave只能对应一个Master。

> Master 与 Slave 的对应关系通过指定相同的BrokerName，不同的 BrokerId 来定义，BrokerId为0表示Master，非 0 表示 Slave。

> Master可以部署多个。每个Broker与NameServer 集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。

> Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从 Name Server 取 Topic 路由信息，并向提供 Topic 服务的 Master 建立长连接，且定时向 Master 发送心跳。

Producer ：

> Producer 完全无状态，可集群部署。

> Producer启动时，也需要指定NameServer的地址，从NameServer集群中选一台建立长连接。如果该NameServer宕机，会自动连其他NameServer。直到有可用的NameServer为止。

> Producer每30秒从NameServer获取Topic跟Broker的映射关系，更新到本地内存中。再跟Topic涉及的所有Broker建立长连接，每隔30秒发一次心跳。在Broker端也会每10秒扫描一次当前注册的Producer，如果发现某个Producer超过2分钟都没有发心跳，则断开连接。

> Producer发送时，会自动轮询当前所有可发送的broker，一条消息发送成功，下次换另外一个broker发送，以达到消息平均落到所有的broker上。

> 假如某个Broker宕机，意味生产者最长需要30秒才能感知到。在这期间会向宕机的Broker发送消息。当一条消息发送到某个Broker失败后，会往该broker自动再重发2次，假如还是发送失败，则抛出发送失败异常。业务捕获异常，重新发送即可。客户端里会自动轮询另外一个Broker重新发送，这个对于用户是透明的。

Consumer：

> Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向 Master、Slave发送心跳。

> Consumer既可以从Master订阅消息，也可以从 Slave 订阅消息，订阅规则由 Broker 配置决定。

> Consumer启动时需要指定NameServer地址，与其中一个NameServer建立长连接。消费者每隔30秒从NameServer获取所有Topic的最新队列情况，这意味着某个Broker如果宕机，客户端最多要30秒才能感知。连接建立后，从NameServer中获取当前消费Topic所涉及的Broker，直连Broker。

> Consumer跟Broker是长连接，会每隔30秒发心跳信息到Broker。Broker端每10秒检查一次当前存活的Consumer，若发现某个Consumer 2分钟内没有心跳，就断开与该Consumer的连接，并且向该消费组的其他实例发送通知，触发该Consumer集群的负载均衡。

> Consumer有两种模式消费：集群消费，广播消费。

> 广播消费：每个消费者消费Topic下的所有队列。

> 集群消费：
> 
> 一个topic可以由同一个ID下所有消费者分担消费。具体例子：假如TopicA有6个队列，某个消费者ID起了2个消费者实例，那么每个消费者负责消费3个队列。如果再增加一个消费者ID相同消费者实例，即当前共有3个消费者同时消费6个队列，那每个消费者负责2个队列的消费。
> 
> 消费者端的负载均衡，就是集群消费模式下，同一个GROUP下的所有消费者实例平均消费该Topic的所有队列。

# 五、RocketMQ的刷盘策略 #

RocketMQ 的所有消息都是持久化的，先写入系统 PAGECACHE，然后刷盘，可以保证内存与磁盘都有一份数据，访问时，直接从内存读取。

异步刷盘（ASYNC\_FLUSH）默认是这种模式：

![format_png 6][]

返回成功状态时，消息只是被写入内存 pagecache，写操作返回快，吞吐量达，当内存里的消息积累到一定程度时，统一出发写磁盘动作，快速写入。

在有 RAID 卡，SAS 15000 转磁盘测试顺序写文件，速度可以达到 300M 每秒左右，而线上的网卡一般都为千兆网卡，写磁盘速度明显快于数据网络入口速度，那举是否可以做到写完内存就吐用户返回，由后台线程刷盘呢？

1.  由于磁盘速度大于网卡速度，那么刷盘的进度肯定可以跟上消息的写入速度。
2.  万一由于此时系统压力过大，可能堆积消息，除了写入 IO，还有读取 IO，万一出现磁盘读取落后情况，  
    会不会导致系统内存溢出，答案是否定的，原因如下：
    
    a) 写入消息到 pagecache 时，如果内存不足，则尝试丢弃干净的 page，腾出内存供新消息使用，策略是 LRU 方式。
    
    b) 如果干净页不足，此时写入 pagecache 会被阻塞，系统尝试刷盘部分数据，大约每次尝试 32 个 page，来找出更多干净 page。

综上，内存溢出的情况不会出现。

同步刷盘（SYNC\_FLUSH）：

![format_png 7][]

返回成功状态时，消息已经被写入磁盘。

消息写入内存 pagecache 后，立即通知刷盘线程，刷盘完成后，返回消息写成功的状态。

同步刷盘与异步刷盘的唯一区别是异步刷盘写完 pagecache 直接返回，而同步刷盘需要等待刷盘完成才返回，  
同步刷盘流程如下：

1.  写入 pagecache 后，线程等待，通知刷盘线程刷盘。
2.  刷盘线程刷盘后，唤醒前端等待线程，可能是一批线程。
3.  前端等待线程吐用户返回成功。

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNTM1NjM0My1lZjlmNDlkMTVmZjBjZDdiLnBuZz9pbWFnZU1vZ3IyL2F1dG8tb3JpZW50L3N0cmlwfGltYWdlVmlldzIvMi93Lzg4MQ?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNTM1NjM0My0wNDE0N2Q4YjAxMjY3ZTI4LnBuZz9pbWFnZU1vZ3IyL2F1dG8tb3JpZW50L3N0cmlwfGltYWdlVmlldzIvMi93LzcxNA?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNTM1NjM0My1iZmU5ZjVhMjBmZmFlODMwLnBuZz9pbWFnZU1vZ3IyL2F1dG8tb3JpZW50L3N0cmlwfGltYWdlVmlldzIvMi93LzEwMjE?x-oss-process=image/format,png
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNTM1NjM0My0xMzg2YWVmNTc4NWRmYTZlLnBuZz9pbWFnZU1vZ3IyL2F1dG8tb3JpZW50L3N0cmlwfGltYWdlVmlldzIvMi93LzkzOQ?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNTM1NjM0My0zMGE1OTAxMzNhYTViYTFmLnBuZz9pbWFnZU1vZ3IyL2F1dG8tb3JpZW50L3N0cmlwfGltYWdlVmlldzIvMi93LzkwNw?x-oss-process=image/format,png
[format_png 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNTM1NjM0My0wYWU0ZDM1NTY5YmFlYzRjLnBuZz9pbWFnZU1vZ3IyL2F1dG8tb3JpZW50L3N0cmlwfGltYWdlVmlldzIvMi93Lzg5NQ?x-oss-process=image/format,png
[format_png 6]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNTM1NjM0My1iMzEzODQ4ZGU4NzlmM2ZhLnBuZz9pbWFnZU1vZ3IyL2F1dG8tb3JpZW50L3N0cmlwfGltYWdlVmlldzIvMi93LzM5OA?x-oss-process=image/format,png
[format_png 7]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNTM1NjM0My0zMTVhNDZjZjc0MjM5MmI1LnBuZz9pbWFnZU1vZ3IyL2F1dG8tb3JpZW50L3N0cmlwfGltYWdlVmlldzIvMi93LzM4OQ?x-oss-process=image/format,png