让数据流动起来，RocketMQ Connect 技术架构解析

标签：Task 架构 Worker Connector Connect 数据 RocketMQ

作者：周波

Why RocketMQ Connect

在业务系统，或者大数据系统中不同数据源之间的数据同步是十分常见的，传统的点对点的数据同步工具，在面临越来越多的数据源点对点的数据同步会产生 N*N 的问题，开发成本，维护成本都是非常高的，因为上下游是耦合的，一个数据源的逻辑调整，也可能会影响多个数据管道之间的数据同步。引入消息中间件将上下游解耦这些问题是不是就迎刃而解了？通过消息中间件上下游的数据源的处理逻辑相对独立，但是如果我们直接用 RocketMQ 的生产者消费者来构建数据管道还需要考虑以下挑战。

异构数据源越来越多，如何实现任意数据源的数据同步?
高性能，如何高效的在源数据源到目的数据源的数据同步？
高可用，故障处理能力，当一个节点挂掉是否这个节点的任务就停止了，任务重新启动是否还可以断点续传。
弹性扩缩容，是否可以根据系统数据的变动动态的增加减少节点？
集群监控，运维管理，随着数据管道的增多，如何管理，运维监控这些数据管道也变的越来越复杂。

我们通过 RocketMQ 的生产者消费者开发一个完整的数据管道是比较简单的，可能几天就可以完成，但是如果想一起解决上述这些问题可能需要几个月才能完成，而且一旦我们完成了这样一个系统，就会发现，它与 RocketMQ Connect 是一个非常类似的系统，而上述这些问题是 RocketMQ Connect 都已经完美解决的问题。下面将系统化的介绍一下 RocketMQ Connect 是如何解决这些问题的。

RocketMQ Connect 介绍

什么是 RocketMQ Connect

RocketMQ Connect 是 RocketMQ 数据集成重要组件，可将各种系统中的数据通过高效，可靠，流的方式，流入流出到 RocketMQ，它是独立于 RocketMQ 的一个单独的分布式，可扩展，可容错系统，它具备低延时，高靠性，高性能，低代码，扩展性强等特点，可以实现各种异构数据系统的连接，构建数据管道，ETL，CDC，数据湖等能力。

RocketMQ Connect 具备以下优势：

上下游解耦，因为上下游都是通过 RocketMQ 做数据中转的，所以上下游系统是异步解耦的，上下游系统变化不会互相影响，只需要关注与 RocketMQ 的数据同步。
低延迟，流式数据传输，秒级甚至毫秒级延迟。
可靠性高，集群部署，支持 failover，数据可回放。
扩展性强，通过 Connect API 可以实现与任意系统之间建立连接。
实现简单，专注数据拷贝，无需关注分布式，故障转移、弹性伸缩等能力。
可伸缩，支持动态扩缩容。
配置化，低代码，已经支持的数据源只需简单配置就可完成不同数据源之间的，不支持的数据源，只需要实现 Connect API 专注数据拷贝即可快速完成新的数据源的支持。

Connector 工作原理

RocketMQ Connect 是一个独立的的分布式，可伸缩，容错的系统，它主要为 RocketMQ 提供与各种外部系统的数据的流入流出能力。用户不需要编程，只需要简单的配置即可使用 RocketMQ Connect，例如从 MySQL 同步数据到 RocketMQ，只需要配置同步所需的 MySQL 的账号密码，链接地址，和需要同步的数据库，表名就可以了。

Connector的使用场景

构建流式数据管道

在业务系统中，利用 MySQL 完善的事务支持，处理数据的增删改，使用 ElasticSearch，Solr 等实现强大的搜索能力,或者将产生的业务数据同步到数据分析系统，数据湖中（例如 Hudi），对数据进一步处理从而让数据产生更高的价值。使用 RocketMQ Connect 很容易实现这样的数据管道的能力，只需要配置 3 个任务，第一个从 MySQL 获取数据的任务，第二、三个是从 RocketMQ 消费数据到 ElasticSearch、Hudi 的任务，配置这 3 个任务就实现了从 MySQL 到 ElasticSearch，MySQL 到 Hudi 的两条数据管道，既可以满足业务中事务的需求，搜索的需求，又可以构建数据湖。

1. 系统迁移

随着业务的发展，往往会对旧的系统进行系统优化，甚至重构，在新的系统中可能选用新的组件，新的技术框架，旧的系统又无法直接停机将数据迁移到新系统。如何动态的将数据从旧存储系统迁移到新的存储系统，例如旧的系统使用的是 ActiveMQ，新的系统选用了 RocketMQ，只需要配置一个从 ActiveMQ 到 RocketMQ 的 Connector 任务即可实时的将旧系统的数据迁移到新系统，实现旧系统的 ActiveMQ 上的数据迁移到新系统 RocketMQ 中，同时也不影响业务，用户无感知。

2. CDC

图片来源：https://luminousmen.com/post/change-data-capture

CDC 作为 ETL 模式之一，可以通过近乎实时的增量捕获数据库的 INSERT、UPDATE，DELETE 变化，RocketMQ Connect 流式数据传输，具备高可用，低延时等特性，通过 Connector 很容易实现 CDC。

数据的序列化与转换

前面对 RoketMQ Connect 已经有了基本的介绍，知道 RocketMQ Connect 可以实现很多数据源之间的数据同步，那么这些数据在 RocketMQ Connect 中是如何流转和处理的呢？Converter 和 Transform在数据的序列化和转换有十分重要的作用，下面介绍下 Converter 和 Transform。

Converter 和 Transform 和 Connector 一样都属于 RocketMQ Connect 的插件，可以使用 RocketMQ Connect 内置的实现，也可以自定义实现个性化的处理逻辑，通过 RocketMQ Connect Worker 插件加载能力可以将自定义实现加载到 Worker 运行时中，通过简单的配置既可以使用自定义的插件。

Converter： 转换器，主要负载数据的序列化和反序列化。Converter 可以将数据和数据的 Schema 进行转换并且完成序列化，还可以与Schema Registry配合，将 Schema 注册到 Schema Register 中。

Transform： 单消息转换，对 Connect 数据对象进行转换，例如过滤掉不需要的数据, 字段过滤，替换数据 Key，大小写转换等，都可以通过 Transform 进行转换。

Connector 的数据处理的基本流程，Source Connect 从源数据系统获取数据封装成 Connect 标准的数据对象，然后经过 Transform 对单条数据进行转换，转换后的数据会经过 Converter，Converter 会将数据和 Schema 处理成支持的格式并进行序列化，如果是支持 Schema Registry，Converter 会将数据的 Schema 注册到 Schema Registry 中，然后序列化数据，最后将序列化后的数据发送到 RocketMQ 中。Sink Connect 处理流程是从 RocketMQ Topic 拉取数据，经过数据的反序列化，然后对每条消息进行转换，如果配置了 Transform，最后 Connector 将数据写入到目标存储。

RESTful 接口

Connector Worker 节点提供了 RESTful 接口方便 Connector 的创建，停止，配置信息获取接口，使用这些接口可以很方便的管理 Connector，只需要调用对应的接口带上对应的配置即可，下面介绍一些常用的接口。

•POST /connectors/{connector name}

•GET /connectors/{connector name}/config

•GET /connectors/{connector name}/status

•POST /connectors/{connector name}/stop

通过这些 API 即可向 Connect Worker 创建 Connector，停止 Connector，查询 Connector 的配置和状态。

Metrics

可观测性在许多系统中都扮演着非常重要的作用，它可以帮助观察系统运行情况，系统是否正常运行，系统的高峰期，空闲期是在什么时候，帮助对系统进行监控问题定位等等。Worker 提供了丰富的 Metrics 信息，可以通过 Metrics 观察到所有 Worker 总的 tps，消息数量，还可以观察到每个 task 的 tps，成功处理的数量和处理失败的数量等等，通过这些 Metrics 信息可以很方便的运维 Connector。

将 Metrics 信息展示在 Prometheus 中是很多系统的选择，Prometheus 也提供了接入方式，通过实现一个 Prometheus Exporter 即可，新增或者修改 Metrics 修改这个 Exporter 即可，这是一种将 Metrics 接入 Prometheus 的方式。通过 RocketMQ Connect 可以提供一种新的接入方式，实现一个通用的 Sink Prometheus Connector，Source Connector 是获取 Metrics 所需的 Connector，例如 RocketMQ 和 RocketMQ Connect 的 Metrics 信息都写在了文件里，那么获取 Metrics 的 Connector 就是 SFTP Source Connector，这样就可以将 Metrics 信息通过 Connector 的方式接入 Prometheus。

Connector 数据质量

RocketMQ Connect 和 RocketMQ 一样，实现了 at least once 的语义，即保障至少一次。Source Connector 发送消息到 RocketMQ 失败，producer 会重试，如果重试失败，可以选择跳过，还是停止任务，可以通过配置进行选择。Sink Connector 拉取消息同样会重试，并且在重试达到一定次数时会将消息发送到死信队列，进而保障消息不丢失。

Connector 部署

在创建 Connector 时，一般是通过配置完成的，Connector 一般包含逻辑的 Connector 连接器和执行数据复制的 Task 即物理线程，如下图所示，两个 Connector 连接器和它们对应的运行 Task 任务。

一个 Connector 也可以同时运行多个任务，提高 Connector 的并行度，例如下图所示的 Hudi Sink Connector 有 2 个任务，每个任务处理不同的分片数据，从而增加 Connector 的并行度，进而提高处理性能。

RocketMQ Connect Worker 支持两种运行模式，集群和单机集群模式，顾名思义，有多个 Worker 节点组成，推荐最少有 2 个 Worker 节点，组成高可用集群。集群间的配置信息，offset 信息，status 信息通过指定 RocketMQ Topic 存储，新增 Worker 节点也会获取到集群中的这些配置，offset，status 信息，并且触发负载均衡，重新分配集群中的任务，使集群达到均衡的状态，减少 Woker 节点或者 Worker 宕机也会触发负载均衡，从而保障集群中所有的任务都可以均衡的在集群中存活的节点中正常运行。

单机模式，Connector 任务运行在单机上，Worker 本身没有高可用，任务 offset 信息持久化在本地。适合一些对高可没有什么要求或者不需要 Worker 保障高可用的场景，例如部署在 K8s 集群中，由 K8s 集群保障高可用。

Connector 原理

概念

Connector

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

Task

是 Connector 任务分片的最小分配单位，是实际将源数据源数据复制数据到RocketMQ（SourceTask），或者将数据从 RocketMQ 读取数据写入到目标系统（SinkTask）真正的执行者，Task 是无状态的可以动态的启停任务，多个 Task 是可以并行执行的，Connector 复制数据的并行度主要体现在 Task 数量上。

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

Worker

Worker 进程是 Connector 和 Task 运行环境，它提供 RESTful 能力，接受 HTTP 请求，将获取到的配置传递给 Connector 和 Task。除此之外它还负责启动 Connector 和 Task，保存 Connector 配置信息，保存 Task 同步数据的位点信息，负载均衡能力，Connect 集群高可用，扩缩容，故障处理主要依赖 Worker 的负责均衡能力实现的。

从上面这张图，看到 Worker 通过提供的 REST API 接收 http 请求，将接收到的配置信息传递给配置管理服务，配置管理服务将配置保存到本地并同步给其它 Worker 节点，同时触发负载均衡。

从下面这张图可以看到 Connect 集群中多个 Worker 节点负载均衡后在各自节点启动 Connector 和 Task 的情况。

Worker集群的服务发现

****学习和使用 RocketMQ 的时候，我们知道 NameSrv 是 RocketMQ 服务注册与发现的重要组件，具备简单轻量易运维等特性。通过前面 Connector 部署方式的介绍，RocketMQ Connect Worker 是可以集群方式部署的，那么 Worker 节点之间的服务发现是如何实现的呢？Worker 的服务发现是否依赖外部组件?

在使用一项技术或者中间件的时候，尽量少的依赖额外的组件，这对系统的稳定性，可维护性都非常重要，学习成本也会低一些，这也很容易理解，如果多依赖一个组件，就需要多维护一个，学习成本和后期维护成本就会越高。RocketMQ Connect 也是这个原则，尽量少或者不依赖额外的外部组件。通过前面对 RocketMQ Connect 的介绍我们知道 RocketMQ Connect 是 RocketMQ 数据集成，数据流入流出的重要组件，所以 RocketMQ Connect 的运行是依赖一个 RocketMQ 集群的，能否利用RocketMQ的特性实现 Worker 的服务发现呢？答案是肯定的。通过 RocketMQ 客户端有消费客户端变化通知机制，只需要在每个 Worker 启动一个消费者，设置相同的 ConsumerGroup，注册监听 NOTIFY_CONSUMER_IDS_CHANGED 事件即可，增加或减少 Worker 节点就会触发 NOTIFY_CONSUMER_IDS_CHANGED 事件，监听到这个事件就可以触发 Worker 集群的负责均衡，通过 RocketMQ 客户端接口既可以获取所有 Worker 对应的客户端标识，最后根据在线的客户端和任务进行负载均衡。

配置同步

Topic 的订阅消费分为两种，集群消费，广播消费，不过不同的客户端使用不同的 ConsumerGroup 也可以达到广播消费的目的，RocketMQ Connect Worker 之间元数据的同步就是根据这种原理实现的。所有的 Worker 节点订阅相同的 Topic（connector-config-topic），每个 Worker 节点使用不同的 consumerGroup，这样就可以实现广播的方式消费同一个 Topic，每个 Worker 节点只需要配置发生变化时向 connector-config-topic 发送配置信息，即可实现各个 Worker 节点间的配置同步。

位点同步

位点同步与配置同步类似只是使用的是不同的 Topic 和 consumerGroup服务发现与配置/位点管理：

Connector 配置管理和 Task 任务的位点管理是类似的，除了需要将配置和位点信息持久化到本地，还会将配置和位点信息同步给集群的其它 Worker 节点，这个同步方式也是通过所有 Worker 节点订阅相同的 Topic，使用不同的 consumerGroup 实现的。

负载均衡

Connect 的负载均衡和 RocketMQ 消费端负载均衡是类似的。都是在每个节点上运行相同的负载均衡算法，只不过负载均衡的对象不一样，RocketMQ 消费端负载均衡是相同 ConsumerGroup 消费端与 MessageQueue 之间负载均衡，Connect 是 Worker 节点与 Connector，Worker 节点与 Task 之间的负载均衡，不过逻辑是类似的，负载均衡算法获取到所有 Worker 和所有的 Connector，Task，并对这些信息排序，根据当前 Worker 节点在排序中所有 Worker 的位置，排序后的 Connector，Task 位置与当前的 Worker 位置取模，就可以为当前 Worker 节点分配好需要运行那些 Connector 和那些 Task，然后 Worker 在本节点启动这些 Connector 和 Task。

Worker 扩缩容，Worker 宕机，新增 Connector 配置都会触发重新分配，因此 Worker 集群是弹性的可以故障处理，动态扩缩容的。

Connector插件加载

Worker 加载 Connector 插件 jar 包类似于 Tomcat 加载 War 包。Tomcat 加载 War 包并不会像 jvm 加载类一样使用双亲委派模型对类进行加载，而是使用自定义的 ClassLoader 加载类，使用不同的 ClassLoader 加载不同的 War 包的不同的类。这样可以避免不同的 War 包中引用相同的 jar 包因为版本的不同各种包引用问题，Worker 也是一样，在加载不同的类插件时为其创建单独的 ClassLoader，从而避免相同的类，因为不同 War 包引用相同 jar，并且版本不同而导致的各种包引用问题，在类加载方面 Worker 像一个分布式的 Tomcat。

动手实现 Connector

了解了 RocketMQ Connect 的实现原理，下面看一下如何自己实现一个 Connector。看下面这个场景：

业务数据写入到 MySQL，然后同步到 Hudi，通过 Hudi 构建数据湖，这个跟我们开头讲到的场景比较相似。通过 Connector 如何实现这个流程?

通过前面的介绍，应该清楚，需要实现两个 Connector，一个是从 MySQL 到 RocketMQ 的 SourceConnector，第二个是从 RocketMQ 读数据写入到 Hudi 的 Sink Connector。

以 MySqlSourceConnector 为例介绍如何自己写一个 Connector。

首先要实现一个 SourceConnector，实现类 MySqlConnectorImpl 继承 SourceConnector 接口,实现配置初始化，taskClass，taskConfigs 等接口。taskClass 方法指定创建 MySqlTask，taskConfigs 将接收到的 Connector 配置并分好每个 Task 配置，这些配置包含连接 MySql 实例相关的账号，密码，address，port 等。Worker 启动 MySqlTask 并将配置传给 MySqlTask，MySqlTask 创建到 MySql 实例的链接并获取 MySql Binlog，解析 Binlog，并将解析的数据封装成 ConnectRecord 放到 BlockingQueue，poll 接口从 BlockingQueue 中取数据返回。这样就实现了一个 Connector。

将写好的 Connector 打包放到 Worker 指定目录加载到 Worker 进程，通过请求 Worker http 接口创建 MySqlSourceConnector。Worker 启动 MysqlConnector，通过 MySqlConnector taskClass 获取 MySqlTask 并启动 Task，WorkerSourceTask 从 poll 接口获取数据，然后把获取到的数据通过 Producer 发送到 RocketMQ，这样就完成了 MySql 到 RocketMQ 的数据同步的 Connector。

Connector 现状与未来

CDC 方面已与 debezium 完成适配，jdbc 标准协议也已支持并且在此基础上与 openmldb 社区合作开发了与面向机器学习数据库 openmldb 的 Connector，数据湖方面与 Hudi 也建立了链接，不久的将来还会与 Doris，clickhouse，es 等流行的存储建立连接，如果有熟悉的的存储，每个人都可以通过 OpenMessaging Connect API 开发一个 RocketMQ 连接器。

提到 OpenMessaging Connect API，简单介绍一下 OpenMessaging，OpenMessaging 是 Linux 基金会下一个开源组织，致力于制定消息领域的标准，除了 OpenMessaging Connect Api 还有存储方面的标准 OMOI，压测方面的标准 OMBI，流计算方面的标准 OMS 等等。

Connector Tutorial

QuickStart

https://github.com/apache/rocketmq-connect

RocketMQ Connect & OpenMLDB

OpenMLDB 是一个开源机器学习数据库，提供线上线下一致的生产级特征平台。通过与 OpenMLDB 社区合作，RocketMQ 与机器学习数据库 OpenMLDB 建立连接。

RocketMQ Connect & Debezium

https://mp.weixin.qq.com/s/YNjylhmo1IlvAEKwpjjMkg

总结

本文介绍了 RocketMQ Connect 的概念，然后讲解了 RocketMQ Connect 的实现原理，对服务发现，配置同步，位点同步，负载均衡都有了初步的介绍，接着以 MySqlSourceConnector 为例讲解了如何自己实现一个 Connector，最后对 Connect API 和生态做了一些介绍，提供了一些 RocketMQ Connect 相关的上手教程，希望本文对学习 RocketMQ Connect 有帮助，更希望感兴趣的同学能参与进来一起贡献社区，实现一个自己熟悉的存储系统到 RocketMQ 的 Connector。

作者介绍：

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

参考链接：

rocketmq-connect