大数据生态中的 RocketMQ 5.0

本文作者：李伟 - Apache RocketMQ Committer，RocketMQ Python 客户端项目 Owner ，Apache Doris Contributor，腾讯云消息队列资深开发工程师，著有《RocketMQ 分布式消息中间件(核心原理与最佳实践)》。

一、RocketMqueue101

RocketMQ 拥有诸多出色的特性：

比如多副本机制，RocketMQ 支持存储层的多副本 Dledger，它是基于 Raft 协议的一致性存储库，保证能够从存储层实现多副本；

比如 ACL 鉴权机制，用于确定哪些 producer 能生产、哪些消费者组能消费，以及服务端的消息过滤；

比如事务消息，它是 RocketMQ 实现的生产者事务，生产者向 broker 发送一条事务消息，由生产者执行本地事务。如果执行成功，则向 broker 端发送 commit 事件，消费者才能消费；如果本地事务处理失败，则发送 rollback 事件，使消费者无法消费该消息。

比如 Request-Reply ，它是类似于同步 RPC 调用的过程，用户相同的逻辑用消息来实现，能够实现同步 RPC 调用的过程，可以将调用 API 和发送消息两套逻辑进行统一。

广播消息指消息发出后，订阅它的所有消费者都能消费到所有实例。负载均衡消费指默认策略下，同一个消费者组的消费者都能平均地消费消息，具体策略可自行调整。RocketMQ 支持 Pull、Push 和 Pop 三种消费模式，支持 java、go、cpp、python、c#等多种语言。

搭建 RocketMQ 集群的流程如下：

第一步：安装 NameServer 集群。NameServer 集群包含一个或多个 NameServer 节点。启动服务时，默认监听 9876 端口。NameServer 集群搭建好之后，启动一套 Broker 集群。

第二步：搭建 Broker 集群，使用经典 master-slave 部署模式， master 提供读写，同时会将数据存储和元数据同步一份到 slave 。通过 10912 的 HA 端口做数据同步。

第三步：写生产者代码生产。生产者集群包含多个生产者实例，通过 broker 的 10911 和 10909 端口向 broker 发送数据。

第四步：消费者通过 10911 或 10909 端口向 broker 拉取数据。

生产者或消费者实例启动时，会先配置 NameServer 地址，由生产者或消费者从 NameServer 集群上拉取 topic、Queue 和 Broker 等路由信息，然后根据路由信息发送或拉取消息。

生产者和消费者均与 broker 之间存在 channel 连接。如果生产者或消费者长时间没有与 broker 联系，则 broker 会将连接剔除。

以下为 RocketMQ101 相关名词解析：

生产者包含生产者组和生产者实例。生产者组是若干个生产者实例的组合，且 RocketMQ 希望同一个生产者组内的实例行为一致。消费者组和消费者实例也同理。行为一致指生产者实例都生产同一种类型的消息，比如都生产订单消息，包括创建订单、订单发货、订单删除等步骤。行为一致的好处在于消息的生产和消费比较规整，不会出现混乱。

Topic 是消息的分类，为字符串形式，可以通过 topic 将某集群内的全部消息进行分类，所有 topic 的消息组成全量的消息。而 Tag 又属于 topic 的子分类。

消费者在订阅消息时，必须先指定 topic 再指定 tag ，这样的一条记录被称为订阅关系。如果订阅关系不一致，则会导致订阅混乱，发生重复消费或不消费、消息堆积等情况。

Queue 类似于分区，但它是逻辑上的概念，并不是物理存储上的概念。Property 类似于 header，property 包含除了主要信息以外的扩展信息，比如消息属于哪个业务 ID、发送者 IP 等。向某个 topic 发送消息时，能够指定 property 。

NameServer 中包含 broker 与 cluster 的关系、Queue\ topic 与 broker 的关系，即路由信息。

Broker 中包含以下四部分：

①　CommitLog——常规的文件存储。RocketMQ 发送的数据会 append 到 CommitLog。

②　Consumer queue——消费者在消费 topic 时，topic 中包含多个 queue，每一个 queue 都被称为 consumer queue，每个消费者对于每个 consumer queue 都存在消费进度。

③　index——在 dashboard 上能够根据 key 来查询消息。

④　Dledger commitlog——由 Dledger 存储库来管理的 CommitLog，能够实现多副本。

RocketMQ 的生产消费模型十分简单。如上图，Topic A 有四个 queue，其中 queue1、queue2 在 Master Broker 1 上，queue3、queue4 在 Master Broker 2 上。ProducerGroup A 下有两个生产者实例，分别向两个 broker 的 queue 发送消息。Consumer Group A 也有两个消费者 consume 1 和 consume 2。

从四个 queue 里取消息时，每个消费者默认的策略是依次向 queue1、queue2、queue3、queue4 循环发消息，以此最大程度地保证消息分布均匀。

消费者的消费模式有负载均衡和广播消费消费两种。

负载均衡策略下，比如共有 4 条 queue，则 consumer instant1 和 consumer instant2 会分别被分配到 2 个 queue，具体分配到哪两条需由算法决定。

广播消费策略下，假设 topic 有 100 条消息，则 consumer instance 1 和 consumer instance 2 每一个消费者实例都会消费到 100 条消息，即同消费者组的每个消费者示例都会消费到全量的消息。

二、RocketMQ 生态项目

RocketMQ 生态项目包含以下几个部分：

客户端：客户端主要分为 Java 客户端与非 Java 客户端，其中 RocketMQ Java 客户端是最原生的客户端，与 RocketMQ 的编写语言一致，功能也最为齐全。

计算：RocketMQ 支持轻量级的预计算，比如轻量级的 ETL。RocketMQ-Flink 能够直接对接 Flink，方便将 RocketMQ 数据传输到 Flink 做计算，利用 Flink 强大的生态同步到下游多种类型的目的地。

RocketMQ-Connect 与 RocketMQ-Streams 是轻量级的计算框架，功能更简单、轻量，部署运维也更容易。

管控：RocketMQ-Dashboard 拥有简单稳定且功能强大的管控端，能够支持常用的运维操作比如修改配置、禁用消费者等。

云原生：RocketMQ-Docker 支持打包 RocketMQ 源码成为 Docker image 项目，能够支持各种不同平台的打包。RocketMQ-Operator 支持 RocketMQ 上 K8s，能够支持比如重启进程、下发配置、拉起集群等操作。

监控：RocketMQ-Exporter 目前能够支持 80+指标，可直接导入到 Prometheus 做告警和监控。开源项目可通过 Prometheus 的数据配置 Grafana 做大盘，实现监控能力。此外，Prometheus 能够支持 Hook 回调，方便公司用户将 RocketMQ 指标监控对接到自己的告警平台。

云原生是技术行业的趋势，能够减少成本、方便运维和管理。RocketMQ 新版本实现了存储计算分离，支持更快速、更方便地上 K8s 。EDA 事件驱动和无服务也是大势所趋，比如腾讯云的云函数、阿里云的 eventbridge 等产品都是 Serverless、EDA 场景，能够直接集成 RocketMQ。流计算方面，RocketMQ 能够支持将数据对接到 Flink 等，方便用户做轻量级的 ETL 等计算并数据推到下游。微服务领域，RocketMQ 也提供了诸多原生支持。

电商、金融等传统领域正在进行数字化转型，消息传递、指标、日志传递等需求都能够利用 Rocket MQ 简单快速地实现。

总而言之，RocketMQ 能够利用自己强大的生态项目，支持企业各种各样形态的数据传输和计算。

三、RocketMQ 数据流构建

RocketMQ 的数据流构建主要包含消息、CDC 数据流、监控数据流以及湖仓数据流。CDC 数据主要负责记录记录数据变更，监控数据流包括业务监控和常规监控。

消息的构建如上图所示。

以订单服务为例，订单服务收到创建订单的请求，创建成功后会将订单的基本信息通过 RocketMQ 发送给 B 服务。假设 B 服务为短信服务消费，由 B 服务向客户发送短信通知，包含订单相关的详细信息。

RocketMQ 发送消息至 B 服务时，通过重试和死信实现最终一致性，以保证消息能够成功发送给消费者。RocketMQ 有 16 次重试机会，且为阶梯性重试，能够持续十几个小时。

RocketMQ 支持通过 Canal/Flink CDC、RocketMQ-collect 的方式，将 Binlog 等数据提供给计算平台，再由 RocketMQ Flink、RocketMQ Streams 等进行轻量级的计算。计算完成后，将结果转发给下游数据库比如 MySQL、ES、Redis 等，进行异构或同构的数据同步。

RocketMQ 支持从 flume 读取日志文件发送至 RocketMQ，再通过 RocketMQ Collect 或 RocketMQ Flink 等将日志数据进行消费、ETL 转换或发送至 ES 。 ES 已与 ELK 产品打通，可以在 Kabana 上查看日志。

除了日志，RocketMQ 能够在业务系统做后端监控埋点，通过 RocketMQ client 将监控埋点数据发到 RocketMQ，再通过 RocketMQ Flink 或 RocketMQ Streams 消费数据并发送给业务监控平台或数据湖仓库等，生成在线报表或实时报表。

前端监控大部分通过 HTTP 请求发送至 RocketMQ，再通过 RocketMQ 相关的轻量级计算框架，根据不同诉求将数据汇总至不同的后端，比如 ES 或自建平台。

所有数据都能入到湖仓，因为所有数据都会有数据分析、数据挖掘或出统计报表的诉求。比如前后端的监控、 TP 数据库里的业务数据、日志文件的指标数据或日志文件都能通过对应的工具发到 RocketMQ，通过 RocketMQ 提供的轻量级计算工具进行计算，然后发送到下游的 Hive、Doris、Clickhouse 或 Hudi 等数据库或数据仓库，产出报表、实时大盘、实时数据表等。

RocketMQ 能够采集各种数据，比如 metrics、TP 数据、log 的数据，然后通过 RocketMQ 提供的轻量级计算工具进行计算，最终汇总到同构/异构的数据库、数据仓库或数据湖等。

数据构建流程中，RocketMQ 作为中间核心的传输链路，是否能够借助本身的特性避免偶然性的因素影响数据的传输？

RocketMQ 的架构十分简单，而简单也意味着稳定和可靠。因此，使用 RocketMQ 做核心数据链路时，其稳定性和可靠性能够避免很多意外，减少不可控因素。

网络抖动往往无法避免，它可能导致数据丢失，而 RocketMQ 能够通过重试机制保证数据的最终一致。比如消息只发一半时发生了网络抖动，网络恢复如何保证数据最终能够被消费者完整地消费？

默认的消费机制下，RocketMQ 有 16 次重试机会，按阶梯重试，重试间隔逐渐增加，最大限度地让消费者能够消费到数据。如果 16 次重试后依然没有消费成功，则消息会进入死信队列，由人工介入处理。产生死信消息后，RocketMQ 能够产生告警，以快速发现并处理问题。

针对数据丢失，RocketMQ 提供了消息轨迹，帮助快速定位，找到问题所在。消费者消息是否成功发送、 broker 是否存储成功、消费者是否成功消费到等问题，都可以通过消息轨迹进行确认。

针对带宽打满的问题，RocketMQ 提供了服务端过滤的功能。假设 Topic 内是访问日志，将 tag 设为域名，消费者组可以只订阅某个域名下的访问日志，RocketMQ 能够在服务端对消息进行过滤，再发送给消费者组。Broker 只会将属于消费者的域名消息发送给消费者，不会发送所有消息，因此能节约大量带宽，可高达 80%-90%。

四、RocketMQ5.0

RocketMQ5.0 架构有两个重大改变，实现了存储计算分离以及轻量级客户端。

存储和计算分离主要为数据层面，将做存储和计算的 broker 拆分成了存储的 broker 和计算的 broker，两类 broker 各司其职，分别负责存储和计算。

此前，RocketMQ 的客户端生态较为丰富，但各个客户端的功能差异较大，难以实现一致。RocketMQ5.0 彻底地解了该问题，实现了轻量级的基于 gRPC 的多语言客户端。RocketMQ5.0 将此前客户端的重逻辑比如 rebalance 等转移至由 Cbroker 负责处理，使客户端逻辑变得非常轻量，客户端只剩消息消费或发消息调用接口，各个语言的逻辑容易统一，兼容性更好，不会出现实现方式不同导致逻辑不一致。

RocketMQ5.0 除了存储和计算分离以外，还实现了数据面和控制面的拆分。控制面主要负责接入，此前只能通过 NameServer 的方式接入，而现在除了 NameServer 以外还提供了一种新的通过 LB Group 的方式接入，更简单易用。 LB Group 能够方便大家用更简单的方式接入，逻辑集群的接入可以通过 LB group 来实现，比如哪些客户端应该连到哪些集群，这也是 NameServer 难以实现的能力。一组 NameServer 会管理一个物理集群，物理集群可拆分为多个逻辑集群，每个逻辑集群能够分给不同的租户使用。

Rocket MQ 可以看作是一个通道，通道有上游和下游，且不同行业的上下游不一样，通道中的数据也不一样。腾讯云认为，当前已经发展到互联网的下半场，也可称为产业互联网。产业互联网分为工业互联网和非工业互联网。非工业的互联网，比如电商、金融、外卖等行业已经发展得如火如荼，对我们的日常生活也产生了巨大影响。

工业互联网包括自动化制造或交通运输行业等，比如道路上的监控摄像头拍摄违章，需要拍下照片后再进行识别处理，往往需要 2-3 天才能对车主进行通知。而随着技术发展，我们期望能够实现实时的违章识别。

互联网已经涉及到每一个领域，但是垂直领域的发展依然非常欠缺。比如配送互联网涉及到交通运输等，需要有交通运输方面的专家与互联网技术进行深度结合，才能对配送行业引起深远广泛的影响。

未来，我们需要技术人员深耕于某一行业，做出真正适用于行业的优秀的互联网产品。

当前，RocketMQ 已经能够支持事件和流。工业互联网行业非常重要的一个元素是 IoT 事件，它可能来自于各种终端设备，每天都会产生大量的、持续的事件，这些大量的数据都需要队列进行传输，提供给下游做计算。

因此，RocketMQ 未来的发展将着力于事件和流。

RocketMQ 已经推出了 RocketMQ Collect、RocketMQ Flink 和 RocketMQ streams ，在流计算上逐渐发力，形成了一整套完善的生态，能够帮助用户快速构建流式应用。而消息更是 RocketMQ 的擅长之处，能够帮助用户在不同场景的消息下方便、快速地接入使用。RocketMQ 已经开源了 MQTT 等协议，使接入设备更快速方便。

随着 RocketMQ 5.0 的发布，RocketMQ 在处理消息、事件和流上实现了统一，有了越来越强大的优势，存储和计算分离的特性也使其能提供更低的成本，使企业上云更省钱、更省力，也更省人力。