RocketMQ 架构简析

消费者类型：

1. 拉取式消费（Pull Consumer）Consumer 消费的一种类型，应用通常主动调用 Consumer 的拉消息方法从 Broker 服务器拉消息、主动权由应用控制。一旦获取了批量消息，应用就会启动消费过程。Pull 方式里，取消息的过程需要用户自己写（包括提交 offset 等操作）。
2. 推动式消费（Push Consumer）Consumer 消费的一种类型，该模式下 Broker 收到数据后会主动推送给消费端，该消费模式一般实时性较高。Push Consumer 原理上也是采取 pull 模式。实际上就是长轮询的 pull 模式。

一些概念

• 主题（Topic）表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是 RocketMQ 进行消息订阅的基本单位。每个 topic 可分为若干个分区（queue）。
• 生产者组（Producer Group）同一类 Producer 的集合，这类 Producer 发送同一类消息且发送逻辑一致。如果发送的是事务消息且原始生产者在发送之后崩溃，则 Broker 服务器会联系同一生产者组的其他生产者实例以提交或回溯消费。
• 消费者组（Consumer Group）同一类 Consumer 的集合，这类 Consumer 通常消费同一类消息且消费逻辑一致。消费者组使得在消息消费方面，实现负载均衡和容错的目标变得非常容易。要注意的是，消费者组的消费者实例必须订阅完全相同的 Topic。RocketMQ 支持两种消息模式：集群消费（Clustering）和广播消费（Broadcasting）。
• 普通顺序消息（Normal Ordered Message）普通顺序消费模式下，消费者通过同一个消费队列收到的消息是有顺序的，不同消息队列收到的消息则可能是无顺序的。
• 严格顺序消息（Strictly Ordered Message）严格顺序消息模式下，消费者收到的所有消息均是有顺序的。
• 消息（Message）消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。RocketMQ 中每个消息拥有唯一的 Message ID，且可以携带具有业务标识的 Key。系统提供了通过 Message ID 和 Key 查询消息的功能。
• 标签（Tag）为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化 RocketMQ 提供的查询系统。消费者可以根据 Tag 实现对不同子主题的不同消费逻辑，实现更好的扩展性。

关于消息中间件

消息中间件需要解决的问题：异步化、削峰填谷。

消息中间件应具备的基础能力是：消息发布、订阅、消费。概念相对简单这里不过多描述。

消息中间件的一些重要的机制：

1. 消息优先级（Message Priority；RocketMQ 不支持）

优先级是指在一个消息队列中，每条消息都有不同的优先级，一般用整数来描述，优先级高的消息先投递，如果消息完全在一个内存队列中，那么在投递前可以按照优先级排序，令优先级高的先投递。由于 RocketMQ 所有消息都是持久化的，所以如果按照优先级来排序，开销会非常大，因此 RocketMQ 没有特意支持消息优先级，但是可以通过变通的方式实现类似功能，即单独配置一个优先级高的队列，和一个普通优先级的队列，将不同优先级发送到不同队列即可。

2. 顺序消息（Message Order）

消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了 3 条消息，分别是订单创建，订单付款，订单完成。消费时，要按照这个顺序消费才能有意义。但是同时订单之间是可以并行消费的。RocketMQ 可以严格的保证消息有序。

• 投递消息的顺序性：投递消息的顺序性可通过将一组消息投递到同一分区实现。例如：借助 MessageQueueSelector 将对相同订单的操作消息投放到同一分区。
• 消费消息的顺序性：RoctetMQ 特性保障：特定分区（queue）中的消息不能同时被同一个消费者组中的多个 Consumer 消费，以避免重复消费。通过自定义或使用预置的 AllocateQueueStrategy 可设定分区的分配策略（哪些分区分配给哪个消费者消费）。

3. 高可用、消息可靠性

3.1 消息持久化

RocketMQ、Kafka 以文件记录形式持久化。

RocketMQ 采用了单一的日志文件，即把同 1 个 broker 上面所有 topic 的所有 queue 的消息，存放在一个文件里面，从而避免了随机的磁盘写入。

如上图所示，所有消息都存在一个单一的 CommitLog 文件里面，然后有后台线程异步的同步到 ConsumeQueue，再由 Consumer 进行消费。

TODO 同步、异步刷盘。

TODO RocketMQ 充分利用 Linux 文件系统内存 cache 来提高性能。TODO CommitLog index Commitlog segment 的大小与页缓存一致。

RocketMQ 消息存储机制会在后面的文章详细说明。

3.2 broker master/salve

TODO broker group master/salve

TODO Async/Sync Master；

4. 高并发、可扩展 ==> 分布式

提高并发效率 => 提高生产、消费并行度 => 提高分区数量。

RocketMQ、kafka 都支持 topic 数据分区存放、动态扩展。

以 RocketMQ 为例：

topic 创建的时候可以用集群模式去创建（这样集群里面每个 broker 的 queue 的数量相同），也可以用单个 broker 模式去创建（这样每个 broker 的 queue 数量可以不一致）。

4.1 生产并行度

RocketMQ 的生产并行度是由其自身机制及 broker 的数量决定的。这块后面的文章会详细分析。

4.2 消费并行度

广播模式下所有消费者会接受并消费当前 topic 下所有 Queue 的消息。

集群模式下，一个 queue 只分配给一个 consumer 实例：这是由于拉取消息是 consumer 主动控制的，如果多个实例同时消费一个 queue 的消息，会导致同一个消息在不同的实例下被消费多次，所以算法上都是一个 queue 只分给一个 consumer 实例，一个 consumer 实例可以允许同时分到不同的 queue。

Kafka 的消费并行度依赖 Topic 配置的分区数，如分区数为 10，那么最多 10 台机器来并行消费（每台机器只能开启一个线程），或者一台机器消费（10 个线程并行消费）。即消费并行度和分区数一致。RocketMQ 消费并行度分两种情况：顺序消费方式并行度同卡夫卡完全一致；乱序方式并行度取决于 Consumer 的线程数，如 Topic 配置 10 个队列，10 台机器消费，每台机器 100 个线程，那么并行度为 1000。

4.3 消息队列分配策略

Producer 使用 MessageQueueSelector 选择将消息投放到哪个分区 使用 AllocateMessageQueueStrategy 将不同分区分配给 Consumer Group 中的不同 Consumer。一个分区（queue）仅允许分配给同一个 Consumer Group 下的一个 Consumer（防止重复消费）。

MessageQueueSelector

内置实现类：SelectMessageQueueByMachineRoom SelectMessageQueueByHash SelectMessageQueueByRandom

可以通过实现 MessageQueueSelector 接口，来自定义 Producer 投递消息时选择分区的算法。

AllocateMessageQueueStrategy

内置实现类：

AllocateMessageQueueAveragely：平均分配算法 

AllocateMessageQueueAveragelyByCircle：基于环形平均分配算法

AllocateMachineRoomNearby：基于机房临近原则算法

AllocateMessageQueueByMachineRoom：基于机房分配算法

AllocateMessageQueueConsistentHash：基于一致性 hash 算法

AllocateMessageQueueByConfig：基于配置分配算法

可以通过实现 AllocateMessageQueueStrategy 来自定义 queue 分配给特定 Consumer Group 下不同 Consumer 的策略。

参考：

https://github.com/apache/rocketmq/blob/master/docs/cn/

https://juejin.im/post/6844903589819875336

https://jaskey.github.io/blog/2016/12/19/rocketmq-rebalance/

http://objcoding.com/2019/09/13/kafka-partition-and-rmq-queue/

http://www.itmuch.com/books/rocketmq

作者：RyanLee86799 来源：https://juejin.im/post/6844904130822029320

文章转载：JAVA 高级架构

（版权归原作者所有，侵删）