張淵源,張琴燕,李 峰,姚祥龍

(1.浙江中醫(yī)藥大學(xué) 信息技術(shù)學(xué)院，浙江 杭州 310053；2.浙江大學(xué) 計(jì)算中心，浙江 杭州 310058；3.浙江大學(xué) 計(jì)算機(jī)科學(xué)與技術(shù)學(xué)院，浙江 杭州 315100；4.浙江省科技信息研究院，浙江 杭州 310006)

JTangSync分布式異構(gòu)數(shù)據(jù)同步系統(tǒng)的設(shè)計(jì)與實(shí)現(xiàn)

張淵源1,張琴燕2,3,李 峰4,姚祥龍3

(1.浙江中醫(yī)藥大學(xué) 信息技術(shù)學(xué)院，浙江 杭州 310053；2.浙江大學(xué) 計(jì)算中心，浙江 杭州 310058；3.浙江大學(xué) 計(jì)算機(jī)科學(xué)與技術(shù)學(xué)院，浙江 杭州 315100；4.浙江省科技信息研究院，浙江 杭州 310006)

集團(tuán)企業(yè)數(shù)據(jù)往往分布在各異構(gòu)環(huán)境中，但業(yè)務(wù)需求又要求企業(yè)進(jìn)行數(shù)據(jù)交換和同步。由于應(yīng)用平臺(tái)和數(shù)據(jù)模式各不相同，傳統(tǒng)的數(shù)據(jù)同步方式缺乏統(tǒng)一的同步框架。針對(duì)這一問(wèn)題，設(shè)計(jì)并實(shí)現(xiàn)了一種分布式異構(gòu)數(shù)據(jù)同步系統(tǒng)JTangSync。該系統(tǒng)采用分布式架構(gòu)，每個(gè)節(jié)點(diǎn)由數(shù)據(jù)源模塊、數(shù)據(jù)傳輸模塊和處理器模塊等組成。每個(gè)模塊都設(shè)計(jì)為可替換插件形式以便于二次擴(kuò)展，各個(gè)節(jié)點(diǎn)依賴Zookeeper組成集群，具有集中管理、故障轉(zhuǎn)移以及斷點(diǎn)續(xù)傳等多項(xiàng)功能。通過(guò)對(duì)比實(shí)驗(yàn)分析，證明該方法在異構(gòu)數(shù)據(jù)同步方面具有較好的效果。

分布式異構(gòu)數(shù)據(jù)集成；數(shù)據(jù)同步方式；JTangSync；數(shù)據(jù)傳輸；Zookeeper集群

0 引 言

處于起步期的企業(yè)，由于業(yè)務(wù)數(shù)量有限，數(shù)據(jù)很可能進(jìn)行集中式存儲(chǔ)。但隨著業(yè)務(wù)的不斷擴(kuò)展，數(shù)據(jù)模式將演變成為分布式數(shù)據(jù)[1]存儲(chǔ)于不同地區(qū)，并且容易導(dǎo)致數(shù)據(jù)的不一致性問(wèn)題[2]。因此，需要研究一套同步系統(tǒng)將不同地區(qū)的數(shù)據(jù)進(jìn)行有效整合，保證所有地區(qū)數(shù)據(jù)的一致性。

分布在不同地域的數(shù)據(jù)傳輸依賴于網(wǎng)絡(luò)，要求整個(gè)數(shù)據(jù)同步系統(tǒng)必須是分布式的，并且由于業(yè)務(wù)的復(fù)雜性、地區(qū)之間業(yè)務(wù)的差異性以及人為原因等，每個(gè)地區(qū)的數(shù)據(jù)很有可能是存儲(chǔ)在異構(gòu)系統(tǒng)中[3]。如A地區(qū)數(shù)據(jù)存儲(chǔ)在MySQL數(shù)據(jù)庫(kù)中，而B地區(qū)數(shù)據(jù)存儲(chǔ)在Oracle數(shù)據(jù)庫(kù)中，這兩種方式雖有相似之處，但是許多方面是不兼容的，在實(shí)現(xiàn)數(shù)據(jù)同步時(shí)需要屏蔽這種差異性。然而，即使數(shù)據(jù)存儲(chǔ)系統(tǒng)是一致的，也有可能出現(xiàn)數(shù)據(jù)庫(kù)名稱、存儲(chǔ)格式等差異，使得數(shù)據(jù)同步系統(tǒng)設(shè)計(jì)變得非常復(fù)雜[4]。

數(shù)據(jù)同步系統(tǒng)應(yīng)具有自動(dòng)化、實(shí)時(shí)性、異構(gòu)性、高可用性、高可靠性。自動(dòng)化是指整個(gè)系統(tǒng)在運(yùn)行時(shí)，基本不需要用戶干預(yù)，能夠自動(dòng)完成數(shù)據(jù)的同步。實(shí)時(shí)性是指處理數(shù)據(jù)的及時(shí)性，從數(shù)據(jù)的產(chǎn)生到處理耗費(fèi)的時(shí)間應(yīng)該在業(yè)務(wù)可承受范圍之內(nèi)，分鐘甚至毫秒。異構(gòu)性是指同步系統(tǒng)不僅支持多種多樣的數(shù)據(jù)源，而且還應(yīng)該支持?jǐn)?shù)據(jù)源的抽象和擴(kuò)展。數(shù)據(jù)源包括文件系統(tǒng)、Web網(wǎng)絡(luò)、FTP網(wǎng)絡(luò)、數(shù)據(jù)庫(kù)、ERP系統(tǒng)等。高可用性是指系統(tǒng)應(yīng)該能夠7*24運(yùn)行，在可用性上至少要達(dá)到4個(gè)9的目標(biāo)。因?yàn)閿?shù)據(jù)同步系統(tǒng)一般承擔(dān)企業(yè)級(jí)的核心數(shù)據(jù)和業(yè)務(wù)流程，對(duì)可用性上有很高的要求。高可靠性是指在運(yùn)行時(shí)可能會(huì)遇到各種突發(fā)情形(比如數(shù)據(jù)庫(kù)連接斷開、網(wǎng)絡(luò)斷開、IP地址改變等環(huán)境因素)導(dǎo)致同步出錯(cuò)或者不能繼續(xù)進(jìn)行，數(shù)據(jù)同步系統(tǒng)應(yīng)該能夠檢測(cè)同步出錯(cuò)，讓系統(tǒng)在環(huán)境恢復(fù)時(shí)重新進(jìn)行同步[5-6]。

工業(yè)界對(duì)同步數(shù)據(jù)問(wèn)題的解決方案大致有兩種，第一種是專門開發(fā)一種系統(tǒng)用來(lái)滿足自己的業(yè)務(wù)邏輯，可以支持自身要求的數(shù)據(jù)格式，最多支持同一種數(shù)據(jù)格式系統(tǒng)的擴(kuò)展。這一類系統(tǒng)以數(shù)據(jù)庫(kù)增量實(shí)時(shí)同步系統(tǒng)為代表。第二種是基于流處理框架開發(fā)，在大數(shù)據(jù)時(shí)代傳統(tǒng)的計(jì)算模式己經(jīng)不能適應(yīng)現(xiàn)在的大數(shù)據(jù)量和數(shù)據(jù)多樣性，多種計(jì)算模式使得流計(jì)算又重新煥發(fā)了生機(jī)，并且隨著云計(jì)算的普及，數(shù)據(jù)的流計(jì)算增加了許多新的特征，比如高擴(kuò)展性、分布式支持等等[7]。由于數(shù)據(jù)同步實(shí)際上是流處理的一種，用戶在流處理之上開發(fā)數(shù)據(jù)同步需要自己解決斷點(diǎn)續(xù)傳、增量數(shù)據(jù)獲取、分布式等難題，所以這種方法雖然擴(kuò)展性更高，但是開發(fā)成本也更高[7-10]。

基于此，文中設(shè)計(jì)了一種異構(gòu)數(shù)據(jù)同步系統(tǒng)JTangSync。與傳統(tǒng)方法相比，用戶基于JTangSync提供的接口可以定義自己的同步邏輯，然后快速開發(fā)期望數(shù)據(jù)格式的系統(tǒng)，并且不需要考慮斷點(diǎn)續(xù)傳、故障恢復(fù)等問(wèn)題[11]，此外監(jiān)控管理平臺(tái)讓用戶省去開發(fā)自己的監(jiān)控管理平臺(tái)。JTangSync專門解決數(shù)據(jù)同步問(wèn)題，尤其是數(shù)據(jù)庫(kù)同步的問(wèn)題。設(shè)計(jì)了大量的異構(gòu)數(shù)據(jù)源和處理程序，用戶開發(fā)一般的數(shù)據(jù)同步只需按照指導(dǎo)在監(jiān)控管理平臺(tái)進(jìn)行配置然后啟動(dòng)即可。

1 相關(guān)技術(shù)

數(shù)據(jù)同步系統(tǒng)設(shè)計(jì)和實(shí)現(xiàn)的相關(guān)技術(shù)，包括Zookeeper，Protocol Buffer，Snappy，Netty，Jetty，F(xiàn)ast JSON等等。

1.1 Zookeeper

JTangSync使用Zookeeper來(lái)管理配置信息，包括所有的數(shù)據(jù)源配置、數(shù)據(jù)處理器配置、監(jiān)控實(shí)例配置、節(jié)點(diǎn)信息和Manager的選舉[12]。

ZooKeeper源于Hadoop的一個(gè)子項(xiàng)目，是一個(gè)應(yīng)用于大型分布式系統(tǒng)的協(xié)調(diào)系統(tǒng)，功能包括分布式同步、配置維護(hù)、名字服務(wù)等。ZooKeeper封裝復(fù)雜易出錯(cuò)的服務(wù)，提供可靠易用的接口和高效穩(wěn)定的系統(tǒng)。一個(gè)分布式的環(huán)境需要Master實(shí)例來(lái)存儲(chǔ)一些配置信息，使用Zookeeper來(lái)維護(hù)信息和配置信息等，確保寫入文件的一致性。

1.2 Netty

JTangSync使Netty作為其核心組件，實(shí)現(xiàn)分布式的網(wǎng)絡(luò)程序。Netty是一個(gè)著名的開源NIO框架，主要功能是提供異步的、事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用程序接口，使用Netty能夠快速開發(fā)高性能、高可靠性的網(wǎng)絡(luò)服務(wù)器和客戶端程序[13]。

使用Netty可以確保網(wǎng)絡(luò)應(yīng)用的開發(fā)過(guò)程更加簡(jiǎn)單和快速，例如實(shí)現(xiàn)了HTTP協(xié)議的客戶和服務(wù)端應(yīng)用。Netty大大簡(jiǎn)化和流線化了網(wǎng)絡(luò)應(yīng)用的編程開發(fā)過(guò)程，例如TCP和UDP的Socket服務(wù)開發(fā)。在設(shè)計(jì)上Netty借鑒各協(xié)議的經(jīng)驗(yàn)，包括FTP、SMTP、HTTP、二進(jìn)制、文本協(xié)議等。

1.3 Protocol Buffer

JTangSync設(shè)計(jì)了網(wǎng)絡(luò)數(shù)據(jù)的交換過(guò)程，如數(shù)據(jù)傳輸、節(jié)點(diǎn)間通信、性能監(jiān)控等等。由于Protocol Buffer在Netty中存在相關(guān)的編碼器和解碼器，使用Protocol Buffer作為JTangSync中的數(shù)據(jù)標(biāo)準(zhǔn)的交換格式[14]。

Protocol Buffer是Google的一種數(shù)據(jù)交換格式，它獨(dú)立于語(yǔ)言和平臺(tái)，提供了三種實(shí)現(xiàn)語(yǔ)言：Java、C++和Python。每一種實(shí)現(xiàn)過(guò)程都包含了相應(yīng)語(yǔ)言的編譯器以及庫(kù)文件。由于它是二進(jìn)制的格式，比使用XML進(jìn)行數(shù)據(jù)傳輸更快，因此可以把它用于分布式應(yīng)用之間的數(shù)據(jù)通信或者異構(gòu)環(huán)境下的數(shù)據(jù)交換。作為一種效率和兼容性都很優(yōu)秀的二進(jìn)制數(shù)據(jù)傳輸格式，可以廣泛運(yùn)用于諸如網(wǎng)絡(luò)傳輸、配置文件、數(shù)據(jù)存儲(chǔ)等諸多領(lǐng)域之中。

1.4 Snappy

為提高效率，JTangSync中網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)需要壓縮，但是傳統(tǒng)的壓縮算法耗時(shí)較長(zhǎng)。該系統(tǒng)使用Snappy算法作為網(wǎng)絡(luò)傳輸?shù)臉?biāo)準(zhǔn)壓縮算法。Snappy是Zippy的開源實(shí)現(xiàn)。它作為一個(gè)壓縮庫(kù)被Google用于許多內(nèi)部項(xiàng)目中，包括MapReduce,RPC和BigTable。Snappy優(yōu)化了64位x86處理器，經(jīng)測(cè)試，Snappy在單個(gè)Intel Core i7處理器內(nèi)核上壓縮速度為每秒250 MB，解壓速率為每秒500 MB。

1.5 Fast JSON

JTangSync使用JSON描述數(shù)據(jù)交換格式。JSON是基于JavaScript的一個(gè)子集，采用完全獨(dú)立于語(yǔ)言的文本格式，使得JSON成為理想的數(shù)據(jù)交換語(yǔ)言，既易于人類閱讀和編寫，又易于機(jī)器解析和生成。

Fast JSON是阿里巴巴公司開發(fā)的一個(gè)JSON開發(fā)庫(kù)[15]。與傳統(tǒng)的JSON解析相比，F(xiàn)ast JSON的解析速度更快，并且Fast JSON的API設(shè)計(jì)比較合理，包含多種Java對(duì)象與JSON轉(zhuǎn)化的方法。其中，最大的優(yōu)點(diǎn)在于可以與Java Bean進(jìn)行相互轉(zhuǎn)化。在JTangSync中，F(xiàn)ast JSON主要用于Web程序的數(shù)據(jù)交互、集群間節(jié)點(diǎn)的性能數(shù)據(jù)交換等方面。

2 數(shù)據(jù)同步系統(tǒng)設(shè)計(jì)

同步系統(tǒng)是將數(shù)據(jù)從一個(gè)系統(tǒng)、軟件或者文件等存儲(chǔ)中提取出來(lái)并且通過(guò)網(wǎng)絡(luò)連接、郵件等方式傳輸?shù)搅硪粋€(gè)不同或者相同的系統(tǒng)中，最終將數(shù)據(jù)導(dǎo)入到目標(biāo)系統(tǒng)的存儲(chǔ)中的自動(dòng)化系統(tǒng)。

如圖1所示，節(jié)點(diǎn)中運(yùn)行的功能單位是同步實(shí)例，代表一個(gè)從數(shù)據(jù)源到數(shù)據(jù)處理器的完整邏輯。一個(gè)同步實(shí)例包含一個(gè)數(shù)據(jù)源、一個(gè)由0到多個(gè)數(shù)據(jù)處理器組成的數(shù)據(jù)源處理器鏈、一個(gè)由0到多個(gè)數(shù)據(jù)處理器組成的本地處理器鏈。

本節(jié)將分別從數(shù)據(jù)源模塊、網(wǎng)絡(luò)傳輸模塊、數(shù)據(jù)處理模塊和集群模塊來(lái)介紹JTangSync異構(gòu)數(shù)據(jù)同步系統(tǒng)。

圖1 系統(tǒng)拓?fù)浣Y(jié)構(gòu)

2.1 數(shù)據(jù)源模塊

數(shù)據(jù)源是指JTangSync獲取數(shù)據(jù)源的一種代理器。由于一般的同步系統(tǒng)中數(shù)據(jù)源與處理邏輯并不在同一個(gè)域中，安全性要求數(shù)據(jù)源從外界進(jìn)行訪問(wèn)時(shí)必須通過(guò)代理來(lái)獲取數(shù)據(jù)。此外，數(shù)據(jù)來(lái)源多種多樣，如MySQL數(shù)據(jù)源、SQL Server數(shù)據(jù)源、Oracle數(shù)據(jù)源、文件數(shù)據(jù)源等。數(shù)據(jù)源的主要職責(zé)就是獲取數(shù)據(jù)并且通過(guò)網(wǎng)絡(luò)轉(zhuǎn)發(fā)出去。數(shù)據(jù)源模塊就是管理數(shù)據(jù)的功能模塊。

2.2 數(shù)據(jù)源工作流程

數(shù)據(jù)源的主要職責(zé)就是獲取數(shù)據(jù)并且轉(zhuǎn)發(fā)到數(shù)據(jù)傳輸模塊。數(shù)據(jù)源管理器管理所有的數(shù)據(jù)源，數(shù)據(jù)源管理器接收到初始化數(shù)據(jù)，然后查找是否存在此類型的數(shù)據(jù)源。如果找到，則初始化數(shù)據(jù)源，數(shù)據(jù)源進(jìn)入工作狀態(tài)。在此過(guò)程中，如果發(fā)生任何異?；蝈e(cuò)誤，數(shù)據(jù)源退出。最后重新初始化數(shù)據(jù)源是同步實(shí)例的職責(zé)，而不是數(shù)據(jù)源的職責(zé)。

2.3 網(wǎng)絡(luò)傳輸模塊

網(wǎng)絡(luò)傳輸模塊的主要職責(zé)是將數(shù)據(jù)盡量及時(shí)轉(zhuǎn)發(fā)出去。實(shí)現(xiàn)難點(diǎn)包括延遲低、數(shù)據(jù)速率高。延遲低是指數(shù)據(jù)到達(dá)網(wǎng)絡(luò)傳輸模塊之后，需要盡快地傳輸出去，以及時(shí)將數(shù)據(jù)提交到對(duì)端。

數(shù)據(jù)速率高是指網(wǎng)絡(luò)傳輸?shù)乃俣纫M量快。當(dāng)數(shù)據(jù)包比較小需要合并時(shí)，這兩者是沖突的，因?yàn)橛捎贗O接口只能支持有限的調(diào)用次數(shù)(現(xiàn)今大約為數(shù)百次/s)，所以必須在每次調(diào)用IO接口時(shí)傳輸盡量多的數(shù)據(jù)。由于數(shù)據(jù)包較小，那么每秒發(fā)送的數(shù)據(jù)量會(huì)很小。因此，必須進(jìn)行緩存。

2.4 數(shù)據(jù)傳輸流程

數(shù)據(jù)傳輸模塊處理數(shù)據(jù)源模塊產(chǎn)生的數(shù)據(jù)。由于IO設(shè)備的限制，每秒的IO次數(shù)是一定的，數(shù)據(jù)要先進(jìn)入緩存。如果數(shù)據(jù)量達(dá)到或超過(guò)限制，或者超時(shí)(超時(shí)限制設(shè)置為200 ms)，將數(shù)據(jù)序列化方便網(wǎng)絡(luò)傳輸[16]。序列化方法要求序列化和反序列化的速度要快，不能影響網(wǎng)絡(luò)傳輸?shù)男?。文中以Protocol Buffer作為序列化技術(shù)。

接著分析當(dāng)前的數(shù)據(jù)是否需要壓縮，判斷策略主要通過(guò)預(yù)測(cè)算法，通過(guò)過(guò)去的數(shù)據(jù)預(yù)測(cè)當(dāng)前時(shí)間點(diǎn)的情況，如果過(guò)去1小時(shí)之內(nèi)使用壓縮傳輸數(shù)據(jù)的平均速度為mkB/s，不使用壓縮傳輸數(shù)據(jù)的平均速度為MkB/s，那么有以下策略：

上式的意義為，當(dāng)m>M，即壓縮的數(shù)據(jù)傳輸速度大于不壓縮，則將當(dāng)前的壓縮模式切換為非壓縮模式。由于Snappy算法在壓縮和解壓縮方面速度遠(yuǎn)勝于GZIP庫(kù)，所以JTangSync在網(wǎng)絡(luò)傳輸模塊中采用Snappy算法。

2.5 數(shù)據(jù)處理模塊

數(shù)據(jù)模塊是管理和配置、執(zhí)行數(shù)據(jù)處理器的模塊。用戶的數(shù)據(jù)處理需求比較復(fù)雜，單一的數(shù)據(jù)處理器不能滿足用戶的全部需求，采用責(zé)任鏈模式來(lái)定義用戶要求的數(shù)據(jù)處理邏輯。此外，如果JTangSync中提供的數(shù)據(jù)處理器不滿足用戶的特殊需求，用戶可以通過(guò)JTangSync數(shù)據(jù)處理開發(fā)接口開發(fā)自己的數(shù)據(jù)處理器。

2.6 JTangSync數(shù)據(jù)處理流程

數(shù)據(jù)處理模塊是處理數(shù)據(jù)的主要模塊，JTangSync中處理數(shù)據(jù)的一個(gè)功能單元模塊，又稱Handler。每一種數(shù)據(jù)處理邏輯在JTangSync中都對(duì)應(yīng)一個(gè)數(shù)據(jù)處理器，多個(gè)數(shù)據(jù)處理器可以配置為一個(gè)數(shù)據(jù)處理器鏈，數(shù)據(jù)處理器鏈?zhǔn)菃蜗虿⑶也豢裳h(huán)的。其中數(shù)據(jù)處理器是數(shù)據(jù)處理模塊的組成單元。當(dāng)配置一個(gè)同步實(shí)例時(shí)，用戶需要配置若干數(shù)據(jù)處理器，這些數(shù)據(jù)處理器按照配置的先后順序組成一個(gè)責(zé)任鏈來(lái)處理這一個(gè)同步實(shí)例的數(shù)據(jù)。

數(shù)據(jù)處理器在使用之前需要初始化，這一過(guò)程可能需要配置信息，然后數(shù)據(jù)處理器開始處理數(shù)據(jù)。初始化所需配置信息統(tǒng)一為Key-Value格式，配置信息數(shù)據(jù)類型為String。

數(shù)據(jù)源處理器鏈由0到多個(gè)數(shù)據(jù)處理器組成，數(shù)據(jù)源會(huì)將數(shù)據(jù)提交給處理器鏈然后進(jìn)行網(wǎng)絡(luò)傳輸。本地處理器鏈由0到多個(gè)數(shù)據(jù)處理器組成，可以從網(wǎng)絡(luò)接收數(shù)據(jù)并且進(jìn)行處理。為與數(shù)據(jù)源處理器鏈相區(qū)別，稱為本地處理器鏈。一個(gè)同步實(shí)例中數(shù)據(jù)處理器需要在兩個(gè)階段運(yùn)行，第一個(gè)階段在網(wǎng)絡(luò)傳輸之前進(jìn)行，主要進(jìn)行數(shù)據(jù)過(guò)濾、合并、統(tǒng)計(jì)等，稱為數(shù)據(jù)源處理器。第二個(gè)階段是在網(wǎng)絡(luò)傳輸之后進(jìn)行，主要責(zé)任是處理數(shù)據(jù)，稱為本地處理器。兩個(gè)階段的處理器是共用的，接口設(shè)計(jì)完全相同，體現(xiàn)了JTangSync的高度靈活性。

2.7 JTangSync集群模塊

集群模塊是管理和控制JTangSync集群的模塊，主要功能有節(jié)點(diǎn)管理、負(fù)載均衡、故障轉(zhuǎn)移和狀態(tài)監(jiān)控、Manager選舉、集群間通信、本地同步實(shí)例管理等。在管理控制平臺(tái)，用戶可以實(shí)現(xiàn)JTangSync集群管理、數(shù)據(jù)源管理、性能監(jiān)控、處理器管理等所有目標(biāo)。

2.8 JTangSync集群流程

JTangSync設(shè)計(jì)為集群模式，多個(gè)JTangSync節(jié)點(diǎn)通過(guò)JTangSync集群模塊可以組成一個(gè)JTangSync集群。節(jié)點(diǎn)指一個(gè)JTangSync進(jìn)程，是指JTangSync集群中的一個(gè)資源單位，一個(gè)JTangSync集群中可能包含1到多個(gè)JTangSync節(jié)點(diǎn)，同步實(shí)例可以運(yùn)行在其中任何一個(gè)節(jié)點(diǎn)上。Node的定義與節(jié)點(diǎn)定義相同。JTangSync集群模塊包含JTangSync集群運(yùn)行的必需功能。

文中使用Zookeeper實(shí)現(xiàn)JTangSync集群模塊，提供一致的分布式事件回調(diào)接口，使用方便。樹形目錄結(jié)構(gòu)與當(dāng)前系統(tǒng)保持一致，并且樹形結(jié)構(gòu)目錄的每個(gè)節(jié)點(diǎn)都提供服務(wù)，這滿足了分布式系統(tǒng)中的配置信息保存的需求[17-18]。

3 JTangSync節(jié)點(diǎn)實(shí)現(xiàn)

3.1 數(shù)據(jù)源的實(shí)現(xiàn)

數(shù)據(jù)源是JTangSync中的核心組件，數(shù)據(jù)源的質(zhì)量和數(shù)量決定了JTangSync是否強(qiáng)大。數(shù)據(jù)庫(kù)增量數(shù)據(jù)源是JTangSync核心中的核心，是指當(dāng)前提取的數(shù)據(jù)都是數(shù)據(jù)庫(kù)在一段時(shí)間之內(nèi)的增量數(shù)據(jù)，與數(shù)據(jù)庫(kù)數(shù)據(jù)源是不一樣的概念。數(shù)據(jù)庫(kù)數(shù)據(jù)源只是簡(jiǎn)單的連接數(shù)據(jù)庫(kù)，然后獲取其中的全量數(shù)據(jù)，實(shí)現(xiàn)的難度較數(shù)據(jù)庫(kù)增量數(shù)據(jù)源低。數(shù)據(jù)庫(kù)增量數(shù)據(jù)源的實(shí)現(xiàn)難度大，對(duì)效率、延遲的要求都會(huì)比其他的數(shù)據(jù)源高。

MySQL數(shù)據(jù)源實(shí)現(xiàn)是基于MySQL本身的Replication協(xié)議，Replication是專門用于MySQL之間復(fù)制的協(xié)議。Replication協(xié)議中需要同步數(shù)據(jù)的MySQL稱為Slave，提供增量數(shù)據(jù)的稱為Master，數(shù)據(jù)傳輸基于Slave和Master之間的TCP連接，Master中增量數(shù)據(jù)存儲(chǔ)的位置是binary log(二進(jìn)制日志)，存儲(chǔ)格式有三種:statement,row,mixed。在JTangSync中選擇Row格式作為MySQL增量數(shù)據(jù)源的二進(jìn)制日志格式。

Sqlserver的增量數(shù)據(jù)源使用Sqlserver的變更數(shù)據(jù)捕獲功能(又稱為Change Data Capture,CDC),Sqlserver的變更數(shù)據(jù)捕獲功能有一些限制。首先CDC有版本限制，只有SQL server2008以及以上版本才支持。其次CDC功能需要為每個(gè)表開啟一個(gè)臨時(shí)表。盡管存在這些限制，比起觸發(fā)器方法還是有很多優(yōu)勢(shì)。

3.2 數(shù)據(jù)傳輸?shù)膶?shí)現(xiàn)

數(shù)據(jù)傳輸基于Netty實(shí)現(xiàn)。Netty不僅是優(yōu)秀的NIO和多線程框架，而且提供了許多協(xié)議庫(kù)，其中包括snappy和protocol buffer的實(shí)現(xiàn)。數(shù)據(jù)傳輸代碼如下：

ChannelPipeline pipeline=ch.pipeline();

ch.pipeline().addLast(new MySnappyDecoder());

ch.pipeline().addLast(new MySnappyEncoder());

ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());

ch.pipeline().addLast(new ProtobufDecoder(SyncEvent.Event.getDefaultInstance()));

ch.pipeline().addLast(new ProtobufVarint32LengthField Prerpender());

ch.pipeline().addLast(new ProtobufEncoder());

pipeline.addLast(“handler”, new SnappyProtobufServerHandler(ch));

自適應(yīng)壓縮算法的主要思想是在Netty中實(shí)現(xiàn)一個(gè)專門用于統(tǒng)計(jì)網(wǎng)絡(luò)傳輸字節(jié)數(shù)的的handler，并且向上層提供接口，上層邏輯可以使用這個(gè)接口實(shí)現(xiàn)自適應(yīng)壓縮算法。例如，此handler統(tǒng)計(jì)到己經(jīng)發(fā)送了n1字節(jié)，經(jīng)過(guò)ts之后變成n2字節(jié)，ts內(nèi)的發(fā)送速度為(n2-n1)/t/1 024，單位為kB/s，相比之前的速度即可決定是否需要進(jìn)行壓縮算法。

3.3 數(shù)據(jù)處理器的實(shí)現(xiàn)

過(guò)濾處理器是指根據(jù)一定規(guī)則過(guò)濾數(shù)據(jù)的處理器，這里的規(guī)則主要有相等、不等、大于、小于、大于等于、小于等于等規(guī)則。過(guò)濾處理器使用的范圍是在數(shù)據(jù)庫(kù)同步領(lǐng)域，比如當(dāng)傳輸數(shù)據(jù)庫(kù)數(shù)據(jù)時(shí)，用戶可能需要過(guò)濾掉多余數(shù)據(jù)。

數(shù)據(jù)庫(kù)處理器是將數(shù)據(jù)導(dǎo)入數(shù)據(jù)庫(kù)的處理器。由于不同的數(shù)據(jù)庫(kù)的SQL標(biāo)準(zhǔn)是不一樣的，所以每個(gè)數(shù)據(jù)庫(kù)對(duì)應(yīng)一個(gè)數(shù)據(jù)庫(kù)處理器。數(shù)據(jù)庫(kù)處理器需要考慮：首先，如果是關(guān)系型數(shù)據(jù)庫(kù)，那么要將數(shù)據(jù)的提交格式設(shè)置為false并且執(zhí)行一定數(shù)量的SQL之后提交；如果是非關(guān)系型數(shù)據(jù)庫(kù)，那么要盡量減少IO的次數(shù)。其次，數(shù)據(jù)傳輸時(shí)要盡量打包，減少網(wǎng)絡(luò)IO的次數(shù)。最后，如果通過(guò)JDBC執(zhí)行SQL，需要使用prepared statement，因?yàn)閿?shù)據(jù)只有insert、delete、update三種，可以將三個(gè)prepared statement緩存起來(lái)，等待重復(fù)使用。實(shí)驗(yàn)證明，性能至少提高了10%～20%。

4 JTangSync集群實(shí)現(xiàn)

JTangSync集群中存在多個(gè)節(jié)點(diǎn)，需要一個(gè)協(xié)調(diào)者擔(dān)任Manager的角色。這個(gè)Manager節(jié)點(diǎn)的主要任務(wù)是啟動(dòng)Web監(jiān)控管理平臺(tái)和協(xié)調(diào)集群運(yùn)行。

4.1 Manager選舉算法

JTangSync中選舉算法的流程如圖2所示。

圖2 JTangSync中選舉算法的流程圖

1)JTangSync在zookeeper目錄中的//nodes/目錄下新建一個(gè)類型為EPHEMERAL_SEQUENTIAL的node,它是zookeeper中的順序內(nèi)存node。SEQUENTIAL保證了創(chuàng)建此node的目錄下所有的SEQUENTIAL類型的node名稱都是順序的，名稱由zookeeper決定；EPHEMERAL類型node保證了此節(jié)點(diǎn)只存在于連接有效的時(shí)間段內(nèi)，如果創(chuàng)建此node的連接失效，那么zookeeper將自動(dòng)把此node刪除。EPHEMERAL_SEQUENTIAL類型就是以上兩種類型的集合。

2)JTangSync在/nodes目錄下新建一個(gè)Watcher。Watcher是zookeeper中的一種分布式事件通知器，如果Watcher所在的node的數(shù)據(jù)發(fā)生變化(包括刪除、修改)或者node下的節(jié)點(diǎn)發(fā)生變化(包括刪除、新建)，Watcher就會(huì)得到通知，并且執(zhí)行Watcher中的回調(diào)方法。

3)如果當(dāng)前Watcher被調(diào)用，那么分兩種情況:

如果nodes下節(jié)點(diǎn)數(shù)量達(dá)到配置中JTangSync集群中節(jié)點(diǎn)的數(shù)量并且當(dāng)前還未啟動(dòng)完成，那么將獲取當(dāng)前nodes下所有的節(jié)點(diǎn)并且排序，名稱最小的節(jié)點(diǎn)將成為manager節(jié)點(diǎn)。Manager節(jié)點(diǎn)將會(huì)完成以下操作:

(1)啟動(dòng)當(dāng)前節(jié)點(diǎn)的web監(jiān)控管理控制平臺(tái)。

(2)分配所有標(biāo)志為“自動(dòng)啟動(dòng)”的同步實(shí)例到JTangSync節(jié)點(diǎn)上運(yùn)行。

如果nodes下節(jié)點(diǎn)數(shù)量沒(méi)有達(dá)到JTangSync集群中節(jié)點(diǎn)的數(shù)量，并且當(dāng)前還未啟動(dòng)完成，那么就返回。

如果當(dāng)前JTangSync集群標(biāo)志為已啟動(dòng)，并且當(dāng)前nodes下的node數(shù)量有變化，那么更新manager，新的manager需要完成以下操作:

(1)啟動(dòng)當(dāng)前節(jié)點(diǎn)的web監(jiān)控管理控制平臺(tái)。

(2)啟動(dòng)故障轉(zhuǎn)移模塊。

JTangSync的選舉非常靈活，利用zookeeper不僅能夠快速準(zhǔn)確地選舉出manager節(jié)點(diǎn)，并且利用zookeeper實(shí)現(xiàn)了在過(guò)去manager節(jié)點(diǎn)失效的情況下選舉新的manager，實(shí)現(xiàn)了分布式系統(tǒng)中的高可靠性。

4.2 節(jié)點(diǎn)熱部署

利用Zookeeper，還可以實(shí)現(xiàn)在運(yùn)行時(shí)新增節(jié)點(diǎn)。新增節(jié)點(diǎn)必須在整個(gè)JTangSync集群己經(jīng)啟動(dòng)的情況下進(jìn)行(如圖3所示)。

圖3 增節(jié)點(diǎn)的流程

從圖3中看到，節(jié)點(diǎn)的熱部署[19-20]大致分為2步:

(1)啟動(dòng)，并且連接Zookeeper，與選舉算法中的描述相同，在nodes下新建watcher，在nodes下新建EPHEMERAL SEQUENTIAL類型node。

(2)此時(shí)，其他watcher都會(huì)收到通知，并且更新本地所有的node信息。

4.3 節(jié)點(diǎn)熱移除

借助Zookeeper的分布式通知功能，可以實(shí)現(xiàn)JTangSync節(jié)點(diǎn)的熱移除功能。JTangSync集群中節(jié)點(diǎn)的移除流程如圖4所示。

由于節(jié)點(diǎn)在Zookeeper中建立的是EPHEMERAL類型的節(jié)點(diǎn)，停止之后，zookeeper中建立的節(jié)點(diǎn)消失，/nodes/節(jié)點(diǎn)下node數(shù)量變化，其他的JTangSync節(jié)點(diǎn)都會(huì)收到通知，最后更新manager和本地的node信息。

5 系統(tǒng)性能測(cè)試

5.1 操作類型的支持

測(cè)試JTangSync在MySQL數(shù)據(jù)庫(kù)insert,update,delete語(yǔ)句。測(cè)試數(shù)據(jù)共426 570條，包括146 570條插入數(shù)據(jù)，140 000條刪除數(shù)據(jù)，以及140 000條更新數(shù)據(jù)。測(cè)試環(huán)境為內(nèi)網(wǎng)環(huán)境。測(cè)試見(jiàn)表1。

圖4 節(jié)點(diǎn)熱移除的流程

測(cè)試類型花費(fèi)時(shí)間/s性能/(條/s)insert207328delete20.36896update226363

5.2 模塊性能測(cè)試

測(cè)試JTangSync的數(shù)據(jù)源的數(shù)據(jù)捕獲性能、數(shù)據(jù)發(fā)送模塊性能、數(shù)據(jù)處理器性能和內(nèi)存占用。測(cè)試配置從Master到Slave，同步426 570條數(shù)據(jù)。Slave自身性能測(cè)試先使用獨(dú)立的20萬(wàn)條插入數(shù)據(jù)，之后與整個(gè)系統(tǒng)的性能比較，得出是否己經(jīng)達(dá)到了整個(gè)系統(tǒng)的瓶頸。

性能測(cè)試結(jié)果如表2所示。

表2 性能測(cè)試結(jié)果

數(shù)據(jù)傳輸模塊的瓶頸在16 000條/s左右，與數(shù)據(jù)源性能相差如此大的原因是因?yàn)樾枰獙?shù)據(jù)源模塊獲取的數(shù)據(jù)轉(zhuǎn)化為JTangSync的統(tǒng)一數(shù)據(jù)格式，這耗費(fèi)了很多時(shí)間。另一方面是JTangSync己經(jīng)達(dá)到了整個(gè)系統(tǒng)的性能瓶頸，JTangSync的性能最高達(dá)到6 874,Salve的性能瓶頸在7 104，除去JTangSync在數(shù)據(jù)傳輸、壓縮、轉(zhuǎn)換的消耗，JTangSync在資源的利用和性能方面做的比較好。

6 結(jié)束語(yǔ)

JTangSync是異構(gòu)數(shù)據(jù)同步框架，主要功能是為用戶提供一個(gè)高可用、高性能、高可擴(kuò)展的數(shù)據(jù)同步開發(fā)平臺(tái)。用戶使用這個(gè)框架無(wú)需關(guān)心數(shù)據(jù)同步中常見(jiàn)的數(shù)據(jù)格式、數(shù)據(jù)傳輸、斷點(diǎn)續(xù)傳、故障恢復(fù)等問(wèn)題，只需要考慮基本的業(yè)務(wù)邏輯和數(shù)據(jù)源、數(shù)據(jù)處理器的開發(fā)即可。JTangSync內(nèi)置了大量的數(shù)據(jù)源和數(shù)據(jù)處理器，免去用戶重復(fù)開發(fā)一些常用數(shù)據(jù)源和數(shù)據(jù)處理器的煩惱。從測(cè)試結(jié)果上看，JTangSync是一個(gè)優(yōu)秀的異構(gòu)數(shù)據(jù)同步框架。

未來(lái)，由于數(shù)據(jù)同步的需求多種多樣，JTangSync將整合用戶的各種需求，開發(fā)新的數(shù)據(jù)源和數(shù)據(jù)處理器，并加入新的數(shù)據(jù)同步邏輯。

[1] Leslie B,Ron B,Charles D L T,et al.Formulary and database synchronization[J].American Journal of Health-System Pharmacy,2011,68(3):204-206.

[2] 潘群華,吳秋云,陳宏盛.分布式數(shù)據(jù)庫(kù)系統(tǒng)中數(shù)據(jù)一致性維護(hù)方法[J].計(jì)算機(jī)工程,2002,28(9):255-257.

[3] 戴婉榮.基于分布式數(shù)據(jù)庫(kù)的數(shù)據(jù)同步機(jī)制的研究與應(yīng)用[D].武漢:武漢理工大學(xué)，2010.

[4] Liao Guoqiong.Data synchronization and resynchronization for heterogeneous databases replication in middleware-based architecture[J].Journal of Networks,2012,7(1):210-217.

[5] 魏祥麟,陳 鳴,范建華,等.數(shù)據(jù)中心網(wǎng)絡(luò)的體系結(jié)構(gòu)[J].軟件學(xué)報(bào),2013,24(2):295-316.

[6] 章 筠.計(jì)算機(jī)網(wǎng)絡(luò)可靠性分析與設(shè)計(jì)[D].杭州：浙江大學(xué),2013.

[7] 蔡 亮,劉 騰.基于寫操作集的數(shù)據(jù)庫(kù)同步復(fù)制模型[J].計(jì)算機(jī)工程，2011，37(13)：61-62.

[8] 秦 森,楊 艷.基于Oracle日志分析的數(shù)據(jù)還原操作的設(shè)計(jì)及實(shí)現(xiàn)[J].電腦知識(shí)與技術(shù):學(xué)術(shù)交流,2007,1(3):626-628.

[9] 紀(jì) 彬，賀 立，白廣利，等.基于JMS的數(shù)據(jù)交換技術(shù)研究與實(shí)現(xiàn)[J].自動(dòng)化技術(shù)與應(yīng)用，2011,30(2):70-73.

[10] 趙 莉,杜思鋒.數(shù)據(jù)交換平臺(tái)中異構(gòu)數(shù)據(jù)轉(zhuǎn)換技術(shù)的研究[J].電子設(shè)計(jì)工程，2011,19(5):91-93.

[11] Hossain M S,Masud M,Muhammad G,et al.Automated and user involved data synchronization in collaborative e-health environments[J].Computers in Human Behavior,2014,30(1):485-490.

[12] 孫大為,張廣艷,鄭緯民.大數(shù)據(jù)流式計(jì)算:關(guān)鍵技術(shù)及系統(tǒng)實(shí)例[J].軟件學(xué)報(bào),2014,25(4):839-862.

[13] 金志國(guó),李 煒.基于Netty的HTTP客戶端的設(shè)計(jì)與實(shí)現(xiàn)[J].電信工程技術(shù)與標(biāo)準(zhǔn)化,2014,27(1):84-88.

[14] 竇進(jìn)業(yè).基于Protocol Buffers的企業(yè)即時(shí)通訊應(yīng)用的研究與實(shí)現(xiàn)[D].青島:中國(guó)海洋大學(xué),2014.

[15] 王 凱,趙庚申,劉國(guó)良.基于Android的電能管理系統(tǒng)的研究[J].南開大學(xué)學(xué)報(bào):自然科學(xué)版,2015,48(2):69-74.

[16] 陳增強(qiáng),郭嘉琳,劉忠信,等.具有斷點(diǎn)續(xù)傳功能的文件傳輸系統(tǒng)的設(shè)計(jì)與關(guān)鍵技術(shù)[J].計(jì)算機(jī)工程,2002,28(12):14-16.

[17] 王 賓.Hadoop集群的部署與管理系統(tǒng)的設(shè)計(jì)與實(shí)現(xiàn)[D].南京:南京大學(xué),2013.

[18] 張 璇.面向移動(dòng)計(jì)算的分布式文件共享服務(wù)平臺(tái)軟件研究[D].杭州:浙江大學(xué),2010.

[19] 李海騁,曹 春,呂 軍,等.支持依賴修復(fù)的熱部署技術(shù)[J].計(jì)算機(jī)科學(xué),2014，41(11):141-145.

[20] 金 海,官象山,吳 松,等.分布式存儲(chǔ)系統(tǒng)中文件傳輸優(yōu)化的設(shè)計(jì)與實(shí)現(xiàn)[J].華中科技大學(xué)學(xué)報(bào):自然科學(xué)版,2005,33(1):4-6.

Design and Implementation of Distributed Heterogeneous Data Synchronization Platform JTangSync

ZHANG Yuan-yuan1,ZHANG Qin-yan2,3,LI Feng4,YAO Xiang-long3

(1.College of Information Technology,Zhejiang Chinese Medical University,Hangzhou 310053,China；2.Computer Center,Zhejiang University,Hangzhou 310058,China;3.College of Computer Science,Zhejiang University,Hangzhou 315100,China；4.Institute of Scientific and Technical Information of Zhejiang Province,Hangzhou 310006,China)

The group enterprise data is usually stored in different heterogeneous environment,but the business requirement needs data exchange and synchronization for enterprise.Due to different application platform and data model,the traditional data synchronization method lacks of an unified framework for data synchronization.To this point,a distributed heterogeneous system JTangSync is designed and implemented for data synchronization.In this distributed architecture,each node contains data source module,the data transmission module and processor module.Each module is a replaceable plug-in which is convenient for secondary extension,and each node uses Zookeeper cluster which has capacity of centralized management,failure shifting and breakpoint continuing.It is proved in experiment that the method has good effect in heterogeneous data synchronization.

distributed heterogeneous data integration;data synchronization;JTangSync;data transmission;Zookeeper clusters

2015-10-16

2016-02-24

時(shí)間：2016-11-21

浙江省省級(jí)公益性技術(shù)應(yīng)用研究計(jì)劃(2014C33081);浙江省科技計(jì)劃項(xiàng)目(2013F20007)

張淵源(1980-)，女，講師，碩士，研究方向?yàn)樾畔⑾到y(tǒng)和醫(yī)療軟件中間件。

http://www.cnki.net/kcms/detail/61.1450.TP.20161121.1633.012.html

TP311.5

A

1673-629X(2016)12-0169-07

10.3969/j.issn.1673-629X.2016.12.037