一、背景     

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

二、MQTT-RocketMQ Connect介紹   

首先，先簡單介紹一下MQTT-RocketMQ Connect架構(gòu)的基石—RocketMQ Connect，它是RocketMQ數(shù)據(jù)集成的重要組件，可將各種系統(tǒng)中的數(shù)據(jù)通過高效、可靠、流的方式，流入流出到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)消息并寫入到其他系統(tǒng)。

圖1 RocketMQ Connect 總覽

MQTT-RocketMQ Connect在開源的Apache RocketMQ Connect組件基礎(chǔ)之上，根據(jù)移動云消息隊(duì)列MQTT的數(shù)據(jù)模型、業(yè)務(wù)場景和流轉(zhuǎn)規(guī)則等特點(diǎn)，做了深度的架構(gòu)優(yōu)化與設(shè)計(jì)，實(shí)現(xiàn)了移動云消息隊(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é)使用來自 RocketMQ的消息并將數(shù)據(jù)寫入目標(biāo)存儲。

Runtime

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

Worker

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

Task

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

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

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

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

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

消息隊(duì)列MQTT以RocketMQ作為消息的存儲層，消息數(shù)據(jù)會在RocketMQ中保存一份。因此，可以將消息隊(duì)列MQTT的存儲層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必須一一對應(yīng)，兩者通過中間的RocketMQ關(guān)聯(lián)。按照現(xiàn)在的架構(gòu)兩端都是RocketMQ，使用一個特殊的Direct Task，讓消息不再經(jīng)過中間的RocketMQ，而是直接流入到目標(biāo)RocketMQ中，反之亦然。通過優(yōu)化架構(gòu)可以有效降低時延，提升速率。

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

在Runtime進(jìn)程組成的集群中，將源消息隊(duì)列的海量數(shù)據(jù)，通過端到端Connector和Task以數(shù)據(jù)解析和轉(zhuǎn)儲的方式異步復(fù)制至目標(biāo)集群，完成異構(gòu)消息隊(duì)列的數(shù)據(jù)流轉(zhuǎn)。其中Runtime集群中每個Worker節(jié)點(diǎn)啟動Connector相關(guān)的配置信息，也會像集群信息一樣在集群中每個節(jié)點(diǎn)全量同步，同時會持久化到每個節(jié)點(diǎn)。集群中如果有某個Worker節(jié)點(diǎn)掛掉，集群信息會發(fā)生變化，當(dāng)每個節(jié)點(diǎn)檢查到集群信息發(fā)生了變化就會觸發(fā)負(fù)載均衡，對集群中運(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)一個簡單的MQTT和RocketMQ之間的消息流轉(zhuǎn)。

通過前面的介紹，應(yīng)該清楚，需要實(shí)現(xiàn)兩個Connector和Task，一個是從作為MQTT存儲層的RocketMQ到目標(biāo)RocketMQ的Connector和Task，第二個是從RocketMQ讀數(shù)據(jù)寫入到目標(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)建對應(yīng)的Task并將接收到的Connector配置信息拆分出每個task的配置信息。

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

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

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

創(chuàng)建并啟動connector

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

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

圖6  Connector啟動階段流程圖

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

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

停止并刪除connector

停止并刪除Connector過程大致可以分為以下幾個階段：

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

圖7  Connector停止階段流程圖

四、MQTT-RocketMQ Connect高可用部署  

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

4.1/集群模式

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

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

4.2 /單機(jī)模式

單機(jī)模式，Connector任務(wù)運(yùn)行在單機(jī)上，Worker本身沒有高可用，任務(wù)offset信息持久化在本地。適合一些對高可用要求不高或者不需要Worker保障高可用的場景，例如部署在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和移動云消息隊(duì)列MQTT本身的架構(gòu)特點(diǎn)，做了深度的架構(gòu)優(yōu)化與設(shè)計(jì)，實(shí)現(xiàn)了移動云消息隊(duì)列RocketMQ與MQTT之間的消息流轉(zhuǎn)與規(guī)則管理。隨著萬物互聯(lián)的持續(xù)深入，未來消息隊(duì)列MQTT團(tuán)隊(duì)還將基于現(xiàn)在的架構(gòu)繼續(xù)優(yōu)化和創(chuàng)新，例如：

   1       ?  ?  ?  ?  ?

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

   2       ?  ?  ?  ?  ?

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

   3       ?  ?  ?  ?  ?

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