【RockerMQ】002-RocketMQ 基本概念、系统架构

怼烎@ 2023-09-24 20:40 48阅读 0赞

## 【RockerMQ】002-RocketMQ 基本概念、系统架构 ##

#### 文章目录 ####

*  【RockerMQ】002-RocketMQ 基本概念、系统架构
 *  一、基本概念
 *   *  1、消息（Message）
     *  2、主题（Topic）
     *  3、标签（Tag）
     *  4、队列（Queue）
     *  5、消息标识（MessageId/Key）
 *  二、系统架构
 *   *  1、架构图
     *  2、Producer（生产者）
     *  3、Consumer（消费者）
     *  4、Name Server（注册中心）
     *   *  功能介绍
         *  主要包括两个功能
         *  路由注册
         *  路由剔除
         *  路由发现
         *  客户端 NameServer 选择策略
     *  5、Broker
     *   *  功能介绍
         *  模块构成
         *  集群部署
     *  6、工作流程
     *   *  具体流程
         *  Topic的创建模式
         *  读/写队列

## 一、基本概念 ##

### 1、消息（Message） ###

**消息**是指，消息系统所传输信息的**物理载体**，生产和消费数据的**最小单位**，每条消息必须属于一个**主题**。

### 2、主题（Topic） ###

![image-20230225164444070][]

Topic 表示**一类消息的集合**，每个主题包含若干条消息，每条消息只能属于一个主题，是 RocketMQ 进行**消息订阅**的基本单位。

一个生产者可以同时发送多种 Topic 的消息；而一个消费者只对某种特定的 Topic 感兴趣，即只可以订阅和消费一种 Topic 的消息。

> 消息的生产者相当于是发布命令的主体，一个主题的需求是多种多样的，满足需求的服务通常是专项的。比如一个人想要吃饭，那么一般就是外卖服务；想要到一个很远的地方，那么一般就需要交通服务……一个消息可以说是一个需求、命令、事件，负责完成它的是专项的服务。
> 
> 生产出一个主题的消息，需要能消费这个主题消息的消费者消费掉！

### 3、标签（Tag） ###

**为消息设置的标签，用于同一主题下区分不同类型**的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的**清晰度和连贯性**，并优化 RocketMQ 提供的查询系统。消费者可以根据 Tag 实现对不同子主题的不同消费逻辑，实现更好的扩展性。

> 标签是作用是什么？增强识别度！比如一个人会给其他人、其他不熟悉的地区贴上标签，贴标签的作用就是简化认知！

Topic是消息的一级分类，Tag是消息的二级分类。

*  Topic：货物
    
     *  tag=上海
     *  tag=江苏
     *  tag=浙江

消费者：

*  topic=货物 tag = 上海
 *  topic=货物 tag = 上海|浙江
 *  topic=货物 tag = \*

### 4、队列（Queue） ###

存储消息的**物理实体**。一个 Topic 中可以包含多个 Queue ，每个 Queue 中存放的就是该 Topic 的消息。一个 Topic 的 Queue 也被称为一个 Topic 中**消息的分区**（Partition）。

一个 Topic 的 Queue 中的消息只能被一个**消费者组**中的一个**消费者**消费。一个 Queue 中的消息不允许同一个消费者组中的多个消费者同时消费。

![image-20230225165527208][]

在学习参考其它相关资料时，还会看到一个概念：分片（Sharding）。\*\*分片不同于分区。\*\*在 RocketMQ 中，分片指的是存放相应 Topic 的 Broker。**每个分片中会创建出相应数量的分区，即 Queue，每个 Queue 的大小都是相同的。**

![img][]

### 5、消息标识（MessageId/Key） ###

RocketMQ 中每个消息拥有**唯一**的 MessageId ，且可以携带具有**业务标识**的 Key ，以方便对消息的**查询**。不过需要注意的是，MessageId 有两个：在生产者 send() 消息时会自动生成一个 MessageId（msgId)，当消息到达 Broker 后，Broker 也会自动生成一个MessageId(offsetMsgId)。msgId、offsetMsgId 与 key 都称为**消息标识**。

*  msgId：由 producer 端生成，其生成规则为：producerIp + 进程 pid + MessageClientIDSetter 类的 ClassLoader 的 hashCode + 当前时间 + AutomicInteger 自增计数器；
 *  offsetMsgId：由 broker 端生成，其生成规则为：brokerIp + 物理分区的 offset（Queue中的偏移量）；
 *  key：由**用户指定**的业务相关的唯一标识。

## 二、系统架构 ##

### 1、架构图 ###

> 不错文章：https://blog.csdn.net/wuyuxiu123/article/details/113063122

![img][img 1]

### 2、Producer（生产者） ###

**消息生产者，负责生产消息。Producer 通过 MQ 的负载均衡模块选择相应的 Broker 集群队列进行消息投递，投递的过程支持快速失败并且低延迟**。

> 例如，业务系统产生的日志写入到 MQ 的过程，就是消息生产的过程

> 再如，电商平台中用户提交的秒杀请求写入到MQ的过程，就是消息生产的过程

RocketMQ 中的消息生产者都是以**生产者组（Producer Group）的形式出现的。生产者组是同一类生产者的集合，这类 Producer 发送相同 Topic 类型的消息。一个生产者组**可以同时发送多个主题的消息。

> **消息的生产者，通过 NameServer 集群获得 Topic 的路由信息，包括 Topic 下面有哪些 Queue，这些 Queue 分布在哪些 Broker 上等。Producer 只会将消息发送到 Master 节点上，因此只需要与 Master 节点建立连接。**

### 3、Consumer（消费者） ###

\*\*消息消费者，负责消费消息。\*\*一个消息消费者会从 Broker 服务器中获取到消息，并对消息进行相关业务处理。

> 例如，QoS 系统从 MQ 中读取日志，并对日志进行解析处理的过程就是消息消费的过程。

> 再如，电商平台的业务系统从 MQ 中读取到秒杀请求，并对请求进行处理的过程就是消息消费的过程。

RocketMQ 中的消息消费者都是以**消费者组（Consumer Group）的形式出现的。消费者组是同一类消费者的集合，这类 Consumer 消费的是同一个 Topic 类型的消息。消费者组使得在消息消费方面，实现负载均衡**（将一个 Topic 中的**不同的 Queue** 平均分配给同一个Consumer Group 的不同的 Consumer，注意，**并不是将消息负载均衡**）和**容错**（一个 Consmer 挂了，该 Consumer Group 中的其它Consumer 可以接着消费原 Consumer 消费的 Queue ）的目标变得非常容易。

> 自：这个负载均衡是一种**表面的负载均衡**，因为无法确定每一个 Queue 里面的消息量是相同的。

![img][img 2]

**一个消费者可以消费多个 queue，一个 queue 只能被一个消费者消费**

消费者组中 Consumer 的数量应该小于等于订阅 Topic 的 Queue 数量。如果超出 Queue 数量，则多出的 Consumer 将不能消费消息。

![img][img 3]

**不过，一个 Topic 类型的消息可以被多个消费者组同时消费。**

> 注意，
> 
>  *  1 ）消费者组只能消费一个 Topic 的消息，不能同时消费多个 Topic 消息；
>  *  2 ）一个消费者组中的消费者必须订阅完全相同的 Topic 。

> **消息的消费者，通过 NameServer 集群获得 Topic 的路由信息，连接到对应的 Broker 上消费消息。注意，由于 Master 和 Slave 都可以读取消息，因此 Consumer 会与 Master 和 Slave 都建立连接。**

### 4、Name Server（注册中心） ###

#### 功能介绍 ####

NameServer 是一个 Broker 与 Topic 路由的**注册中心**，支持 Broker 的**动态注册与发现**。

RocketMQ 的思想来自于 Kafka，而 Kafka 是依赖了 Zookeeper 的。所以，在 RocketMQ 的早期版本，即在 MetaQ v1.0 与 v2.0 版本中，也是依赖于 Zookeeper 的。从 MetaQ v3.0，即 RocketMQ 开始去掉了 Zookeeper 依赖，使用了自己的 NameServer。

> Name Server 是专为 RocketMQ设计的**轻量级**名称服务，具有简单、可集群横吐扩展、无状态，节点之间互不通信等特点。

#### 主要包括两个功能 ####

*  `Broker管理：`接受 Broker 集群的**注册信息**并且保存下来作为**路由信息**的基本数据；提供**心跳检测机制**，检查 Broker 是否还存活。
 *  `路由信息管理：`每个 NameServer 中都保存着 Broker 集群的整个**路由信息**和用于客户端查询的**队列信息**。Producer 和 Conumser 通过 NameServer 可以获取整个 Broker 集群的**路由信息**，从而进行消息的**投递和消费**。

#### 路由注册 ####

NameServer 通常也是以**集群**的方式部署，不过，NameServer 是**无状态**的，即 NameServer 集群中的**各个节点间是无差异的**，各节点间相互**不进行信息通讯**。那各节点中的数据是如何进行数据同步的呢？在 Broker 节点启动时，轮询 NameServer 列表，与每个NameServer 节点建立**长连接**，发起**注册请求**。在 NameServer 内部维护着一个 Broker 列表，用来动态存储 Broker 的信息。

> 注意，这是与其它像zk、Eureka、Nacos等注册中心不同的地方。
> 
> 这种 NameServer 的**无状态**方式，有什么优缺点：
> 
> 优点：NameServer 集群搭建简单，扩容简单。
> 
> 缺点：对于 Broker，必须明确指出所有 NameServer 地址。否则未指出的将不会去注册。也正因为如此，NameServer 并不能随便扩容。因为，若 Broker 不重新配置，新增的 NameServer 对于 Broker 来说是**不可见**的，其不会向这个 NameServer 进行注册。

Broker 节点**为了证明自己是活着的**，为了维护与 NameServer 间的长连接，会将最新的信息以**心跳包**的方式上报给 NameServer，**每 30 秒**发送一次心跳。心跳包中包含 BrokerId、Broker地址(IP+Port)、Broker 名称、Broker 所属**集群名称**等等。NameServer 在接收到心跳包后，会更新心跳时间戳，记录这个 Broker 的最新存活时间。

#### 路由剔除 ####

由于 Broker 关机、宕机或网络抖动等原因，NameServer 没有收到 Broker 的心跳，NameServer 可能会将其从 Broker 列表中剔除。

NameServer 中有一个**定时任务**，**每隔 10 秒**就会扫描一次 Broker 表，查看每一个 Broker 的最新心跳时间戳距离当前时间**是否超过 120 秒**，如果超过，则会判定 Broker 失效，然后将其从 Broker 列表中剔除。

> 扩展：对于 RocketMQ 日常运维工作，例如 Broker 升级，需要停掉 Broker 的工作。OP 需要怎么做？  
> OP 需要将 Broker 的读写权限禁掉。一旦 client(Consumer或Producer) 向 broker 发送请求，都会收到 broker 的 NO\_PERMISSION 响应，然后 client 会进行对其它 Broker 的重试。  
> 当 OP 观察到这个 Broker 没有流量后，再关闭它，实现 Broker 从 NameServer 的移除。  
> OP：运维工程师  
> SRE：Site Reliability Engineer，现场可靠性工程师

#### 路由发现 ####

RocketMQ 的路由发现采用的是 Pull 模型。当 Topic 路由信息出现变化时，NameServer **不会主动推送**给客户端，而是**客户端定时拉取**主题最新的路由。默认客户端**每 30 秒**会拉取一次最新的路由。

> 扩展：  
> 1 ）Push 模型：**推送模型**。其**实时性**较好，是一个“发布-订阅”模型，需要维护一个长连接。而长连接的维护是需要资源成本的。该模型适合于的场景：  
> \* 实时性要求较高  
> \* Client数量不多，Server数据变化较频繁  
> 2 ）Pull 模型：**拉取模型**。存在的问题是，实时性较差。  
> 3 ）Long Polling 模型：**长轮询模型**。其是对Push与Pull模型的整合，充分利用了这两种模型的优势，屏蔽了它们的劣势。

#### 客户端 NameServer 选择策略 ####

> 这里的客户端指的是 Producer 与 Consumer

客户端在配置时必须要写上 NameServer **集群的地址**，那么客户端到底连接的是哪个 NameServer 节点呢？客户端首先会生产一个**随机数**，然后再与 NameServer **节点数量取模**，此时得到的就是所要连接的**节点索引**，然后就会进行连接。如果连接失败，则会采用 round-robin 策略，**逐个尝试**着去连接其它节点。

首先采用的是`随机策略`进行的选择，失败后采用的是`轮询策略`。

> **随机 + 轮询！**

> 扩展：Zookeeper Client 是如何选择 Zookeeper Server 的？  
> 简单来说就是，经过两次 Shuffle，然后选择第一台 Zookeeper Server。  
> 详细说就是，将配置文件中的 zk server 地址进行第一次 Shuffle，然后随机选择一个。这个选择出的一般都是一个 hostname。然后获取到该 hostname 对应的所有 ip，再对这些 ip 进行第二次 Shuffle，从 Shuffle过的结果中取第一个 server 地址进行连接。Shuffle：打散、搅乱。

### 5、Broker ###

#### 功能介绍 ####

Broker 充当着**消息中转**角色，负责**存储消息、转发**消息。Broker 在 RocketMQ 系统中负责**接收并存储**从生产者发送来的消息，同时**为消费者的拉取请求作准备**。Broker 同时也存储着消息相关的**元数据**，包括消费者组消费进度偏移 offset、主题、队列等。

**Broker是 RocketMQ 的核心，大部分‘重量级”工作都是由 Broker完成的。包括接收 Producer 发过来的消息、处理 Consumer 的消费消息请求、消息的持久化存储、消息的 HA 机制以及服务端过滤功能等 。**

#### 模块构成 ####

![image-20230226193513867][]

`Remoting Module`：整个 Broker 的**实体**，负责处理来自 clients 端的请求。而这个 Broker 实体则由以下模块构成。

*  `Client Manager：`**客户端管理器**。负责**接收、解析**客户端(Producer/Consumer)请求，\*\*管理客户端。\*\*例如，维护Consumer的Topic订阅信息
 *  `Store Service：`**存储服务**。提供方便简单的 API 接口，处理消息**存储到物理硬盘和消息查询**功能。
 *  `HA Service：`**高可用服务**，提供 Master Broker 和 Slave Broker 之间的**数据同步功能**。
 *  `Index Service：`**索引服务**。根据特定的 Message key，对投递到 Broker 的消息进行索引服务，同时也提供根据 Message Key 对消息进行**快速查询**的功能。

#### 集群部署 ####

为了增强 Broker **性能与吞吐量**，**Broker 一般都是以集群形式出现的**。各集群节点中可能存放着**相同 Topic 的不同 Queue**。不过，这里有个问题，如果某 Broker 节点宕机，如何保证数据不丢失呢？其解决方案是，将每个 Broker 集群节点进行**横向扩展**，即将 Broker节点 再建为一个 HA 集群，解决单点问题。

Broker 节点集群是一个**主备集群**，即集群中具有 Master 与 Slave 两种角色。Master 负责处理**读写操作请求**，Slave 负责对 Master 中的数据进行**备份**(平常操作都是操作 Master )。当 Master 挂掉了，Slave 则会**自动切换为 Master** 去工作。所以这个 Broker 集群是**主备集群**。一个 Master 可以包含多个 Slave，但一个 Slave 只能隶属于一个 Master。Master 与 Slave 的对应关系是通过指定相同的BrokerName、不同的 BrokerId 来确定的。BrokerId 为 0 表示 Master，非 0 表示 Slave。每个 Broker 与 NameServer 集群中的**所有节点**建立长连接，定时注册 Topic 信息到所有 NameServer。  
![img][img 4]

### 6、工作流程 ###

#### 具体流程 ####

*  1 ）启动 NameServer，NameServer 启动后开始**监听端口**，等待 Broker、Producer、Consumer 连接。
 *  2 ）启动 Broker 时，Broker 会与**所有的 NameServer** 建立并保持长连接，然后每 30 秒向 NameServer 定时发送心跳包。
 *  3 ）发送消息前，可以先**创建 Topic**，创建 Topic 时需要指定该 Topic 要存储在哪些 Broker 上，当然，在创建 Topic 时也会将 Topic 与 Broker 的关系写入到 NameServer 中。不过，这步是可选的，也可以在发送消息时自动创建 Topic。
 *  4 ）Producer 发送消息，启动时先跟 NameServer 集群中的其中一台建立**长连接**，并从 NameServer 中获取路由信息，即当前发送的 Topic 消息的 Queue 与 Broker 的地址（IP+Port）的**映射关系**。然后根据算法策略从队选择一个 Queue ，与队列所在的 Broker建立长连接从而向 Broker 发消息。当然，在获取到路由信息后，Producer 会首先将路由信息**缓存到本地**，再每 30 秒从NameServer **更新**一次路由信息。
 *  5 ）Consumer 跟 Producer 类似，跟**其中一台 NameServer** 建立**长连接**，获取其所订阅 Topic 的路由信息，然后根据算法策略从路由信息中获取到其所要消费的 Queue，然后直接**跟 Broker 建立长连接**，开始消费其中的消息。Consumer 在获取到路由信息后，同样也会每 30 秒从 NameServer 更新一次路由信息。不过不同于 Producer 的是，**Consumer 还会向 Broker 发送心跳**，以确保 **Broker 的存活状态**。

#### Topic的创建模式 ####

手动创建 Topic 时，有两种模式：

**集群模式：** 该模式下创建的 Topic 在该集群中，**所有 Broker 中的 Queue 数量是相同的**。(全局定义)

**Broker模式：** 该模式下创建的Topic在该集群中，每个Broker中的Queue数量可以不同。

自动创建 Topic 时，默认采用的是 Broker 模式，会**为每个 Broker 默认创建 4 个 Queue**。

#### 读/写队列 ####

\*\*从物理上来讲，读/写队列是同一个队列。\*\*所以，不存在读/写队列数据同步问题。读/写队列是逻辑上进行区分的概念。**一般情况下，读/写队列数量是相同的。**

> 例如，创建 Topic 时设置的**写队列数量为 8 ，读队列数量为 4** ，此时系统会创建 8 个 Queue，分别是 0 1 2 3 4 5 6 7。Producer 会将消息写入到这 8 个队列，但 Consumer 只会消费0 1 2 3 这 4 个队列中的消息，4 5 6 7 中的消息是不会被消费到的。
> 
> 再如，创建 Topic 时设置的写队列数量为 4 ，读队列数量为 8 ，此时系统会创建 8 个 Queue，分别是 0 1 2 3 4 5 6 7。Producer 会将消息写入到 0 1 2 3 这 4 个队列，但 Consumer 会消费0 1 2 3 4 5 6 7 这 8 个队列中的消息，但是 4 5 6 7 中是没有消息的。此时假设Consumer Group 中包含两个 Consumer，Consumer1 消费 0 1 2 3，而 Consumer2 消费 4 5 6 7。但实际情况是，Consumer2 是没有消息可消费的。
> 
> 也就是说，\*\*当读/写队列数量设置不同时，总是有问题的。\*\*那么，为什么要这样设计呢？
> 
> 其这样设计的目的是为了，方便 Topic 的 Queue 的**缩容**。
> 
> 例如，原来创建的 Topic 中包含 16 个 Queue，如何能够使其 Queue 缩容为 8 个，还不会丢失消息？可以动态修改写队列数量为 8 ，读队列数量不变。此时新的消息只能写入到前 8 个队列，而消费都消费的却是 16 个队列中的数据。当发现后 8 个 Queue 中的消息消费完毕后，就可以再将读队列数量动态设置为 8 。**整个缩容过程，没有丢失任何消息。**
> 
> **perm 用于设置对当前创建 Topic 的操作权限： 2 表示只写， 4 表示只读， 6 表示读写。**

[image-20230225164444070]: https://img-blog.csdnimg.cn/img_convert/85ed7946a61e9432c3fc3397c438aac3.png
[image-20230225165527208]: https://img-blog.csdnimg.cn/img_convert/64e7cb35759d2557afa81216bd642586.png
[img]: https://img-blog.csdnimg.cn/img_convert/899b0edc6b3e7430a1722ab9470e80ed.png
[img 1]: https://img-blog.csdnimg.cn/img_convert/7892ffa03bd4ce0f16609f15f596cffd.png
[img 2]: https://img-blog.csdnimg.cn/img_convert/75cb1915308581c830debecbe96f1065.png
[img 3]: https://img-blog.csdnimg.cn/img_convert/2c38c7bd63fe87254851ce464f26d3b2.png
[image-20230226193513867]: https://img-blog.csdnimg.cn/img_convert/8e46932589ff2074912dbb0bb57b6290.png
[img 4]: https://img-blog.csdnimg.cn/img_convert/ba7eca5b5c10c2812f03df28b8d52137.png