1.背景5G时代，万物互联，越来越多的企业期望搭建数据分析业务的中台。计划管理企业的数据资产。然而，业务系统或大数据系统中异构数据源之间的数据同步是非常必要的。传统的点对点数据同步工具在处理越来越多的异构数据源同步时会产生N*N个问题。问题，付出的开发成本和维护成本非常高。为此，移动云消息队列MQTT团队积极突破数据孤岛，基于开源的RocketMQConnect组件，推出全新的MQTT-RocketMQConnect架构，助力海量物联网消息自由流动，为万物互联保驾护航。2、MQTT-RocketMQConnect简介首先简单介绍一下MQTT-RocketMQConnect架构的基石——RocketMQConnect，它是RocketMQ数据集成的重要组成部分。对RocketMQ的流入流出，可以实现各种异构数据系统的连接，构建数据管道、ETL、CDC、数据湖等能力。从架构上看，RocketMQConnect通过RocketMQ从其他异构系统中获取数据作为中转消息发送给RocketMQ，然后消费来自RocketMQ的消息并写入其他系统。图1RocketMQConnectMQTT概述-RocketMQConnect基于开源的ApacheRocketMQConnect组件，根据移动云消息队列MQTT的数据模型、业务场景和流规则的特点，做了深入的架构优化设计，实现了云消息队列RocketMQ与MQTT之间的移动消息流和规则管理。主要由Connector、Runtime、Worker和Task组成。Connector包括SourceConnector和SinkConnector，其中，1.SourceConnector：负责从源数据中获取数据，并发送给RocketMQ。2.SinkConnector：负责消费来自RocketMQ的消息，并将数据写入目标存储。RuntimeRuntime是Source和SinkConnectors的运行时环境。负责加载Connector，提供RESTful接口，启动Connector任务，集群节点间服务发现，配置同步，消费进度保存，故障转移，负载均衡等能力。Worker一个Worker进程代表一个Runtime运行环境进程，多个Worker进程组成一个集群来支持更多的Connector和Task并行运行。TaskTask是执行具体数据解析和转储的任务。其中，1.SourceTask：从源数据系统执行并完成数据解析工作，并通过poll()接口暴露给Runtime。2、SinkTask：Runtime从内存中获取数据，通过put()接口方法解析到目标数据源系统。3、DirectTask：同时包含SourceTask和SinkTask，两者直接交互，不经过Runtime。3、MQTT-RocketMQConnect架构设计消息队列MQTT使用RocketMQ作为消息的存储层，消息数据会在RocketMQ中保存一份。因此，消息队列MQTT的存储层RocketMQ可以作为源数据端。使用标准的Connect架构实现异构数据源的数据流转，SourceTask和SinkTask必须一一对应，两者通过中间的RocketMQ进行关联。按照目前的架构，两端都是RocketMQ，使用特殊的DirectTask，让消息不再经过中间的RocketMQ，而是直接流入目标RocketMQ，反之亦然。通过优化架构，可以有效降低延迟，提高速度。图2移动云消息队列MQTT消息存储架构在由运行时进程组成的集群中，源消息队列的海量数据通过端到端的Connector和Task以数据的形式异步复制到目标集群分析和转储，完成消息队列的异构数据流。其中，Runtime集群中的各个Worker节点启动Connector相关的配置信息，会像集群信息一样在集群中的各个节点上完全同步，同时持久化到各个节点。如果集群中某个Worker节点挂掉，集群信息就会发生变化。当各节点检查到集群信息发生变化时，会触发负载均衡，重新分配集群中运行的Connector和Task，以保证故障节点的安全。任务分配给其他节点处理，保证高可用。图3MQTT-RocketMQConnect架构图了解了MQTT-RocketMQConnect的架构，我们来看看如何在MQTT和RocketMQ之间实现一个简单的消息流。通过前面的介绍，应该清楚需要实现两个Connector和Task，一个是RocketMQ作为MQTT存储层到目标RocketMQ的Connector和Task，二是从RocketMQ读取数据的Connector和Task并将其写入目标MQTT。图4MQTT消息流向RocketMQ流程图以消息从MQTT流向RocketMQ为例，主要由三组接口组成：SourceConnector、SourceTask和SinkTask。图5Connector和Task接口概览1.SourceConnector负责connector的生命周期管理，创建相应的Task，将接收到的Connector配置信息拆分为每个任务的配置信息。2.SourceTask负责拉取消息，管理消费者的生命周期。用户也可以根据实际需要添加消息封装、转储等方法。3、SinkTask负责接收SourceTask推送的消息，管理生产者的生命周期。同样，用户也可以根据实际需要添加消息解析、过滤等方法。连接器的生命周期主要分为启动、运行和停止三个阶段。创建和启动连接器创建和启动连接器的过程大致可以分为以下几个阶段：控制台创建规则阶段初始化配置阶段负载均衡阶段图6连接器启动阶段流程图运行任务任务连接器实例启动后，Connector可以根据配置信息进行配置，拆分解析任务，分配任务。这样做的目的是增加并行度，提高处理效率。停止和删除连接器停止和删除连接器的过程大致分为以下几个阶段：控制台停止规则阶段更新配置阶段负载均衡阶段图7连接器停止阶段流程图4.MQTT-RocketMQConnect高可用部署MQTT-RocketMQConnectWorker支持集群和单机两种运行模式。4.1/Cluster模式Cluster模式顾名思义就是由多个Worker节点组成一个高可用的集群。集群之间的config、offset和status信息通过指定RocketMQTopic进行存储，新加入的Worker节点也会获取集群中的config、offset和status信息，并触发负载均衡，重新分配集群中的任务，并使集群达到平衡状态。减少Woker节点或Workers的宕机时间也会触发负载均衡，从而保证集群中的所有任务都能在集群中的存活节点上均衡地正常运行。图8MQTT-RocketMQConnect集群模式示意图4.2/单机模式单机模式，Connector任务运行在单机上，Worker本身不高可用，任务偏移信息持久化到本地。适用于一些不需要高可用或者不需要Worker保证高可用的场景。比如部署在k8s集群中，k8s集群保证高可用。五、MQTT-RocketMQConnect的优秀特性为了保证MQTT和RocketMQ之间有一个高速稳定的消息流通道，MQTT-RocketMQConnect有很多优秀的特性：六、总结与展望本文介绍了异构消息队列的设计海量数据流和实践，基于RocketMQConnect和MQTT的架构特点，做了深入的架构优化和设计，实现了RocketMQ和MQTT之间的消息流转和规则管理。随着万物互联的不断深入，未来消息队列MQTT团队将在现有架构的基础上不断优化创新，例如：1?????为其他异构数据源（Redis、Redis、MySQL、Kafka）循环支持2?????增加集群Worker、Connector、Task状态的管理3?????优化不支持poll方式获取消息的服务
