技术分享｜RocketMQ 跨集群消息同步（消息路由）的实现与应用

RocketMQ作为一款分布式的消息中间件，不管是采用Master/Slave的主从模式，还是采用Dledger的多副本模式，均保证了在同一个地域集群中的可靠性和可用性。但不同RocketMQ集群的Broker之间无法进行数据备份，如果单个资源池中的RocketMQ集群出现机房毁坏、磁盘故障、服务器损毁等非可预见性重大问题时，就会导致整个集群的可靠性失效。

另一方面，随着消息队列RocketMQ陆续在上海、苏州、郑州等节点上线，RocketMQ得到了大范围应用。用户在使用过程中，可能存在跨资源池消息消费的业务需求，如业务系统A作为Producer部署于华东资源池，业务系统B作为Consumer部署于华北资源池，A发送消息需要B来进行消费。

基于以上两点，我们设计并实现了RocketMQ消息路由功能，即通过路由规则动态规划消息的传输路径，使消息按照过滤条件，从消息源节点路由到目标节点。RocketMQ消息路由（RocketMQ Connect）打通了多个地域RocketMQ集群之间的消息互通，使消息能够按照过滤规则从源集群同步到目标集群 ，高效地实现了分布在不同地域之间的RocketMQ集群的消息同步复制，保证地域之间的数据一致性。

在现有的主流消息中间件中，如Pulsar、Kafka都有提供消息跨集群同步功能的实现方案。

Geo replication是消息队列Pulsar的一种跨机房复制方案，把分散在不同地域的集群通过一定的配置方式，让各集群之间可以相互复制。Geo replication工作原理是在Broker内部，为跨地域的数据复制启动了一组内嵌的额外的生产者和消费者。当一方集群消息被产生后，内嵌的消费者会读取消息，读取完成后，调用内嵌的生产者将消息立即发送到另一方集群。

（图片源自网络）http://pulsar.apache.org/docs/en/administration-geo/

MirrorMaker是消息队列Kafka自带的集群镜像工具，用来实现Kafka集群间的数据同步。MirrorMaker包含了一组消费者，并为每个消费者分配一个线程，这些消费者属于同一个Group，从多个Topic上读取数据，所有的Topic均使用该Group进行消息消费，每个MirrorMaker进程仅有一个生产者，该生产者将数据发送到目标集群上。在数据同步过程中，可以启动多个MirrorMaker进程，即使某个进程挂掉，Topic的同组消费者也可以进行Rebalance，提高容错性。

和上述两种跨集群消息复制方案原理类似，RocketMQ遵循了OpenMessaging Connect规范，通过一个第三方消息中间件对消息进行生产和消费来实现源和目标集群消息的同步。

OpenMessaging是消息中间件领域和流计算领域的一个消息规范，旨在创立与厂商无关、平台无关的分布式消息及流处理领域的应用开发标准。同其他标准（MQTT、JMS、AMQP）不同的是OpenMessaging不仅仅提供了领域模型，还提供了消息领域几种常见的接口（OMOI、OMRI、OMSI、OMCI、OMBI），旨在与其他的生态打通。OpenMessaging Connect的整体目标是解决不同源之间的异构数据拷贝问题，它解决的思路是在各个数据源之间加一层消息中间件，所有的数据都经过消息中间件进行存储和分发。

OpenMessaging作为分布式消息传递的开放标准，其他支持OpenMessaging的消息中间件只需要实现相应的接口就会具有Connect的能力。下文提到的RocketMQ Connect就是OpenMessaging Connect的一个实现，但和上面不一样的是，RocketMQ Connect需要同步的双方是同构数据源，都是RocketMQ集群。

从整体架构上来说，OpenMessaging Connect包含了三个重要的领域模型：Connector、Worker以及Runtime，分别抽象了数据解析，服务运行以及服务调度等具体逻辑。

Connector从整体上规定了一个数据同步服务的生命周期以及拆解了数据同步的具体阶段，其中包含两个核心概念，Source Connector以及Sink Connector，Source负责将数据从源集群数据同步到消息中间件，Sink则负责将数据从消息中间件提取到目标集群。Connector可以创建和管理多个任务（Task），每一个Task执行具体的数据解析和转储任务；

Worker抽象了Connector的运行环境，是所有Connector任务调度的基础。一个Worker可以理解为一个进程，Connector和Task都运行在Worker之上。

Runtime 为Connector提供了高可用，可伸缩，快速容错保障，是所有connector运行的调度中心，为其上运行的各种类型Connector及其关联的Task任务提供统一的运行时环境（包括负载均衡，实例间的调度，配置管理以及集群管理）。

RocketMQ Connect（RocketMQ消息路由）遵循OpenMessaging Connect的设计理念，依托RocketMQ实现。主要包括Replicator以及Runtime，Replicator是运行在RocketMQ Runtime上的消息同步Connector，其主要实现了Connector的机制，能够同步两个独立RocketMQ集群之间的消息 。Runtime是Replicator的运行时环境，负责加载SourceConnector、SinkConnector，并提供RESTful接口，启动Task任务 ，同时，它也是一个分布式系统，支持集群节点之间的服务发现、配置同步、负载均衡等功能，总体架构如下：

在Runtime进程组成的集群中，将源RocketMQ集群各个用户创建Topic中的业务数据，通过端到端Replicator和Task以数据解析和转储的方式异步复制至目标RocketMQ集群，完成跨消息队列集群的消息复制和备份。其中Runtime集群中每个Worker节点启动Replicator相关的配置信息，也会像集群信息一样在集群中每个节点全量同步，同时会持久化到每个节点。集群中如果有某个Worker节点挂掉，集群信息会发生变化，当每个节点检查到集群信息发生了变化就会触发负载均衡，对集群中运行的Replicator和Task重新分配，从而保证故障节点的任务分配到其它节点处理，保证高可用。

Runtime为Connector提供了高可用，可伸缩，快速容错保障，是所有Connector运行的调度中心，负责将不同的数据同步任务均衡的调度到不同的Worker上执行，其核心类如下所示：

Runtime启动后，运行Connector所需的所有环境，包含多个模块。首先，Runtime集群配置相关的实现都在ConfigManagementService中，当有新的节点上线会将其配置发送到RocketMQ进行全量同步，以便其他节点获取并更新配置。每个Runtime在参数storePathRootDir所指定的本地文件夹中都会有四个配置文件，其中connectorConfig.json包含了当前Connector运行的Topic、集群和任务等信息，taskConfig.json中保存了Task执行的具体任务配置。PositionManagementSevice和OffsetManagementSevice是一对相互的服务，分别对应SourceConnector和SinkConenctor的处理进度。RestHandler采用Javalin（用于Java和Kotlin的轻量级Web框架）内置了一系列RESTful接口，负责Connector的创建、停止以及获取Runtime集群的节点和负载信息。RebalanceService负载均衡服务与RocketMQ Consumer以及Queue之间的负载均衡类似，每个节点都运行相同的负载均衡算法来计算该节点应该处理的Task。Runtime启动流程各模块执行流程如下：

Replicator是运行在RocketMQ Runtime上的RocketMQ集群消息同步Connector的别称，其主要功能从源RocketMQ集群同步指定Topic消息数据到目标RocketMQ集群中。在消息同步的过程中，Replicator首先会加载到Runtime中，然后通过Runtime提供的RESTful接口来创建新的Replicator实例和同步Task来实现源和目标RocketMQ集群消息的同步。具体步骤如下：

1、调用Runtime提供的/connectors/:connectorName?config={config}接口发送创建Connector请求，请求参数如下：

2、参数配置过程中，读取接口中目标和源的集群信息（如Namesrv地址，需要同步消息的Topic、消息同步过滤规则等）添加到ConnectorConfig中， 加载Repliactor实现的类文件，同步Repliactor配置和Task配置。

3、触发Connector配置更新，从而触发负载均衡，并启动对应的Task执行消息数据的读取和写入。相关时序图如下：

目前RocketMQ消息路由功能已成功上线，方便用户在不同资源池之间按需同步消息，实现数据跨资源池的一致性。