一、背景     

5G時(shí)代，萬(wàn)物互聯(lián)，越來(lái)越多的企業(yè)期望搭建數(shù)據(jù)分析業(yè)務(wù)中臺(tái)，利用大數(shù)據(jù)技術(shù)、通過(guò)全局規(guī)劃來(lái)治理企業(yè)的數(shù)據(jù)資產(chǎn)。而在業(yè)務(wù)系統(tǒng)，或者大數(shù)據(jù)系統(tǒng)中異構(gòu)數(shù)據(jù)源之間的數(shù)據(jù)同步是十分有必要的，傳統(tǒng)的點(diǎn)對(duì)點(diǎn)的數(shù)據(jù)同步工具，應(yīng)對(duì)越來(lái)越多的異構(gòu)數(shù)據(jù)源同步會(huì)產(chǎn)生N*N的問(wèn)題，付出的開(kāi)發(fā)成本和維護(hù)成本都是非常高的。因此，移動(dòng)云消息隊(duì)列MQTT團(tuán)隊(duì)積極打通數(shù)據(jù)孤島，基于開(kāi)源RocketMQ Connect組件推出全新的MQTT-RocketMQ Connect架構(gòu)，助力海量物聯(lián)網(wǎng)消息自由流轉(zhuǎn)，為萬(wàn)物互聯(lián)保駕護(hù)航。

二、MQTT-RocketMQ Connect介紹   

首先，先簡(jiǎn)單介紹一下MQTT-RocketMQ Connect架構(gòu)的基石—RocketMQ Connect，它是RocketMQ數(shù)據(jù)集成的重要組件，可將各種系統(tǒng)中的數(shù)據(jù)通過(guò)高效、可靠、流的方式，流入流出到RocketMQ，可以實(shí)現(xiàn)各種異構(gòu)數(shù)據(jù)系統(tǒng)的連接，構(gòu)建數(shù)據(jù)管道、ETL、CDC、數(shù)據(jù)湖等能力。

從架構(gòu)上看，RocketMQ Connect就是借助RocketMQ從其他異構(gòu)系統(tǒng)獲取數(shù)據(jù)且以消息的方式發(fā)送到RocketMQ作為中轉(zhuǎn)，然后從RocketMQ消費(fèi)消息并寫(xiě)入到其他系統(tǒng)。

圖1 RocketMQ Connect 總覽

MQTT-RocketMQ Connect在開(kāi)源的Apache RocketMQ Connect組件基礎(chǔ)之上，根據(jù)移動(dòng)云消息隊(duì)列MQTT的數(shù)據(jù)模型、業(yè)務(wù)場(chǎng)景和流轉(zhuǎn)規(guī)則等特點(diǎn)，做了深度的架構(gòu)優(yōu)化與設(shè)計(jì)，實(shí)現(xiàn)了移動(dòng)云消息隊(duì)列RocketMQ與MQTT之間的消息流轉(zhuǎn)與規(guī)則管理。它主要由Connector、Runtime、Worker和Task組成。

Connector

包含 Source Connector和 Sink Connector兩類，其中，

1.Source Connector：負(fù)責(zé)從源數(shù)據(jù)中獲取數(shù)據(jù)并將其發(fā)送到 RocketMQ。

2.Sink Connector：負(fù)責(zé)使用來(lái)自 RocketMQ的消息并將數(shù)據(jù)寫(xiě)入目標(biāo)存儲(chǔ)。

Runtime

Runtime是Source、Sink Connector的運(yùn)行時(shí)環(huán)境，負(fù)責(zé)加載Connector，提供RESTful接口，啟動(dòng)Connector任務(wù)，集群節(jié)點(diǎn)之間服務(wù)發(fā)現(xiàn)、配置同步、消費(fèi)進(jìn)度保存、故障轉(zhuǎn)移、負(fù)載均衡等能力。

Worker

一個(gè)Worker進(jìn)程代表一個(gè)Runtime 運(yùn)行時(shí)環(huán)境進(jìn)程，多個(gè)Worker進(jìn)程組成了一個(gè)集群，支持更多的Connector 和 Task的并行運(yùn)行工作。

Task

Task是執(zhí)行具體的數(shù)據(jù)解析和轉(zhuǎn)儲(chǔ)的任務(wù)，其中，

1.SourceTask：從源數(shù)據(jù)系統(tǒng)中，執(zhí)行完成數(shù)據(jù)解析工作，通過(guò)poll()接口暴露給Runtime。

2.SinkTask：Runtime從內(nèi)存獲取數(shù)據(jù)并通過(guò)put()接口方法解析至目標(biāo)數(shù)據(jù)源系統(tǒng)中。

3.DirectTask：同時(shí)包含SourceTask和SinkTask，兩者直接交互，不再經(jīng)過(guò)Runtime。

 三、MQTT-RocketMQ Connect架構(gòu)設(shè)計(jì)  

消息隊(duì)列MQTT以RocketMQ作為消息的存儲(chǔ)層，消息數(shù)據(jù)會(huì)在RocketMQ中保存一份。因此，可以將消息隊(duì)列MQTT的存儲(chǔ)層RocketMQ作為源數(shù)據(jù)端。采用標(biāo)準(zhǔn)的Connect架構(gòu)要實(shí)現(xiàn)異構(gòu)數(shù)據(jù)源的數(shù)據(jù)流轉(zhuǎn)，Source Task 和Sink Task必須一一對(duì)應(yīng)，兩者通過(guò)中間的RocketMQ關(guān)聯(lián)。按照現(xiàn)在的架構(gòu)兩端都是RocketMQ，使用一個(gè)特殊的Direct Task，讓消息不再經(jīng)過(guò)中間的RocketMQ，而是直接流入到目標(biāo)RocketMQ中，反之亦然。通過(guò)優(yōu)化架構(gòu)可以有效降低時(shí)延，提升速率。

圖2 移動(dòng)云消息隊(duì)列MQTT消息存儲(chǔ)架構(gòu)

在Runtime進(jìn)程組成的集群中，將源消息隊(duì)列的海量數(shù)據(jù)，通過(guò)端到端Connector和Task以數(shù)據(jù)解析和轉(zhuǎn)儲(chǔ)的方式異步復(fù)制至目標(biāo)集群，完成異構(gòu)消息隊(duì)列的數(shù)據(jù)流轉(zhuǎn)。其中Runtime集群中每個(gè)Worker節(jié)點(diǎn)啟動(dòng)Connector相關(guān)的配置信息，也會(huì)像集群信息一樣在集群中每個(gè)節(jié)點(diǎn)全量同步，同時(shí)會(huì)持久化到每個(gè)節(jié)點(diǎn)。集群中如果有某個(gè)Worker節(jié)點(diǎn)掛掉，集群信息會(huì)發(fā)生變化，當(dāng)每個(gè)節(jié)點(diǎn)檢查到集群信息發(fā)生了變化就會(huì)觸發(fā)負(fù)載均衡，對(duì)集群中運(yùn)行的Connector和Task重新分配，從而保證故障節(jié)點(diǎn)的任務(wù)分配到其它節(jié)點(diǎn)處理，保證高可用。

圖3  MQTT-RocketMQ Connect架構(gòu)圖

了解了MQTT-RocketMQ Connect的架構(gòu)，下面看一下如何自己實(shí)現(xiàn)一個(gè)簡(jiǎn)單的MQTT和RocketMQ之間的消息流轉(zhuǎn)。

通過(guò)前面的介紹，應(yīng)該清楚，需要實(shí)現(xiàn)兩個(gè)Connector和Task，一個(gè)是從作為MQTT存儲(chǔ)層的RocketMQ到目標(biāo)RocketMQ的Connector和Task，第二個(gè)是從RocketMQ讀數(shù)據(jù)寫(xiě)入到目標(biāo)MQTT的Connector和Task。

圖4  MQTT消息流轉(zhuǎn)到RocketMQ流程圖

以消息從MQTT流轉(zhuǎn)到RocketMQ為例，主要由三組接口組成：SourceConnector、SourceTask和SinkTask。

圖5  Connector和Task接口概覽

1.SourceConnector負(fù)責(zé)connector生命周期的管理、創(chuàng)建對(duì)應(yīng)的Task并將接收到的Connector配置信息拆分出每個(gè)task的配置信息。

2.SourceTask負(fù)責(zé)拉取消息，并對(duì)消費(fèi)者的生命周期進(jìn)行管理。用戶還可以根據(jù)實(shí)際需要添加消息封裝、轉(zhuǎn)存等方法。

3.SinkTask負(fù)責(zé)接收SourceTask推送的消息，并對(duì)生產(chǎn)者的生命周期進(jìn)行管理。同樣的，用戶還可以根據(jù)實(shí)際需要添加消息解析，過(guò)濾等方法。

一個(gè)connector的生命周期主要分為三個(gè)階段：啟動(dòng)、運(yùn)行、停止。

創(chuàng)建并啟動(dòng)connector

創(chuàng)建并啟動(dòng)Connector過(guò)程大致可以分為以下幾個(gè)階段：

• 控制臺(tái)創(chuàng)建規(guī)則階段
• 初始化配置階段
• 負(fù)載均衡階段

圖6  Connector啟動(dòng)階段流程圖

運(yùn)行task任務(wù)

• 在Connector 實(shí)例被啟動(dòng)后，Connector可以根據(jù)配置信息，對(duì)解析任務(wù)進(jìn)行拆分，分配出task。這么做的目的是為了提高并行度，提升處理效率。

停止并刪除connector

停止并刪除Connector過(guò)程大致可以分為以下幾個(gè)階段：

• 控制臺(tái)停止規(guī)則階段
• 更新配置階段
• 負(fù)載均衡階段

圖7  Connector停止階段流程圖

四、MQTT-RocketMQ Connect高可用部署  

MQTT-RocketMQ Connect Worker支持兩種運(yùn)行模式，集群和單機(jī)模式。

4.1/集群模式

集群模式，顧名思義，由多個(gè)Worker節(jié)點(diǎn)組成高可用集群。集群間的config、offset和status信息通過(guò)指定RocketMQ Topic存儲(chǔ)，新增Worker節(jié)點(diǎn)也會(huì)獲取到集群中的這些config、offset和status信息，并且觸發(fā)負(fù)載均衡，重新分配集群中的任務(wù)，使集群達(dá)到均衡的狀態(tài)。減少Woker節(jié)點(diǎn)或者Worker宕機(jī)也會(huì)觸發(fā)負(fù)載均衡，從而保障集群中所有的任務(wù)都可以均衡的在集群中存活的節(jié)點(diǎn)中正常運(yùn)行。

圖8  MQTT-RocketMQ Connect集群模式示意圖

4.2 /單機(jī)模式

單機(jī)模式，Connector任務(wù)運(yùn)行在單機(jī)上，Worker本身沒(méi)有高可用，任務(wù)offset信息持久化在本地。適合一些對(duì)高可用要求不高或者不需要Worker保障高可用的場(chǎng)景，例如部署在k8s集群中，由k8s集群保障高可用。

五、MQTT-RocketMQ Connect優(yōu)秀特性

為了保證MQTT和RocketMQ之間有高速穩(wěn)定的消息流轉(zhuǎn)通道，MQTT-RocketMQ Connect具有許多優(yōu)秀的特性：

六、總結(jié)與展望     

本文介紹了異構(gòu)消息隊(duì)列海量數(shù)據(jù)流轉(zhuǎn)的設(shè)計(jì)與實(shí)踐，基于RocketMQ Connect和移動(dòng)云消息隊(duì)列MQTT本身的架構(gòu)特點(diǎn)，做了深度的架構(gòu)優(yōu)化與設(shè)計(jì)，實(shí)現(xiàn)了移動(dòng)云消息隊(duì)列RocketMQ與MQTT之間的消息流轉(zhuǎn)與規(guī)則管理。隨著萬(wàn)物互聯(lián)的持續(xù)深入，未來(lái)消息隊(duì)列MQTT團(tuán)隊(duì)還將基于現(xiàn)在的架構(gòu)繼續(xù)優(yōu)化和創(chuàng)新，例如：

   1       ?  ?  ?  ?  ?

增加對(duì)其他異構(gòu)數(shù)據(jù)源（Redis、MySQL、Kafka）等組件的消息流轉(zhuǎn)支持

   2       ?  ?  ?  ?  ?

增加對(duì)集群Worker、Connector、Task狀態(tài)的管理

   3       ?  ?  ?  ?  ?

優(yōu)化不支持poll方式獲取消息的服務(wù)