隨行付微服務之資料同步Porter

摘要： Porter是一款資料同步中介軟體，主要用於解決同構/異構資料庫之間的表級別資料同步問題。 背景 在微服務架構模式下深刻的影響了應用和資料庫之間的關係，不像傳統多個服務共享一個數據庫，微服務架構下每個服務都要有自己的資料庫。如果你想獲得微服務帶來的好處，每個服務獨有一個數據庫是必須的，...

Porter是一款資料同步中介軟體，主要用於解決同構/異構資料庫之間的表級別資料同步問題。

背景

在微服務架構模式下深刻的影響了應用和資料庫之間的關係，不像傳統多個服務共享一個數據庫，微服務架構下每個服務都要有自己的資料庫。如果你想獲得微服務帶來的好處，每個服務獨有一個數據庫是必須的，因為微服務強調的就是鬆耦合。我們希望資料庫就和服務一樣，要有充分的獨立性、可以和服務一起部署、一起擴充套件、一起重構。同時，還需要兼顧資料中心的資料聚合、DBA的多種資料庫備份、報表中心的業務報表等等矛盾問題。因此便產生了「Porter」專案。

微服務改造過程中，無法避免的一個坎，那就是垂直拆庫，根據不同的子服務，把過去的「一庫多服」拆分成「一庫一服」。

一庫一服還是一庫多服？

不管是否是微服務架構，應用的各個模組之間都需要頻繁的通訊、協作、共享資料，實現系統的整體價值。區別點在於單體應用是通過本地方法呼叫來完成；在微服務中是通過遠端API呼叫完成。

而共享資料最賤的方式就是採用共享資料庫模式，也就是單體應用中最常用的方式，一般只有一個數據庫，如圖一庫多服和一庫一服的方式:

一庫多服的架構模式通常會被認為是微服務架構下的反正規化，它的問題在於：

• 穩定性：單點故障，一個數據庫掛掉，整批服務全部停止。服務獨立性被扼殺？

• 耦合性：資料在一起，會給貪圖方便的開發或者DBA工程師編寫很多資料間高度依賴的程式或者工具；

• 擴充套件性：無法針對某一個服務進行精準優化或擴充套件，服務會大體分為兩個讀多寫少、寫多讀少，資料庫優化是根據服務而來的，不是一篇而論。

所以隨行付內部一般推薦的做法：是為每一個微服務準備一個單獨的資料庫，即一庫一服模式。這種模式更加適合微服務架構，它滿足每一個服務是獨立開發、獨立部署、獨立擴充套件的特性。當需要對一個服務進行升級或者資料架構改動的時候，無須影響到其他的服務。需要對某個服務進行擴充套件的時候，也可以手術式的對某一個服務進行區域性擴容。

那麼問題來了，在改造中我們發現，以下問題，誕生了該專案：

• 報表中心和前端詳細頁都存在SQL Join方式，經歷我們一庫一服的拆分後，無法在繼續使用SQL Join方式了...
• 資料中心，做得是資料聚合，資料拆分後，給資料中心帶來了很大的麻煩...
• 微服務之後，各個應用模組對資料庫的要求出現了分歧，資料庫型別多元化自主選擇還是統一...
• 等等...

Proter介紹

Porter是一個集中式的資料處理通道，所有的資料都在這個資料處理平臺匯聚、分發。Porter是一個無中心、外掛友好型分散式資料同步中介軟體。預設註冊中心外掛實現為zookeeper, 當然，你也可以基於註冊中心介面實現自定義註冊中心模組。在Porter的主流程外分佈著叢集外掛、源端消費外掛、源端訊息轉換器外掛、目標端寫入外掛、告警外掛、自定義資料定義外掛等外掛模組，除了叢集外掛、告警外掛是Porter任務節點全域性作用域外，其餘外掛模組都隨著同步任務的不同而相應組合。得益於良好的設計模式，Porter才能為大家呈現如此靈活的擴充套件性與易用性。

功能

Porter始於2017年，提供資料同步功能，但並不僅僅侷限於資料同步，在隨行付內部廣泛使用。主要提供一下功能：

• 原生支援Oracle|Mysql到Jdbc關係型資料庫最終一致同步
• 外掛友好化，支援自定義源端消費外掛、目標端載入外掛、告警外掛等外掛二次開發。
• 支援自定義源端、目標端表、欄位對映
• 支援節點基於配置檔案的同步任務配置。
• 支援管理後臺同步任務推送，節點、任務管理。提供任務執行指標監控，節點執行日誌、任務異常告警。
• 支援節點資源限流、分配。
• 基於Zookeeper叢集外掛的分散式架構。支援自定義叢集外掛。

架構設計

Porter節點通過註冊中心實現分散式叢集，並根據資源需求動態擴縮容。Portert與註冊中心協商了一套任務、節點、統計介面，Porter節點通過監聽註冊中心介面資料的變化實現任務的分配管理。配置管理後臺遵守並實現註冊中心的介面規範，實現對Porter節點遠端管理。註冊中心同樣有一套分散式鎖機制，用於任務資源的分配。

在這個機制外，Porter節點可以通過本地配置檔案的方式實現任務的定義。

原理介紹:

• 1、基於Canal開源產品，獲取MySql資料庫增量日誌資料。
• 2、管理系統架構。管理節點(web manager)管理工作節點任務編排、資料工作節點(TaskWork)彙報工作進度
• 3、基於Zookeeper叢集外掛的分散式架構。支援自定義叢集外掛
• 4、基於Kafka訊息元件，每張表對應一個Topic，資料節點分Topic消費工作

處理流程

為了保證資料的一致性，源端資料提取與目標端插入採用單執行緒順序執行，中間階段通過多執行緒執行提高資料處理速度。對照上圖就是SelectJob與LoadJob單執行緒執行，ExtractJob、TransformJob執行緒並行執行，然後在LoadJob階段對資料包進行排序，順序寫入目標端。

正如文章開頭所說，告警外掛與註冊中心外掛在多個任務間共享，每個任務根據源端與目標端的型別、源端資料格式選擇與之相匹配的處理外掛。也就是說告警外掛、註冊中心外掛與Porter節點配置相關，資料消費外掛、目標端外掛、自定義資料處理外掛等外掛與任務配置相關。

外掛化設計

Porter通過SPI規範結合單例、工廠、監聽者模式等設計模式，實現了極大的靈活性與鬆耦合，滿足不同場景的二次開發。具體涵蓋如下四個方面的外掛化設計：

• 註冊中心外掛
• 源端消費外掛
• 目標端載入外掛
• 自定義資料處理外掛

註冊中心外掛

在common包META-INF/spring.factories有如下內容:

cn.vbill.middleware.porter.common.cluster.ClusterProvider=
cn.vbill.middleware.porter.common.cluster.impl.zookeeper.ZookeeperClusterProvider
複製程式碼

摘抄ClusterProvider介面定義：

/**
* 匹配配置檔案指定的註冊中心實現
* @param type
* @return
*/
boolean matches(ClusterPlugin type);
複製程式碼

porter-boot的配置檔案對註冊中心的配置如下:

#叢集配置
porter.cluster.strategy=ZOOKEEPER
porter.cluster.client.url=127.0.0.1:2181
porter.cluster.client.sessionTimeout=100000
複製程式碼

看到這裡，有了配置檔案和外掛定義，我們還差使配置生效的程式碼。程式碼在Porter-boot的啟動類NodeBootApplication中:

//初始化叢集提供者中介軟體,spring spi外掛
try {
//獲取叢集配置資訊
ClusterProviderProxy.INSTANCE.initialize(config.getCluster());
} catch (Exception e) {
ClusterProviderProxy.INSTANCE.stop();
throw new RuntimeException("叢集配置引數ClusterConfig初始化失敗, 資料同步節點退出!error:" + e.getMessage());
}
複製程式碼

ClusterProviderProxy是一個單例列舉類，在initialize中根據spring.factories配置的實現類順序通過實現類的matches方法匹配配置檔案的配置:

List<ClusterProvider> providers = SpringFactoriesLoader.loadFactories(
ClusterProvider.class, JavaFileCompiler.getInstance());

for (ClusterProvider tmp : providers) {
if (tmp.matches(config.getStrategy())) {
tmp.start(config);
provider = tmp;
break;
}
}
複製程式碼

Porter節點通過註冊ClusterListener監聽感知註冊中心的通知事件，Porter的zookeeper實現在包porter-cluster裡,通過ZookeeperClusterMonitor啟用:

cn.vbill.middleware.porter.common.cluster.impl.zookeeper.ZookeeperClusterListener=\
cn.vbill.middleware.porter.cluster.zookeeper.ZKClusterTaskListener,\
cn.vbill.middleware.porter.cluster.zookeeper.ZKClusterNodeListener,\
cn.vbill.middleware.porter.cluster.zookeeper.ZKClusterStatisticListener,\
cn.vbill.middleware.porter.cluster.zookeeper.ZKClusterConfigListener
複製程式碼

源端消費外掛

本地配置檔案任務配置引數如下，指定了源端消費外掛，源端連線資訊，資料轉換外掛，目標端外掛等:

node.task[0].taskId=任務ID
node.task[0].consumer.consumerName=KafkaFetch #源端消費外掛
node.task[0].consumer.converter=oggJson #資料轉換外掛
node.task[0].consumer.source.sourceType=KAFKA
node.task[0].consumer.source.servers=127.0.0.1:9200
node.task[0].consumer.source.topics=kafka主題
node.task[0].consumer.source.group=消費組
node.task[0].consumer.source.oncePollSize=單次消費資料條數
node.task[0].consumer.source.pollTimeOut=100
node.task[0].loader.loaderName=JdbcBatch #目標端外掛
node.task[0].loader.source.sourceName=全域性目標端資料來源名字
複製程式碼

kafka消費外掛“KafkaFetch”定義在porter-plugin/kafka-consumer包，通過META-INF/spring.factories暴露實現:

cn.vbill.middleware.porter.core.consumer.DataConsumer=
cn.vbill.middleware.porter.plugin.consumer.kafka.KafkaConsumer
複製程式碼

通過消費器工廠類DataConsumerFactory查詢並激活,這裡的consumerName就是在配置檔案中配置的“KafkaFetch”

public DataConsumer newConsumer(String consumerName) throws DataConsumerBuildException {
for (DataConsumer t : consumerTemplate) {
if (t.isMatch(consumerName)) {
try {
return t.getClass().newInstance();
} catch (Exception e) {
LOGGER.error("%s", e);
throw new DataConsumerBuildException(e.getMessage());
}
}
}
return null;
}
複製程式碼

目標端載入外掛

要實現目標端載入外掛需要繼承cn.vbill.middleware.porter.core.consumer.AbstractDataConsumer

//為該外掛指定的外掛名稱，用於在任務配置中指定目標端外掛型別
protected String getPluginName();

//LoadJob階段執單執行緒執行，實際的目標端插入邏輯，插入物件通過mouldRow()在TransformJob構造
public Pair<Boolean, List<SubmitStatObject>> load(ETLBucket bucket) throws TaskStopTriggerException, InterruptedException;

//transform階段多執行緒並行執行,用於自定義處理資料行
public void mouldRow(ETLRow row) throws TaskDataException;

複製程式碼

完成自定義目標端外掛開發後，通過spring SPI機制釋出外掛

cn.vbill.middleware.porter.core.loader.DataLoader=\ cn.vbill.middleware.porter.plugin.loader.DemoLoader
複製程式碼

通過載入器工廠類DataLoaderFactory查詢並激活,這裡的loaderName就是在getPluginName()指定的外掛名稱

public DataLoader newLoader(String loaderName) throws DataLoaderBuildException {
for (DataLoader t : loaderTemplate) {
if (t.isMatch(loaderName)) {
try {
return t.getClass().newInstance();
} catch (Exception e) {
LOGGER.error("%s", e);
throw new DataLoaderBuildException(e.getMessage());
}
}
}
複製程式碼

那麼在任務配置時如何指定呢?看這裡:

porter.task[0].loader.loaderName=目標端外掛名稱
複製程式碼

實現細節參考porter-plugin包下的kafka-loader、jdbc-loader、kudu-loader三個目標端

自定義資料處理外掛

假設我們要將mysql表T_USER同步到目標端Oracle T_USER_2，源端表T_USER表結構與目標端表T_USER_2一致。我們的需求是隻保留FLAG欄位等於0的使用者資料。

需求有了，接下來我們就要實現EventProcessor介面做自定義資料過濾

package cn.vbill.middleware.porter.plugin;
public class UserFilter implements cn.vbill.middleware.porter.core.event.s.EventProcessor {
@Override
public void process(ETLBucket etlBucket) {
List<ETLRow> rows = etlBucket.getRows().stream().filter(r -> {
//第一步 找到表名為T_USER的記錄
boolean tableMatch = r.getFinalTable().equalsIgnoreCase("T_USER");
if (!tableMatch) return tableMatch;
//第二步 找到欄位FLAG的值不等於0的記錄
boolean columnMatch = r.getColumns().stream().filter(c -> c.getFinalName().equalsIgnoreCase("FLAG")
&& (null == c.getFinalValue() || !c.getFinalValue().equals("0"))).count() > 0;
return tableMatch && columnMatch;
}).collect(Collectors.toList());
//第三步 清除不符合條件的集合
etlBucket.getRows().removeAll(rows);
}
}

複製程式碼

在任務中指定自定義資料處理外掛:

porter.task[0].taskId=任務ID
porter.task[0].consumer.consumerName=CanalFetch
porter.task[0].consumer.converter=canalRow
porter.task[0].consumer.source.sourceType=CANAL
porter.task[0].consumer.source.slaveId=0
porter.task[0].consumer.source.address=127.0.0.1:3306
porter.task[0].consumer.source.database=資料庫
porter.task[0].consumer.source.username=賬號
porter.task[0].consumer.source.password=密碼
porter.task[0].consumer.source.filter=*.\.t_user
porter.task[0].consumer.eventProcessor.className=cn.vbill.middleware.porter.plugin.UserFilter
porter.task[0].consumer.eventProcessor.content=/path/UserFilter.class

porter.task[0].loader.loaderName=JdbcBatch #目標端外掛
porter.task[0].loader.source.sourceType=JDBC
porter.task[0].loader.source.dbType=ORACLE
porter.task[0].loader.source.url=jdbc:oracle:thin:@//127.0.0.1:1521/oracledb
porter.task[0].loader.source.userName=demo
porter.task[0].loader.source.password=demo

porter.task[0].mapper[0].auto=false
porter.task[0].mapper[0].table=T_USER,T_USER_2

複製程式碼

叢集機制

Porter的叢集模式依賴叢集外掛，預設的叢集外掛基於zookeeper實現。Porter任務節點和管理節點並不是強制繫結關係，任務部署可以通過任務配置檔案，也可以通過管理節點推送。管理節點還可以管理節點、收集、展示監控指標資訊等，是一個不錯的、簡化運維的管理平臺。同樣的，可以基於zookeeper資料結構協議實現你自己的管理平臺。

叢集模式下的系統結構：

zookeeper叢集模式外掛

zookeeper資料結構協議： {% asset_img zk_data_schema.png zk_data_schema.png %}

Porter的叢集機制主要有以下功能:

• 實現節點任務的負載，當前任務節點失效後自動漂移到其他任務節點
• 實現任務節點與管理節點的通訊
• 實現任務處理進度的儲存與拉取
• 實現統計指標資料的上傳(最新的開發版本支援自定義統計指標上傳客戶端，原生支援kafka)
• 用於節點、任務搶佔的分散式鎖實現

基於檔案系統的單機模式外掛

最新開發版支援Porter任務節點以單機模式執行,不依賴管理後臺和zookeeper，通過配置檔案配置任務。單機模式是一種特殊的叢集模式，僅支援部分叢集功能，但簡化了任務部署的複雜性，靈活多變。

• 實現任務處理進度的儲存與拉取
• 實現統計指標資料的上傳

Porter任務節點執行模式的配置方式

#zookeeper叢集配置
porter.cluster.strategy=ZOOKEEPER
porter.cluster.client.url=127.0.0.1:2181
porter.cluster.client.sessionTimeout=100000

#單機模式配置
porter.cluster.strategy=STANDALONE
porter.cluster.client.home=/path/.porter

複製程式碼

最後

筆者在開源Porter之前有幸參與apache skywalking社群並受其感召，隨後參與sharding-sphere等多個開源專案。我們竭盡所能提供優質的開源軟體，為中國的開源社群貢獻一份力量。但受限於技術能力及開源社群的運營經驗，不足之處，懇求大家的批評指正。

Porter的更多實現細節，請移步開源網站。

開源中國： ofollow,noindex">gitee.com/sxfad/porte…

GitHub： github.com/sxfad/porte…

Porter外掛開發demo： github.com/sxfad/porte…

Porter開源：835209101

本分類文章，與「隨行付研究院」微信號文章同步，第一時間接收公眾號推送，請關注「隨行付研究院」公眾號。
複製程式碼