基于 RocketMQ Connect 构建数据流转处理平台

Posted 2023-04-15 阿里系统软件技术

RocketMQ 是一个可靠的数据集成组件，具备分布式、伸缩性、故障容错等能力，可以实现 RocketMQ 与其他数据系统之间的数据流入与流出。通过 RocketMQ Connect 可以实现 CDC，构建数据湖，结合流计算可实现数据价值。

作者：周波，阿里云智能高级开发工程师， Apache RocketMQ Committer

01 从问题中来的 RocketMQ Connect

在电商系统、金融系统及物流系统，我们经常可以看到 RocketMQ 的身影。原因不难理解，随着数字化转型范围的扩大及进程的加快，业务系统的数据也在每日暴增，此时为了保证系统的稳定运行，就需要把运行压力分担出去。RocketMQ 就担任着这样的角色，它的异步消息处理与高并发读写能力，决定了系统底层的重构不会影响上层应用的功能。而 RocketMQ 的另一个优势——可伸缩能力，使系统在面临流量的不确定性时，实现对流量的缓冲处理。此外，RocketMQ 的顺序设计特性使其成为一个天然的排队引擎，例如，三个应用同时对一个后台引擎发起请求，确保不引起“撞车”事故。因此，RocketMQ 被用在异步解耦、削峰填谷以及事务消息等场景中。

但是，数字化转型浪潮也带来了更多用户对数据价值的关注——如何让数据产生更大利用价值？RocketMQ 自身不具备数据分析能力，但是有不少用户希望从 RocketMQ Topic 中获取数据并进行在线或离线的数据分析。然而，使用市面上的数据集成或数据同步工具，将 RocketMQ Topic 数据同步到一些分析系统中虽然是一种可行方案，却会引入新的组件，造成数据同步的链路较长，时延相对较高，用户体验不佳。

举个例子，假设业务场景中使用 OceanBase 作为数据存储，同时希望将这些数据同步到 Elasticsearch 进行全文搜索，有两种可行的数据同步方案。

方案一： 从 OceanBase 中获取数据，写入 Elasticsearch 组件并进行数据同步，在数据源较少时此方案没什么问题，一旦数据增多，开发和维护都非常复杂，此时就要用到第二种方案。

方案二： 引入消息中间件对上下游进行解藕，这能解决第一种方案的问题，但是一些较为复杂的问题还没有完全解决。比如，如何将数据从源数据同步到目标系统并保证高性能，如果保证同步任务的部分节点挂掉，数据同步依然正常进行，节点恢复依然可以断点续传，同时随着数据管道的增多，如何管理数据管道也变得十分困难。

总的来说，数据集成过程中的挑战主要有五个。

挑战一： 数据源多，市面上可能有上百个数据源，且各数据源的系统差异较大，实现任意数据源之间的数据同步工作量较大，研发周期很长。

挑战二： 高性能问题，如何高效地从源数据系统同步到目的数据系统，并保障其性能。

挑战三： 高可用问题，即 Failover 能力，当一个节点挂掉是否这个节点的任务就停止了，任务重新启动是否还可以断点续传。

挑战四： 弹性扩缩容能力，根据系统流量动态增加或减少节点数量，既能通过扩容满足高峰期业务，也能在低峰期缩减节点，节省成本。

挑战五： 数据管道的管理运维，随着数据管道的增多，运维监控的数据管道也会变得越来越复杂，如何高效管理监控众多的同步任务。

面对上述挑战 RocketMQ 如何解决？

第一，标准化数据集成 API （Open Messaging Connect API）。在 RocketMQ 生态中增加 Connect 组件，一方面对数据集成过程抽象，抽象标准的数据格式以及描述数据的 Schema，另一方面对同步任务进行抽象，任务的创建、分片都抽象成一套标准化的流程。

第二，基于标准的 API 实现 Connect Runtime。Runtime 提供了集群管理、配置管理、位点管理、负载均衡相关的能力，拥有了这些能力，开发者或者用户就只需要关注数据如何获取或如何写入，从而快速构建数据生态，如与OceanBase、MySQL、Elasticsearc 等快速建立连接，搭建数据集成平台。整个数据集成平台的构建也非常简单，通过 Runtime 提供的 RESTFull API 进行简单调用即可。

第三，提供完善的运维工具，方便管理同步任务，同时提供丰富的 Metrics 信息，方便查看同步任务的 TPS、流量等信息。

02 RocketMQ Connect 两大使用场景

这里为大家整理了 RocketMQ Connect 的两大使用场景。

场景一，RocketMQ 作为中间媒介，可以将上下游数据打通，比如在新旧系统迁移的过程中，如果在业务量不大时使用 MySQL 就可以满足业务需求，而随着业务的增长，MySQL 性能无法满足业务要求时，需要对系统进行升级，选用分布式数据库OceanBase 提升系统性能。

如何将旧系统数据无缝迁移到 OceanBase 中呢？在这个场景中 RocketMQ  Connect 就可以发挥作用，RocketMQ Connect 可以构建一个从 MySQL 到 OceanBase 的数据管道，实现数据的平滑迁移。RocketMQ Connect 还可以用在搭建数据湖、搜索引擎、ETL 平台等场景。例如将各个数据源的数据集成到 RocketMQ Topic当中，目标存储只需要对接 Elasticsearch 就可以构建一个搜索平台，目标存储如果是数据湖就可以构建一个数据湖平台。

除此之外，RocketMQ 自身也可以作为一个数据源，将一个 RocketMQ 集群的数据同步到另一个集群，可以构建 RocketMQ 多活容灾能力，这是社区正在孵化的Replicator可以实现的能力。

场景二，RocketMQ 作为端点。 RocketMQ 的生态中提供了流计算能力组件 -RocketMQ Streams，Connector 将各个存储系统的数据集成到 RocketMQ Topic 当中，下游使用 RocketMQ Streams 流计算的能力就可以构建一个实时的流计算平台。当然也可以配合业务系统的 Service 实现业务系统快速从其它存储统一快速获取数据的能力。

还可以将 RocketMQ 作为端点的上游，将业务消息发到 Topic 中，使用 Connector 对数据做持久化或转存的操作。

如此一来，RocketMQ 就具备数据集成能力，可以实现任意任意异构数据源之间的数据同步，同时也具备统一的集群管理、监控能力及配置化搭建数据管道搭建能力，开发者或者用户只需要专注于数据拷贝,简单配置就可以得到一个具备配置化、低代码、低延时、高可用，支持故障处理和动态扩缩容数据集成平台。

那么， RocketMQ Connect 是如何实现的呢？

03 RocketMQ Connect 实现原理

在介绍实现原理前，先来了解两个概念。

概念一，什么是 Connector（连接器）？ 它定义数据从哪复制到哪，是从源数据系统读取数据写入RocketMQ，这种是 SourceConnector，或从 RocketMQ 读数据写入到目标系统，这种是 SinkConnector。Connector 决定需要创建任务的数量，从 Worker 接收配置传递给任务。

概念二，什么是 Task ？ Task 是 Connector 任务分片的最小分配单位，是实际将源数据源数据复制到 RocketMQ（SourceTask），或者将数据从 RocketMQ 读出写入到目标系统（SinkTask）真正的执行者，Task 是无状态的，可以动态的启停任务，多个 Task 可以并行执行，Connector 复制数据的并行度主要体现在 Task 上。一个 Task 任务可以理解为一个线程，多个 Task 则以多线程的方式运行。

通过 Connect 的 API 也可以看到 Connector 和 Task 各自的职责，Connector 实现时就已经确定数据复制的流向，Connector 接收数据源相关的配置，taskClass 获取需要创建的任务类型，通过 taskConfigs 的数量确定任务数量，并且为 Task 分配好配置。Task 拿到配置以后数据源建立连接并获取数据写入到目标存储。通过下面的两张图可以清楚的看到，Connector 和 Task 处理基本流程。

一个 RocketMQ Connect 集群中会有多个 Connector ，每个 Connector 会对应一个或多个 Task，这些任务运行在 Worker（进程）中。Worker 进程是 Connector 和 Task 运行环境，它提供 RESTFull 能力，接收 HTTP 请求，将获取到的配置传递给 Connector 和Task，它还负责启动 Connector 和 Task，保存 Connector 配置信息，保存 Task 同步数据的位点信息，除此以外，Worker 还提供负载均衡能力，Connect 集群高可用、扩缩容、故障处理主要依赖 Worker 的负责均衡能力实现的。Worker 提供服务的流程如下：

Worker 提供的服务发现及负载均衡的实现原理如下：

服务发现：

用过 RocketMQ 的开发者应该知道，它的使用很简单，就是发送和接收消息。消费模式分为集群模式和广播模式两种，集群消费模式下一个 Topic 可以有多个 Consumer 消费消息，任意一个 Consumer 的上线或下线 RocketMQ 服务端都有感知，并且还可以将客户端上下线信息通知给其它节点，利用 RocketMQ 这个特性就实现了 Worker 的服务发现。

配置/Offset同步：

Connector 的配置 /Offset 信息同步通过每个 Worker 订阅相同的 Topic，不同 Worker 使用不同的 Consumer Group 实现的， Worker 节点可以通过这种方式消费到相同Topic的所有数据，即 Connector 配置 /Offset 信息，这类似于广播消费模式，这种数据同步模式可以保证任何一个 Worker 挂掉，该Worker上的任务依旧可以在存活的 Worker 正常拉起运行 ，并且可以获取到任务对应的 Offset 信息实现断点续传， 这是故障转移以及高可用能力的基础。

负载均衡：

RocketMQ 消费场景中，消费客户端 与Topic Queue 之间有负载均衡能力，Connector 在这一部分也是类似的，只不过它负载均衡的对象不一样，Connector 是 Worker 节点和 Task 之间的负载均衡，与 RocketMQ 客户端负载均衡一样，可以根据使用场景选择不同负载均衡算法。

上文提到过 RocketMQ Connect 提供 RESTFull API能力。通过 RESTFull AP 可以创建 Connector，管理 Connector 以及查看 Connector 状态，简单列举：

• POST /connectors/
• GET /connectors/connector name/config
• GET /connectors/connector name/status
• POST /connectors/connector name/stop

目前 Connector 支持单机、集群两种部署模式。集群模式至少要有两个节点，才能保证它的高可用。并且集群可以动态增加或者减少，做到了动态控制提升集群性能和节省成本节省的能力。单机模式更多方便了开发者开发测试 Connector 。

如何如何实现一个 Connector 呢？ 还是结合一个具体的场景看一看，例如业务数据当前是写入 MySQL 数据库中的，希望将 MySQL 中数据实时同步到数据湖 Hudi 当中。只要实现 MySQL Source Connector 、Hudi Sink Connector 这两个 Connector 即可。

下面就以 MySQLSource Connector 为例，来看一下具体的如何实现。

实现 Connector 最主要的就是实现两个 API 。第一个是 Connector API ，除了实现它生命周期相关的 API 外，还有任务如何分配，是通过 Topic、Table 还是通过数据库的维度去分。第二个API 是需要创建的 Task，Connector 通过任务分配将相关的配置信息传递给 Task， Task 拿到这些信息，例如数据库账号，密码，IP，端口后就会创建数据库连接，再通过 MySQL 提供的 BINLOG 机智获取到表的数据，将这些数据写到一个阻塞队列中。Task 有个 Poll 方法，实现 Connector 时只要调用到 Poll 方法时可以获取到数据即可，这样 Connector 就基本写完了。然后打包以 Jar 包的形式提供出来，将它加载到 Worker 的节点中。

创建 Connector 任务后， Worker 中会创建一个或者多个线程，不停的轮询 Poll 方法，从而获取到 MySQL 表中的数据，再通过 RocketMQ Producer 发送到 RocketMQ Broker中，这就是 Connector 从实现到运行的整体过程（见下图）。

04

04 RocketMQ Connect 现状与未来

RocketMQ Connect 的发展历程分为三个阶段。

第一阶段：Preview 阶段

RocketMQ Connect 发展的初期也即 Preview 阶段，实现了 Open Messaging Connect API 1.0 版本，基于该版本实现了 RocketMQ Connect Runtime ，同时提供了 10+ Connector 实现（MySQL，Redis，Kafka，Jms，MongoDB……）。在该阶段，RocketMQ Connect 可以简单实现端到端的数据源同步，但功能还不够完善，不支持数据转换，序列化等能力，生态相对还比较贫乏。

第二阶段：1.0 阶段

在 1.0 阶段，Open Messaging Connect API 进行了升级，支持 Schema、Transform，Converter 等能力，在此基础上对 Connect Runtime 也进行了重大升级，对数据转换，序列化做了支持，复杂Schema也做了完善的支持。该阶段的 API、Runtime 能力已经基本完善，在此基础上，还有30+ Connecotor 实现，覆盖了 CDC、JDBC、SFTP、NoSQL、缓存 Redis、HTTP、AMQP、JMS、数据湖、实时数仓、Replicator、等 Connector 实现，还做了 Kafka Connector Adaptor 可以运行 Kafka 生态的 Connector。

第三阶段：2.0 阶段

RocketMQ Connect 当前处于这个阶段，重点发展 Connector 生态，当 RocketMQ 的 Connector 生态达到 100 + 时，RocketMQ 基本上可以与任意的一个数据系统去做连接。

目前 RocketMQ 社区正在和 OceanBase 社区合作，进行 OceanBase 到 RocketMQ Connect 的研发工作，提供 JDBC 和 CDC 两种模式接入模式，后续会在社区中发布，欢迎感兴趣的同学试用。

05 总结

RocketMQ 是一个可靠的数据集成组件，具备分布式、伸缩性、故障容错等能力，可以实现 RocketMQ 与其他数据系统之间的数据流入与流出。通过 RocketMQ Connect 可以实现 CDC，构建数据湖，结合流计算可实现数据价值。

加入 Apache RocketMQ 社区

十年铸剑，Apache RocketMQ 的成长离不开全球 800+ 位开发者的积极参与贡献，相信在下个版本你就是 Apache RocketMQ 的贡献者，在社区不仅可以结识社区大牛，提升技术水平，也可以提升个人影响力，促进自身成长。

社区 5.x 版本正在进行着如火如荼的开发，以及 30 +个 SIG（兴趣小组）等你加入，欢迎立志打造世界级分布式系统的同学通过以下方式加入社区：

RocketMQ 集成生态再升级：轻松构建云上数据管道

阿里云消息队列 RocketMQ 版是阿里云基于 Apache RocketMQ 构建的低延迟、高并发、高可用、高可靠的分布式“消息、事件、流”统一处理平台，面向互联网分布式应用场景提供微服务异步解耦、流式数据处理、事件驱动处理等核心能力。其自诞生以来一直为阿里集团提供稳定可靠的消息服务，历经多年双十一万亿级流量洪峰的验证。

随着业务需求场景日渐丰富，在多年经验积累后，阿里云 RocketMQ 也迎来了革命性的更新，正式发布了阿里云消息队列 RocketMQ 版 5.0，在架构、网络、负载均衡、存储等诸多方面进行了显著优化。其定位不再局限于消息解耦场景，将全新布局事件驱动和消息流式处理场景。

阿里云 EventBridge 作为云上事件枢纽一直以来都保持着对云上事件、数据的友好生态支持。随着 RocketMQ 5.0版本的用户日渐增多，EventBridge 在近期对 RocketMQ Connector 进行了全面升级。升级之后的 RocketMQ Connector 不仅可以支持RocketMQ 5.0 版本，同时也能支持云上自建 RocketMQ 实例。除此之外，基于成熟的事件流能力，用户使用 EventBridge 也能轻松构建消息路由能力，实现对灾备、数据同步的需求。

本文将从业务架构和 API 使用等方面讲解如何使用 EventBridge 创建阿里云 RocketMQ 4.0、5.0 版本，开源自建版本以及消息路由的相关任务。

EventBridge-RocketMQ 4.0

业务架构

RocketMQ 4.0 版本使用较为经典的 client-nameserver-broker 架构，整个应用主要由生产者、消费者、NameServer 和 Broker 组成。

• Name Server：是一个几乎无状态节点，可集群部署，在消息队列 RocketMQ 版中提供命名服务，更新和发现 Broker 服务。
• Broker：消息中转角色，负责存储消息，转发消息。分为 Master Broker 和 Slave Broker，一个 Master Broker 可以对应多个 Slave Broker，但是一个 Slave Broker 只能对应一个 Master Broker。Broker 启动后需要完成一次将自己注册至 Name Server 的操作；随后每隔 30s 定期向 Name Server 上报 Topic 路由信息。
• 生产者：与 Name Server 集群中的其中一个节点（随机）建立长连接（Keep-alive），定期从 Name Server 读取 Topic 路由信息，并向提供 Topic 服务的 Master Broker 建立长连接，且定时向 Master Broker 发送心跳。
• 消费者：与 Name Server 集群中的其中一个节点（随机）建立长连接，定期从 Name Server 拉取 Topic 路由信息，并向提供 Topic 服务的 Master Broker、Slave Broker 建立长连接，且定时向 Master Broker、Slave Broker 发送心跳。Consumer 既可以从 Master Broker 订阅消息，也可以从 Slave Broker 订阅消息，订阅规则由 Broker 配置决定。

EventBridge在获取用户授权之后，利用生成的 sts 临时授权对客户的 RocketMQ 实例进行消息读取或写入。

API 使用

在 API 介绍方面，我们以创建「自定义总线--自定义事件源」为例，事件目标以及事件流中的API基本一致。

基于 EventBridge 创建 RocketMQ 4.0 任务的 API 和之前基本保持了一致。具体参数如下

• 版本：代表阿里云消息队列 RocketMQ 版本，可选择 4.x 或 5.x；
• RocketMQ 实例：RocketMQ 对应的实例 ID。用户在阿里云 RocketMQ控制台每创建一个实例都会有一个对应的实例 ID，如MQ_INST_123456789***_BX6zY7ah；
• Topic：RocketMQ Topic。选择此 topic 作为事件源的读取对象或者事件目标的写入对象；
• Tag：RocketMQ 消费 Tag，用于消费者过滤消息使用；
• Group ID：RocketMQ 消费组，标识一组特定的消费者，仅事件源有此参数；
• 消费位点：初始消费位点。可选择最新位点、最早位点、或者指定时间戳。

EventBridge-RocketMQ 5.0

业务架构

RocketMQ 5.0 版将通用的存储逻辑下沉，集中解决消息存储的多副本、低延迟、海量队列分区等技术问题，将上层的消息处理剥离出完全的无状态计算层，主要完成协议适配、权限管理、消费状态、可观测运维体系支持，Broker 则继续专注于存储能力的持续优化。存算分离的架构设计，使得从 SDK 接入到线上运维全链路带来全面提升：

1. 轻量版 SDK 的开放和全链路可观测系统的提升：同时支持 4.x 通信协议和全新的 gRPC 通信协议，并内置 OpenTelemetry 埋点支持，新版本 SDK 新增了 10 余个指标埋点。
2. 消息级负载均衡：新版本 SDK 不再参与实际存储队列的负载均衡，消息负载均衡将更加轻量，以单条消息为调度最小单元。
3. 多网络访问支持：新版本支持单一实例同时暴露公网、内网等访问形式，方便客户多网络接入访问。
4. 海量分级存储：新版本开放分级存储历史消息保存能力，消息低成本无大小限制，最长保存 30 天。冷热数据进行分离设计，极大降低消费历史消息对实例的性能影响。

RocketMQ 5.0 版本 可以支持 VPC 内部安全识别，用户上云无需修改代码。在用户授予 EventBridge 网络和 RocketMQ 相关权限之后，用户在 EventBridge 创建 MQ 5.0 Source&Sink 任务的时，EventBridge 会根据 RocketMQ 5.0 实例的 VPC 信息，调用网络组件获取相应代理信息。MQ sdk 侧通过配置代理实现消息的收发。

API 使用

相比于 4.0 实例，5.0 实例多了 VPC、交换机和安全组 3 个参数。

5.0 实例新增了 VPC 属性，用户需要在对应 vpc 内去访问 MQ 5.0 实例。EventBridge 在获得用户授权之后，也是经由 5.0 实例对应的 VPC 内进行消息的收发。创建任务时前端会自动填充好实例的 vpc 和交换机信息。

安全组参数限制了 EventBridge 在 vpc 内的访问策略，用户可以选择使用已有安全组也可以选择快速创建，让 EventBridge 快速创建一个安全组供任务使用。安全组策略推荐使用默认的安全组策略。使用上推荐第一次在此vpc内创建任务时，使用 EventBridge 自动创建一个安全组，后续在此 VPC 内再创建其他任务时，在使用已有中选择 EventBridge 创建的安全组。

EventBridge-自建 Apache RocketMQ

针对用户在阿里云自建 Apache RocketMQ 集群的场景，EventBridge 也支持了消息导出能力。用户通过配置接入点、topic、groupID、VPC 等信息，即可将自建集群中的消息导入 EventBridge，进而对接 EventBridge 目前支持的大量下游生态。

业务架构

抽象来看，EventBridge 访问自建 MQ 实例的链路和阿里云 5.0 版本基本一致，都是从用户 vpc 发起对 MQ 实例的访问。区别在于接入点的不同，前者是用户自建 MQ 集群的nameserver，而后者为阿里云 RocketMQ 提供的接入点，不需要感知真实的 MQ 集群是部署在用户 vpc 还是阿里云 RocketMQ 自身的生产环境。

API 使用

在 API 使用方面，自建集群的大部分参数需要用户手动填入。

• 接入点：nameserver 地址。后续会支持 proxy 地址；
• Topic：RocketMQ Topic。选择此 topic 作为事件源的读取对象或者事件目标的写入对象；
• Tag：RocketMQ 消费 Tag，用于消费者过滤消息使用；
• Group ID：RocketMQ 消费组，标识一组特定的消费者，仅事件源有此参数；
• FilterType：过滤模式，目前支持 Tag 过滤；
• 认证模式：如果开启 ACL 鉴权，可在此配置鉴权信息；
• 消费位点：初始消费位点；
• VPC：自建 MQ 集群对应的 VPC 参数信息；
• 交换机：自建 MQ 集群对应的交换机信息；
• 安全组：EventBridge使用此安全组访问用户自建 MQ 集群，安全组规定了 EventBridge 在此 vpc 内的访问策略。

RocketMQ 消息路由

当用户有灾备或者消息同步的需求时，可能就会需要消息路由能力，即将 A region 下某实例 topic 的消息同步到 B region 的某 topic 中。

对于 EventBridge 而言，消息路由并非单独的一个产品能力，用户通过使用事件流即可实现消息路由。

针对非跨境场景的消息路由，如从北京同步消息到上海，跨 region 网络打通能力由 EventBridge 来实现，用户无需关注过多实现细节。

针对跨境场景，如北京同步消息到新加坡，EventBridge 使用的是公网链路完成对目标实例的写入，使用的是目标 MQ 实例的公网接入点。消息出公网的能力需要用户提供，即需要用户提供 VPC、交换机和安全组配置，此VPC须带有NAT等访问公网能力， EventBridge 使用此 VPC 实现写入目标端公网接入点。

在 API 使用方面，创建消息路由任务本质上是创建事件流，API 参数和上面各类型 RocketMQ 实例任务一致，这里以创建一个青岛到呼和浩特的 RocketMQ 消息路由为例。

1.进入 EventBridge 控制台，regionBar 选择到呼和浩特，点击左侧“事件流”，然后选择“创建事件流”。

2.在事件源页面，事件提供方选择“消息队列 RocketMQ 版”，地域选择青岛，剩余 RocketMQ 相关参数按需求选择。

3.规则页面按需填写，这里选择默认内容。

4.在“目标”页面，服务类型选择“消息队列 RocketMQ 版”，剩余参数按需填写。

5.点击“创建”，等待事件流任务启动即可。

总结

本文介绍了 EventBridge 对接各类型 RocketMQ 实例的基本原理与对应的 API 使用说明，便于已经使用了 RocketMQ 5.0 版本和自建 MQ 实例的用户可以借助 EventBridge 的能力实现事件驱动业务架构的搭建。同时针对灾备和业务消息同步的场景，本文也基于事件流讲解了如何基于 EventBridge 创建 RocketMQ 消息路由任务。

作者：昶风

原文链接

本文为阿里云原创内容，未经允许不得转载。