[系统运维]RocketMQ-RocketMQ 系统架构以及消息的概念

一、RocketMQ的消息模型

1、RocketMQ的基础消息模型，一个简单的Pub/Sub模型

上图就是一个基本的消息系统模型，包括生产者 (Producer)，消费者 (Consumer)，中间进行基于消息主题（Topic） 的消息传送。

在基于主题的系统中，消息被发布到主题或命名通道上。消费者将收到其订阅主题上的所有消息，生产者负责定义订阅者所订阅的消息类别。这是一个基础的概念模型，而在实际的应用中，结构会更复杂。例如为了支持高并发和水平扩展，中间的消息主题需要进行分区，同一个Topic会有多个生产者，同一个信息会有多个消费者，消费者之间要进行负载均衡等。

2、RocketMQ 扩展后的消息模型

上图就是一个扩展后的消息模型，包括两个生产者，两个消息Topic，以及两组消费者 Comsumer。

存储消息Topic的 代理服务器( Broker )，是实际部署过程对应的代理服务器。

• 为了消息写入能力的水平扩展，RocketMQ 对 Topic进行了分区，这种操作被称为队列（MessageQueue）。
• 为了消费能力的水平扩展，ConsumerGroup的概念应运而生。

3、RocketMQ的部署模型

Producer、Consumer又是如何找到Topic和Broker的地址呢？消息的具体发送和接收又是怎么进行的呢？当 当 当，RocketMQ的部署模型亮相啦～

在接下来 系统结构 模型中详细讲解RocketMQ的系统架构～

二、RocketMQ的系统架构

先为大家罗列出各个组件以及作用

• Producer：消息的发送者，如：寄件人
• Producer Group ：生产者组，生产者组是同一类生产者的集合，这类 Producer 发送相同 Topic 类型的消息。一个生产者组可以同时发送多个主题的消息。
• Consumer：消息的接受者，如：收件人
• Consumer Group：消费组，每一个 Consumer 实例都属于一个 Consumer Group。
  （注意：每一条消息只会被同一个 Consumer Group 里的一个 Consumer 实例消费。广播模式除外，同一个 ConsumerGroup 中的每个 Consumer 实例都处理全部的队列。不同的 Consumer Group 可以同时消费同一条消息）
• Broker：暂存和传输消息，如：快递公司
• NameServer：管理Broker，如：快递公司的管理机构
• Topic：区分消息的种类。
  • 一个发送者可以发送消息给一个或者多个Topic
  • 一个消息的接受者可以订阅一个或者多个Topic
• Message Queue：相当于 Topic 的分区，用于并行发送和接收消息

RocketMQ架构上主要由四部分构成：

1、Producer

消息生产者, 负责生产消息。 一般由业务系统负责生产消息。一个消息生产者会把业务应用系统里产生的消息发送到broker服务器。RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。

? 例如，业务系统产生的日志写入到 MQ 的过程，就是消息生产的过程。再如，电商平台中用户提交的秒杀请求写入到 MQ 的过程，就是消息生产的过程。

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

2、Consumer

? 消息消费者, 负责消费消息。一个消息消费者会从 Broker 服务器中获取到消息，并对消息进行相关业务处理 。一般是后台系统负责异步消费。从用户应用的角度而言提供了两种消费形式：拉取式消费、推动式消费。

• 支持以推（push），拉（pull）两种模式对消息进行消费。
• 同时也支持集群方式和广播方式的消费。
• 提供实时消息订阅机制，可以满足大多数用户的需求。

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

? 为了消费能力的水平扩展，RocketMQ 中的消息消费者都是以消费者组 ( Consumer Group ) 的形式出现的. 消费者组是同一类消费者的集合, 这类 Consumer 消费的是同一个 Topic 类型的消息。

相同的ConsumerGroup下的消费者主要有两种负载均衡模式，即广播模式，和集群模式（图中是最常用的集群模式）。

• 在集群模式下，消费者组使得在消费消息方面，实现 负载均衡 和 容错 的目标变得非常容易 。同一个 ConsumerGroup 中的 Consumer 实例是负载均衡消费，如图中 ConsumerGroupA 订阅 TopicA，TopicA 对应 3个队列，则 GroupA 中的 Consumer1 消费的是 MessageQueue 0和 MessageQueue 1的消息，Consumer2是消费的是MessageQueue2的消息。

  • 负载均衡：将一个 Topic 中的不同 Queue 平均分配给同一个 Consumer Group的不同的 Consumer， 注意：并不是将消息负载均衡。
  • 容错：一个 Consumer 挂了, 该 Consumer Group 中的其他 Consumer 可以接着消费 原Consumer 消费的Queue

• 在广播模式下，同一个 ConsumerGroup 中的每个 Consumer 实例都处理全部的队列。需要注意的是，广播模式下因为每个 Consumer 实例都需要处理全部的消息，因此这种模式仅推荐在通知推送、配置同步类小流量场景使用。

需要注意以下问题：

1. 消费者组在某一时刻只能消费一个 Topic 的消息，不能同时消费多个 Topic 消息；
2. 一个消费者组中的消费者必须订阅完全相同的 Topic；
3. 消费者组中 Consumer 的数量应该小于等于订阅 Topic 的 Queue 数量。如果超出 Queue 数量，则多出的Consumer 将不能消费消息。
   
4. 一个Topic类型的消息可以被多个消费者组同时消费。(当多个消费者组订阅同一个Topic时)
   

3、Name Server

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

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

其主要提供了以下两个功能：

• 管理 Broker ，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；
• 路由信息管理，每个NameServer将保存关于 Broker 集群的整个路由信息和用于客户端查询的队列信息。Producer和Consumer通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。

详细谈谈 Name Server 的功能吧～

3.1、路由注册

NameServer通常会有多个实例部署，不过， NameServer 是无状态的，即 NameServer 集群中的各个节点间是无差异的，各节点间相互不进行信息通讯。那么各节点是如何进行数据同步的呢？

• 当 Broker 节点启动时，会轮询 NameServer 列表，与每个 NameServer 节点建立长连接，并发起 注册请求。所以每一个NameServer实例上面都保存一份完整的路由信息。当某个NameServer因某种原因下线了，客户端仍然可以向其它NameServer获取路由信息。

• 在 NameServer 内部维护着?个 Broker 列表，用来动态存储 Broker 的信息。

注意，这是与其它像 zk 、 Eureka 、 Nacos 等注册中心不同的地方。那么NameServer这种无状态的方式，有什么优缺点呢？

• 优点：NameServer 集群搭建简单，扩容简单。
• 缺点：对于 Broker ，必须明确指出所有 NameServer 地址，否则未指出的将不会去注册。也正因为如此，NameServer 并不能随便扩容。因为，若 Broker 不重新配置，新增的 NameServer 对于 Broker 来说是不可见的，即其不会向这个 NameServer 进行注册。

Broker 节点是如何证明自己活着呢？

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

3.2、路由剔除

由于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 的移除。

3.3、路由发现

我们先来谈谈路由发现有哪几种模型？

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

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

3.4、Client 对 NameServer选择策略

这里的Client（客户端）指的是 Producer 与 Consumer

客户端在配置时必须要写上 NameServer 集群的地址，那么客户端到底连接的是哪个 NameServer 节点 呢？

? 客户端首先会生产一个随机数，然后再与NameServer 节点数量取模，此时得到的就是所要连接的节点索引，然后就会进行连接。如果连接失败，则会采用round-robin 策略，逐个尝试着去连 接其它节 点。即 : 首先采用的是 随机策略 进行的选择，失败后采用的是 轮询策略 。

4、Broker

4.1、Broker概述

功能介绍

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

模块构成

下图为 Broker Server 的功能模块示意图。

Remoting Moudle，整个 Broker 的实体, 负责处理来自 Client 端的请求. 而这个 Broker 实体则由以下模块构成

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

4.2、关于Broker的集群部署

NameServer几乎无状态节点，因此可集群部署，节点之间无任何信息同步。Broker部署相对复杂。

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

? 在 Master-Slave 架构中，Broker 分为 Master 与 Slave。一个Master可以对应多个Slave，但是一个Slave只能对应一个Master。Master 与 Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId 来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。

• Master 负责处理读写操作请求;
• Slave 负责对 Master 中的数据进行备份。

当 Master 挂掉了， Slave 则会自动切换为 Master 去工作。所以这个 Broker 集群是主备集群。

部署模型小结

• 每个 Broker 与 NameServer 集群中的所有节点建立长连接，定时注册 Topic 信息到所 NameServer。
• Producer 与 NameServer 集群中的其中一个节点建立长连接，定期从 NameServer 获取Topic路由信息，并向提供 Topic 服务的 Master 建立长连接，且定时向 Master 发送心跳。Producer 完全无状态。
• Consumer 与 NameServer 集群中的其中一个节点建立长连接，定期从 NameServer 获取 Topic 路由信息，并向提供 Topic 服务的 Master、Slave 建立长连接，且定时向 Master、Slave发送心跳。Consumer 既可以从 Master 订阅消息，也可以从Slave订阅消息。

梳理一下整个工作流程

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的存活状态。

4.3、创建 Topic

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

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

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

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

关于读写队列

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

1. 如：当写队列 小于 读队列时

   ? 创建 Topic 时设置的写队列数量为 4 ，读队列数量为 8 ，此时系统会创建 8 个 Queue ，分别是 0 1 2 3 4 5 6 7 。 Producer 会将消息写入到 0 1 2 3 这 四 个队列，但Consumer 只会消费 0 1 2 3 4 5 6 7 这 八 个队列中 的消息，但是 4 5 6 7 中是没有消息的。
   此时假设 Consumer Group 中包含两个 Consuer ， Consumer1 消 费 0 1 2 3 ，而 Consumer2 消费 4 5 6 7 。但实际情况是， Consumer2 是没有消息可消费的。

2. 如：当写队列 大于 读队列时

   ? 创建 Topic 时设置的写队列数量为 8 ，读队列数量为 4 ，此时系统会创建 8 个 Queue ，分别是 0 1 2 3 4 5 6 7 。 Producer 会将消息写入到这 8 个队列，但 Consumer 只会消费 0 1 2 3 这 四 个队列中的消息， 4 5 6 7 中的消息是不会被消费到的。

也就是说，当读 / 写队列数量设置不同时，总是有问题的。那么，为什么要这样设计呢？

其这样设计的目的是为了，方便 Topic 的 Queue 的缩容(将不需要的站点从原网络中移除的过程)。

如：原来创建的 Topic 中包含 16 个 Queue ，如何能够使其 Queue 缩容为 8 个，还不会丢失消息？

? 可以动态修改写队列数量为 8 ，读队列数量不变。此时新的消息只能写入到前 8 个队列，而消费都消费的却是 16 个队列中的数据。当发现后 8 个 Queue 中的消息消费完毕后，就可以再将读队列数量动态设置为 8 。整个缩容过程，没有丢失任何消息。

perm 用于设置对当前创建 Topic的操作权限：

• 2 表示只写 ;
• 4 表示只读 ;
• 6 表示读写 。

三、RocketMQ中的消息

首先什么是消息？

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

RocketMQ 消息构成非常简单，如下图所示。

• topic，表示要发送的消息的主题。
• body 表示消息的存储内容
• properties 表示消息属性
• transactionId 会在事务消息中使用。
• Tag: 不管是 RocketMQ 的 Tag 过滤还是延迟消息等都会利用 Properties 消息属性机制，这些特殊信息使用了系统保留的属性Key，设置自定义属性时需要避免和系统属性Key冲突。
• Keys: 服务器会根据 keys 创建哈希索引，设置后，可以在 Console 系统根据 Topic、Keys 来查询消息，由于是哈希索引，请尽可能保证 key 唯一，例如订单号，商品 Id 等。注意：使用过程中需要避免RocketMQ系统保留的属性key。

Message 可以设置的属性值包括：

1、Topic 主题

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

• Topic:message 1:n，主题与消息是一对多的关系。
• Message:Topic 1:1，消息与主题是一对一的关系。

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

• Producer:Topic 1:n，消息生产者与主题是一对多的关系。
• Consumer:Topic 1:1，消息消费者与主题是一对一的关系。
  

2、Tag 标签

Topic 与 Tag 都是业务上用来归类的标识，区别在于 Topic 是一级分类，而 Tag 可以理解为是二级分类。使用 Tag 可以实现对 Topic 中的消息进行过滤。

• Topic：消息主题，通过 Topic 对不同的业务消息进行分类。
• Tag：消息标签，用来进一步区分某个 Topic 下的消息分类，消息从生产者发出即带上的属性。

Topic 和 Tag 的关系如下图所示～

eg：一天黎明港口来了许多集装箱（消息），每个集装箱外都会有信息牌，我们可以根据信息知道集装箱中是什么货物（Topic）以及 货物来自于哪里（Tag）。

# Topic ：货物
tag= 上海
tag= 江苏
tag= 浙江
------- 消费者 ---0--
topic= 货物A tag = 上海
topic= 货物B tag = 上海 | 浙江
topic= 货物C tag = *

既然Topic和Tag都是业务上用来归类的标识，那我们什么时候该用Topic，什么时候该用 Tag？

我们可以从以下几个方面进行判断：

• 消息类型是否一致：如普通消息、事务消息、定时（延迟）消息、顺序消息，不同的消息类型使用不同的Topic，无法通过Tag进行区分。
• 业务是否相关联：没有直接关联的消息，如淘宝交易消息，京东物流消息使用不同的 Topic 进行区分；而同样是天猫消息，电气类订单、女装类订单、化妆品类订单的消息可以用 Tag 进行区分。
• 消息优先级是否一致：如同样是物流消息，河马必须小时内送达，天猫超市 24 小时内送达，淘宝物流则相对会慢一些，不同优先级的消息用不同的 Topic 进行区分。
• 消息量级是否相当：有些业务消息虽然量小但是实时性要求高，如果跟某些万亿量级的消息使用同一个 Topic，则有可能会因为过长的等待时间而“饿死”，此时需要将不同量级的消息进行拆分，使用不同的Topic。

总的来说，针对消息分类，您可以选择创建多个 Topic，或者在同一个 Topic 下创建多个 Tag。但通常情况下，不同的 Topic 之间的消息没有必然的联系，而 Tag 则用来区分同一个 Topic 下相互关联的消息，例如全集和子集的关系、流程先后的关系。

3、Queue 队列

为了支持高并发和水平扩展，需要对 Topic 进行分区，在 RocketMQ 中这被称为队列，一个 Topic 可能有多个队列，并且可能分布在不同的 Broker 上。

? 队列，是存储消息的物理实体。一个 Topic 中可以包含多个 Queue，每个 Queue 中存放的就是该 Topic 的消息. 一个 Topic 的 Queue 也被称为一个 Topic 中消息的分区 (Partrion) 。

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

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

一般来说一条消息，如果没有重复发送（比如因为服务端没有响应而进行重试），则只会存在在 Topic 的其中一个队列中，消息在队列中按照先进先出的原则存储，每条消息会有自己的位点，每个队列会统计当前消息的总条数，这个称为最大位点 MaxOffset；队列的起始位置对应的位置叫做起始位点 MinOffset。队列可以提升消息发送和消费的并发度。

4、MessageId/Key 消息标识

? RocketMQ 中每个消息拥有唯一的 MessageId ，且可以携带具有业务标识的 Key ，以方便对消息的查询。

? 不过需要注意的是， MessageId 有两个，当生产者在发送消息时会自动生成一个 MessageID，即 msgId。当消息到达 Broker 后，Broker 也会自动生成一个 MessageID，即 offsetMsgId。msgId 、offsetMsgId与key 都 称为消息标识。

• msgId，由 Producer 端生成，其生成规则为： producerIp + 进程pid + MessageClientIDServer 类的ClassLoader的hashCode + 当前时间 + AutomicInteger自增计数器；
• offserMsgId，由 Broker 端生成，其生成规则为：brokerIp + 物理分区的offset（Queue的偏移量）

Key，由用户制定的业务相关的唯一标识

? RocketMq 每个消息可以在业务层面的设置唯一标识，即 key 字段，方便将来定位消息丢失问题。应用可以通过 topic、key 来查询这条消息内容，以及消息被谁消费。

? 由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。

// 订单Id
String orderId = "20034568923546";
message.setKeys(orderId);