劉麗，胡曉勤

（四川大學(xué)網(wǎng)絡(luò)空間安全學(xué)院，成都610065）

MySQL是最流行的關(guān)系型數(shù)據(jù)庫管理系統(tǒng)之一，在國內(nèi)互聯(lián)網(wǎng)中應(yīng)用十分廣泛，原生的MySQL主從復(fù)制缺乏的靈活性，無法滿足源端、目標(biāo)端庫表結(jié)構(gòu)不一致的同步場景，也不能同步到異構(gòu)數(shù)據(jù)存儲(chǔ)中。基于MySQL Replication協(xié)議，設(shè)計(jì)并實(shí)現(xiàn)一種基于MySQL的數(shù)據(jù)同步中間件，支持增量數(shù)據(jù)的自定義規(guī)則處理，提供數(shù)據(jù)同步定制能力，目標(biāo)數(shù)據(jù)存儲(chǔ)支持MySQL、Kafka，以及其他異構(gòu)數(shù)據(jù)存儲(chǔ)。

MySQL；數(shù)據(jù)同步；異構(gòu)；中間件

0 引言

在互聯(lián)網(wǎng)高速發(fā)展的今天，針對MySQL的數(shù)據(jù)同步需求越來越繁雜，原生的MySQL主從復(fù)制僅支持MySQL實(shí)例之間的全量復(fù)制，且主從數(shù)據(jù)庫之間庫、表結(jié)構(gòu)一致，這在一些數(shù)據(jù)遷移場景需求下無法滿足。如將源MySQL實(shí)例的A庫遷移到目標(biāo)MySQL實(shí)例的B庫、在同步過程中根據(jù)某些數(shù)據(jù)字段的變化情況進(jìn)行復(fù)制等，針對這些業(yè)務(wù)定制很強(qiáng)的場景，MySQL原生主從并不能解決。此外，在OLAP[1]場景下，往往需要將關(guān)注的MySQL增量數(shù)據(jù)采集到Kafka、ElasticSearch等異構(gòu)數(shù)據(jù)存儲(chǔ)中，因此，也需要解決MySQL同步至異構(gòu)數(shù)據(jù)存儲(chǔ)的問題。

針對上述問題，本文提出了一種基于MySQL的數(shù)據(jù)同步中間件，支持MySQL實(shí)例之間的單向數(shù)據(jù)同步，并提供了自定義增量數(shù)據(jù)同步規(guī)則的定制入口，能滿足所有的映射同步、過濾同步等復(fù)雜的同步場景。此外，還支持將MySQL增量數(shù)據(jù)同步至Kafka等異構(gòu)數(shù)據(jù)存儲(chǔ)中，應(yīng)用也可自行實(shí)現(xiàn)自己的數(shù)據(jù)存儲(chǔ)同步適配器，具備極高的可拓展性。

1 數(shù)據(jù)同步中間件架構(gòu)

本文提出的數(shù)據(jù)同步中間件系統(tǒng)架構(gòu)如圖1所示，整體分為邏輯層、協(xié)調(diào)層。本數(shù)據(jù)同步中間件的邏輯層包含Replicator組件和Consumer組件，Replicator組件負(fù)責(zé)從存儲(chǔ)層的源MySQL實(shí)例中采集增量的Binlog Event數(shù)據(jù)，經(jīng)過解析、過濾后，使用自定義的數(shù)據(jù)結(jié)構(gòu)封裝并保留在存儲(chǔ)模塊中，Consumer組件負(fù)責(zé)從Replicator組件中消費(fèi)增量數(shù)據(jù)，并還原至目標(biāo)MySQL實(shí)例中；協(xié)調(diào)層負(fù)責(zé)為邏輯層組件提供分布式協(xié)調(diào)服務(wù)，基于Zookeeper[2]為Replicator組件和Con?sumer組件提供狀態(tài)同步、集群管理、服務(wù)發(fā)現(xiàn)的分布式服務(wù)能力。

圖1 數(shù)據(jù)同步中間件系統(tǒng)架構(gòu)

2 Replicator組件

2.1 訂閱模塊

訂閱模塊負(fù)責(zé)與訂閱的源MySQL實(shí)例進(jìn)行網(wǎng)絡(luò)通信交互，完成增量數(shù)據(jù)的持續(xù)采集，如DML（Data Manipulation Language）和DDL（Data Definition Lan?guage）操作產(chǎn)生的Binlog Event數(shù)據(jù)。訂閱模塊與MySQL進(jìn)行網(wǎng)絡(luò)交互的協(xié)議包括TCP三次握手協(xié)議、MySQL HandShake協(xié)議、MySQL Replication協(xié)議[3]，交互時(shí)序如圖2所示，該交互過程中涉及到的MySQL協(xié)議均遵循原生協(xié)議規(guī)范。

圖2 訂閱模塊與MySQL協(xié)議交互流程

（1）訂閱模塊與MySQL進(jìn)行TCP三次握手，建立TCP連接；

（2）完成TCP連接建立后，MySQL會(huì)主動(dòng)給訂閱模塊發(fā)送一個(gè)MySQL初始化握手?jǐn)?shù)據(jù)包，訂閱模塊對該握手?jǐn)?shù)據(jù)包解析后，將自身登錄使用的賬號、密碼等信息反饋給MySQL服務(wù)端；

（3）當(dāng)訂閱模塊與MySQL服務(wù)端成功握手后，訂閱模塊通過MySQL Replication協(xié)議，給MySQL服務(wù)端發(fā)送COM_REGISTER_SLAVE命令，將本模塊注冊為MySQL服務(wù)端的從節(jié)點(diǎn)；

（4）當(dāng)訂閱模塊成功注冊成為MySQL服務(wù)端的從節(jié)點(diǎn)后，訂閱模塊通過MySQL Replication協(xié)議，給MySQL服務(wù)端發(fā)送COM_BINLOG_DUMP命令，使MySQL服務(wù)端持續(xù)推送增量的Binlog Event數(shù)據(jù)給訂閱模塊；

完成上述的MySQL Replication協(xié)議交互后，本模塊將使用Netty框架監(jiān)聽訂閱MySQL推送Binlog Event數(shù)據(jù)流的寫入事件，并將其緩存在字節(jié)緩沖區(qū)中，用于解決粘包/拆包問題，截取完整Binlog Event數(shù)據(jù)包流程如圖3所示。

圖3 Binlog Event粘包/拆包解析流程

（1）訂閱模塊接收到MySQL服務(wù)端推送的數(shù)據(jù)包后，嘗試解析前四個(gè)字節(jié)，并獲取完整數(shù)據(jù)包的字節(jié)大小packet_length；

（2）如果本數(shù)據(jù)包剩余字節(jié)大小大于pack?et_length，說明出現(xiàn)了粘包現(xiàn)象，緩沖區(qū)中包含了完整的Binlog Event數(shù)據(jù)，則根據(jù)packet_length進(jìn)行截取，獲得完整的Binlog Event數(shù)據(jù)包；如果本數(shù)據(jù)包剩余字節(jié)大小小于packet_length，說明出現(xiàn)了拆包現(xiàn)象，需要等待下一個(gè)數(shù)據(jù)包到達(dá)；

通過上述方法進(jìn)行完整Binlog Event數(shù)據(jù)包的截取后，將其交給解析模塊進(jìn)行深度解析。

2.2 解析模塊

解析模塊負(fù)責(zé)解析Binlog Event數(shù)據(jù)包，將其封裝成自描述的數(shù)據(jù)結(jié)構(gòu)。解析模塊從訂閱模塊中獲取的Binlog Event數(shù)據(jù)包遵循原生MySQL中Binlog Event數(shù)據(jù)包協(xié)議，通過對其解析后，獲取完整的行數(shù)據(jù)變更描述。DDL（Data Definition Language，數(shù)據(jù)定義語言）語句產(chǎn)生的Binlog Event數(shù)據(jù)，解析后獲得其執(zhí)行時(shí)的SQL語句；DML（Data Manipulation Language，數(shù)據(jù)操縱語言）語句產(chǎn)生的Binlog Event數(shù)據(jù)，解析后獲得行數(shù)據(jù)變更前后的數(shù)據(jù)描述，包括列名、列值、列類型等信息；

本文描述的數(shù)據(jù)同步中間件基于MySQL Server的行復(fù)制（Row-Based Replication，RBR）模式，在這種模式下Binlog Event會(huì)詳細(xì)描述行數(shù)據(jù)變更前后的狀態(tài)，可以最大限度的保證主從復(fù)制的一致性，需要MySQL服務(wù)端開啟binlog_format=ROW?；贛ySQL服務(wù)端的行復(fù)制模式，當(dāng)客戶端在MySQL服務(wù)端提交事務(wù)時(shí)，二進(jìn)制日志中會(huì)產(chǎn)生QUERY_EVENT、TA?BLE_MAP_EVENT、ROWS_EVENT（WRITE/UPDATE/DELETE）、XID_EVENT四種類型的事件，如圖4所示。

圖4 Rows-Replication事件結(jié)構(gòu)圖

（1）QUERY_EVENT。當(dāng)客戶端執(zhí)行DML導(dǎo)致行數(shù)據(jù)變更時(shí)，該事件記錄了事務(wù)開始的發(fā)生，該類型的Binlog Event數(shù)據(jù)包中可以記錄諸如BEGIN、END、XA START、XA END、XA COMMIT、ROLLBACK等語義；當(dāng)客戶端執(zhí)行DDL語句時(shí)，該事件記錄了DDL語句操作的庫名稱以及客戶端執(zhí)行的SQL語句；

（2）TABLE_MAP_EVENT。當(dāng)客戶端執(zhí)行DML導(dǎo)致行數(shù)據(jù)變更時(shí)會(huì)產(chǎn)生該類型的事件，該類型的Bin?log Event數(shù)據(jù)包中記錄了行數(shù)據(jù)變更的庫、表、字段信息。解析模塊在完成本類型Binlog Event數(shù)據(jù)包解析后，將在本地內(nèi)存中緩存該表信息的內(nèi)容，供下文解析ROWS_EVENT使用；

（3）ROWS_EVENT（WRITE/UPDATE/DELETE）。該事件類型的Binlog Event數(shù)據(jù)詳細(xì)描述了DML語句修改數(shù)據(jù)的前后狀態(tài)，ROWS_EVENT類型的Binlog Event數(shù)據(jù)包中記錄了變更行數(shù)據(jù)的表序號（table_id），結(jié)合TABLE_MAP_EVENT中解析的表信息，可以從ROWS_EVENT數(shù)據(jù)包中獲取行數(shù)據(jù)變更前后的列值信息；

（4）XID_EVENT。當(dāng)客戶端在MySQL服務(wù)端提交事務(wù)時(shí)會(huì)產(chǎn)生該類型事件，標(biāo)識(shí)了一個(gè)事務(wù)的結(jié)尾；

通過對上述類型的Binlog Event數(shù)據(jù)進(jìn)行解析，解析模塊將需要的解析結(jié)果用一個(gè)定義好的對象結(jié)構(gòu)RowData存儲(chǔ)，完整的對象結(jié)構(gòu)描述如表1，RowData數(shù)據(jù)結(jié)構(gòu)中存儲(chǔ)了本事件的語義類型（如DML、DDL、事務(wù)開始/結(jié)束等類型）、Binlog位點(diǎn)（Binlog文件名、偏移量）、DML語句執(zhí)行后行數(shù)據(jù)變化前后的列信息，該列信息數(shù)據(jù)結(jié)構(gòu)如表2所示，其中，字段名稱在Binlog Event中并未提供，需要解析模塊自行查詢源MySQL獲取。

表1 Row Data對象結(jié)構(gòu)

表2 Column對象結(jié)構(gòu)

2.3 過濾模塊

過濾模塊負(fù)責(zé)過濾不關(guān)注的庫、表數(shù)據(jù)。本模塊使用Aviator引擎框架完成正則表達(dá)式的高性能求值，實(shí)現(xiàn)了Binlog Event黑白名單功能，過濾類型包括以下：

（1）Binlog Event類型過濾。本文中只保留了其中的事務(wù)開始事件、事務(wù)結(jié)束事件、DML產(chǎn)生的行數(shù)據(jù)變更事件和DDL產(chǎn)生的變更事件；

（2）Binlog Event庫表過濾。除了對事件類型進(jìn)行過濾外，過濾模塊還將根據(jù)用戶指定的庫、表黑白名單對Binlog Event數(shù)據(jù)進(jìn)行過濾，只保留本數(shù)據(jù)同步中間件關(guān)注的庫表Binlog Event數(shù)據(jù)，減輕下游數(shù)據(jù)處理壓力。

2.4 存儲(chǔ)模塊

存儲(chǔ)模塊負(fù)責(zé)需要同步的Binlog Event數(shù)據(jù)的管理與分發(fā)，本模塊使用分段數(shù)據(jù)文件的形式存儲(chǔ)被解析的Binlog Event數(shù)據(jù)，在經(jīng)過前置過濾模塊的篩選后，數(shù)據(jù)文件中只保存了下游需要消費(fèi)的事件數(shù)據(jù)。

數(shù)據(jù)文件使用二進(jìn)制形式存儲(chǔ)數(shù)據(jù)，通過Proto?buf[4]序列化算法將Binlog Event數(shù)據(jù)的位點(diǎn)信息和事件內(nèi)容信息轉(zhuǎn)換成一連串的字節(jié)描述，并使用緊湊的變長變量存儲(chǔ)以提高磁盤利用率。具體的，每一條事件數(shù)據(jù)的存儲(chǔ)結(jié)構(gòu)如表3。其中，位點(diǎn)信息和事件內(nèi)容信息為變長數(shù)據(jù)，分別使用8字節(jié)固定長度的空間存儲(chǔ)變長數(shù)據(jù)的字節(jié)數(shù)，并通過CRC32（循環(huán)冗余校驗(yàn)）算法計(jì)算出前(8+positon_length+8+rowdata_length)字節(jié)內(nèi)容的的校驗(yàn)碼，用于數(shù)據(jù)訪問時(shí)做簡單的數(shù)據(jù)檢查，避免數(shù)據(jù)損失導(dǎo)致數(shù)據(jù)讀取異常。位點(diǎn)信息的數(shù)據(jù)結(jié)構(gòu)如表4，包含了Binlog文件名、Binlog偏移量以及該Binlog Event的時(shí)間戳。

表3 數(shù)據(jù)文件Binlog Event數(shù)據(jù)存儲(chǔ)結(jié)構(gòu)

表4 位點(diǎn)信息數(shù)據(jù)結(jié)構(gòu)

3 Consumer組件

3.1 消費(fèi)模塊

消費(fèi)模塊負(fù)責(zé)通過自定義RPC協(xié)議，從Replicator組件中批量獲取關(guān)注的Binlog Event數(shù)據(jù)。該消費(fèi)通信遵循2PC（Two-Phase Commit protocol，兩階段提交協(xié)議）協(xié)議，第一階段中，消費(fèi)模塊從Replicator組件中獲取批量數(shù)據(jù)，第二階段，消費(fèi)模塊確認(rèn)第一階段數(shù)據(jù)處理完成或進(jìn)行消費(fèi)回滾，時(shí)序流程如圖5所示。

圖5 兩階段流程

（1）兩階段Ack流程。消費(fèi)模塊首先進(jìn)行第一階段，從存儲(chǔ)模塊中批量獲取RowData數(shù)據(jù)，存儲(chǔ)模塊將為該批次數(shù)據(jù)分配一個(gè)序號batchId，消費(fèi)模塊完成該批次RowData的處理后，第二階段對該批次batchId進(jìn)行確認(rèn)，表示該部分?jǐn)?shù)據(jù)已經(jīng)消費(fèi)完成，至此兩端Ack流程完成；

（2）兩階段Rollback流程。區(qū)別于Ack流程，當(dāng)消費(fèi)模塊沒有正確處理完該批RwoData數(shù)據(jù)，認(rèn)為需要重試時(shí)，第二階段將向存儲(chǔ)模塊發(fā)起回滾請求，消費(fèi)模塊下次獲取數(shù)據(jù)時(shí)仍從本批次開始。

3.2 腳本引擎模塊

腳本引擎模塊負(fù)責(zé)對用戶自定義規(guī)則腳本生命周期的維護(hù)，包括加載、編譯、編排、執(zhí)行。基于規(guī)則腳本，用戶可以對RowData數(shù)據(jù)自定義加工處理，實(shí)現(xiàn)字段級數(shù)據(jù)過濾、字段加工、映射同步等功能，為業(yè)務(wù)提供靈活的增量數(shù)據(jù)加工處理入口。

在腳本引擎模塊中將解析兩種類型的文件，一種是腳本編排文件，一種是腳本規(guī)則文件，其中腳本規(guī)則文件中存儲(chǔ)的是用戶編寫的數(shù)據(jù)處理邏輯，本文中包括Java文件和Groovy文件，腳本編排編排文件中存儲(chǔ)了規(guī)則文件的編排方式信息，以Yaml格式進(jìn)行存儲(chǔ)，每一個(gè)腳本信息的描述數(shù)據(jù)結(jié)構(gòu)如表5所示，本模塊通過指定每個(gè)腳本的執(zhí)行順序進(jìn)行執(zhí)行拓?fù)涞木幣拧?/p>

基于上述的腳本編排文件格式，如圖7所示，在文件系統(tǒng)中存在編排文件engine.yaml以及多個(gè)規(guī)則腳本文件，腳本引擎模塊將根據(jù)編排文件中描述的腳本信息進(jìn)行腳本的編譯、編排，如圖6所示。

表5 腳本編排信息數(shù)據(jù)結(jié)構(gòu)

圖6 腳本引擎工作流程圖

（1）腳本引擎初始化時(shí)讀取指定目錄下的編排文件engine.yaml文件并進(jìn)行解析，獲得編排文件中存儲(chǔ)的腳本編排信息，包括腳本文件的腳本名稱、腳本路徑、腳本類型以及執(zhí)行序號；

（2）根據(jù)編排文件中的腳本信息，對涉及的規(guī)則腳本進(jìn)行編譯，如果是Java類型的規(guī)則文件，則使用JDK6開始提供的JavaCompiler工具進(jìn)行動(dòng)態(tài)編譯；如果是Groovy類型的規(guī)則文件，則使用GroovyClassLoader加載器進(jìn)行編譯，并緩存在腳本引擎模塊中；

（3）根據(jù)編排文件中描述的執(zhí)行順序，對規(guī)則腳本的執(zhí)行進(jìn)行編排，當(dāng)有數(shù)據(jù)輸入時(shí)，根據(jù)該規(guī)則拓?fù)溥M(jìn)行數(shù)據(jù)加工處理。

3.3 同步適配器模塊

3.3.1 MySQL同步適配器

本中間件描述的方案基于InnoDB存儲(chǔ)引擎，且同步表含有主鍵。對于DDL語句，直接在目標(biāo)MySQL執(zhí)行RowData中記錄的SQL即可。對于DML，本中間件通過RowData合并、SQL還原以及并行復(fù)制的方式完成目標(biāo)MySQL的寫入。工作流程如圖7所示。

（1）Merger對本批次RowData進(jìn)行合并，在保證數(shù)據(jù)一致性的前提下，將本批次中相同主鍵的行數(shù)據(jù)變更記錄進(jìn)行合并，如一條行記錄先后經(jīng)歷INSERT、多次UPDATE，最終被DELETE，該行數(shù)據(jù)在結(jié)果上不需要同步到目標(biāo)數(shù)據(jù)庫，即以最終結(jié)果為準(zhǔn)，減少了目標(biāo)MySQL回放的數(shù)據(jù)量；

（2）Partitioner對合并后的數(shù)據(jù)進(jìn)行分區(qū)。根據(jù)行數(shù)據(jù)的庫、表、主鍵三元組進(jìn)行哈希分組，將其切分成N個(gè)小批次數(shù)據(jù)，并提交到線程池；

（3）線程池調(diào)度執(zhí)行，每個(gè)線程處理一個(gè)小批次的數(shù)據(jù)，通過SQL還原的方法將RowData描述的數(shù)據(jù)變化轉(zhuǎn)換成DML語句，針對每個(gè)事件類型進(jìn)行如下處理：

①本事件屬于INSERT類型。使用INSERT IN?TO...ON DUPLICATE KEY UPDATE語義，該語義在目標(biāo)MySQL存在該行數(shù)據(jù)時(shí)進(jìn)行更新操作，不存在該行數(shù)據(jù)時(shí)進(jìn)行插入操作。根據(jù)RowData的schema、table、afterChange構(gòu)造SQL語句如下：

INSERT INTO`schema`.`table`(`filed_1`，...，`field_n`)VALUES(value_1，…，value_n)ONDUPLICATEKEY UP?DATE field_1=VALUES(field_1)，…，field_n=VALUES(field_n);

②本事件屬于UPDATE類型。使用REPLACE INTO語 義，根 據(jù)RowData的schema、table、before?Change、afterChange構(gòu)造SQL語句如下：

REPLACE INTO `schema`.`table`(`filed_1`，…，`field_n`)VALUES(value_1，…，value_n);

③本事件屬于DELETE類型。根據(jù)RowData的schema、table、beforeChange進(jìn)行SQL還原，根據(jù)主鍵刪除目標(biāo)MySQL數(shù)據(jù)，構(gòu)造DELETE語句如下：

DELETE FROM`schema`.`table`WHERE`pri_key_1`=pri_value_1…AND`pri_key_2`=pri_value_2;

（4）多線程并行寫入目標(biāo)MySQL，完成數(shù)據(jù)同步。

圖7 MySQL同步適配器并行復(fù)制過程

3.3.2異構(gòu)同步適配器

在本數(shù)據(jù)同步中間件中，每一個(gè)異構(gòu)數(shù)據(jù)存儲(chǔ)類型對應(yīng)了一種同步適配器，適配器負(fù)責(zé)獲取MySQL增量行數(shù)據(jù)，在進(jìn)行數(shù)據(jù)處理后，通過異構(gòu)數(shù)據(jù)存儲(chǔ)的SDK完成寫入。結(jié)合本中間件提供的腳本引擎模塊，可以完成更多目標(biāo)數(shù)據(jù)存儲(chǔ)的支持，有很高的可拓展性。當(dāng)目標(biāo)為Kafka等隊(duì)列存儲(chǔ)時(shí)，同步適配器需要著重考慮Binlog Event分發(fā)的順序性，此時(shí)提供兩種寫入策略。

（1）對于需要保證行數(shù)據(jù)級別Binlog Event順序的場景，在適配器處理數(shù)據(jù)時(shí)，根據(jù)庫、表、主鍵計(jì)算哈希值，并對寫入隊(duì)列的數(shù)量取余，將相同主鍵的行數(shù)據(jù)事件分發(fā)到同一個(gè)隊(duì)列中，只要下游應(yīng)用對每個(gè)隊(duì)列啟用一個(gè)線程進(jìn)行消費(fèi)，則可以保證行數(shù)據(jù)級別的Bin?log Event的順序性；

（2）對于需要保證全局Binlog Event順序的場景，該適配器寫入隊(duì)列的數(shù)量只能為1，并且使用單線程寫入隊(duì)列，下游應(yīng)用也只能啟用一個(gè)線程消費(fèi)該隊(duì)列，吞吐量將十分有限。

4 基于Zookeeper的分布式增強(qiáng)

4.1 組件高可用

Replicator組件和Consumer組件的集群化，實(shí)現(xiàn)組件的主備切換，提供高可用服務(wù)能力。以Replicator組件為例，當(dāng)多個(gè)組件啟動(dòng)初始化時(shí)，分別向Zookeeper相同路徑下中注冊臨時(shí)節(jié)點(diǎn)，路徑為/data/sync/{task_id}/replicator/node，其中task_id為本同步任務(wù)的唯一標(biāo)識(shí)，注冊時(shí)使用的值為各自組件的IP地址和服務(wù)端口。由于Zookeeper臨時(shí)節(jié)點(diǎn)的特性，只有一個(gè)節(jié)點(diǎn)會(huì)注冊成功，注冊成功的節(jié)點(diǎn)將作為主節(jié)點(diǎn)，進(jìn)行正常的工作，其余節(jié)點(diǎn)將注冊該臨時(shí)節(jié)點(diǎn)的監(jiān)聽事件，當(dāng)該臨時(shí)節(jié)點(diǎn)發(fā)生刪除事件時(shí)，將會(huì)重新進(jìn)行新一輪的選主過程，以此保證組件的高可用特性。對于Consum?er組件同樣采用上述邏輯實(shí)現(xiàn)高可用。

4.2 服務(wù)發(fā)現(xiàn)

當(dāng)Replicator集群發(fā)生主備切換時(shí)，Consumer組件需要及時(shí)切換到新的Replicator組件節(jié)點(diǎn)進(jìn)行訂閱消費(fèi)。當(dāng)Consumer組件啟動(dòng)后，會(huì)在Zookeeper中查詢/data/sync/{task_id}/replicator/node是否有注冊成功的Replicator組件，如果有成功注冊的節(jié)點(diǎn)，則獲取節(jié)點(diǎn)中存儲(chǔ)的IP地址和服務(wù)端口，隨后通過RPC協(xié)議與Replicator組件進(jìn)行2PC交互，如果沒有發(fā)現(xiàn)成功注冊的Replicator節(jié)點(diǎn)，則Consumer組件處于掛起監(jiān)聽狀態(tài)，直到有Replicator組件在該同步任務(wù)下創(chuàng)建臨時(shí)節(jié)點(diǎn)，以此實(shí)現(xiàn)Replicator組件的服務(wù)發(fā)現(xiàn)。

4.3 消費(fèi)位點(diǎn)管理

對于Replicator組件進(jìn)行消費(fèi)位點(diǎn)的管理流程如圖8所示，當(dāng)Replicator接收到二階段Ack請求后，根據(jù)批次號獲取該批數(shù)據(jù)的位點(diǎn)信息，在更新內(nèi)存消費(fèi)位點(diǎn)后，完成本次請求后直接反饋成功，之后，Replica?tor組件由內(nèi)部異步任務(wù)定時(shí)向Zookeeper中更新位點(diǎn)信息，位點(diǎn)信息在Zookeeper中的路徑為/data/sync/rep?licator/{consumer_cluster_id}/process，其 中 consum?er_cluster_id為Consumer集群的唯一標(biāo)識(shí)，存入的值為本批次最后一條Binlog Event的位點(diǎn)。

圖8 Replicator位點(diǎn)記錄時(shí)序圖

5 測試

測試所用的機(jī)房有兩套，分別為北京機(jī)房和廣州機(jī)房，兩個(gè)機(jī)房之間的網(wǎng)絡(luò)延遲約為30ms左右，環(huán)境配置如表6。

表6 測試環(huán)境配置表

本次實(shí)驗(yàn)的拓?fù)浣Y(jié)構(gòu)如圖9所示，兩個(gè)MySQL實(shí)例間都已經(jīng)打開二進(jìn)制日志并設(shè)置為ROW模式。

圖9 單向同步實(shí)驗(yàn)拓?fù)鋱D

下面將從INSERT、UPDATE以及INSERT/UP?DATE混合三種場景下，對比單向同步復(fù)制中不同并行度下的性能表現(xiàn)，具體數(shù)據(jù)如表7所示。

表7 單向同步并行復(fù)制性能測試報(bào)告

由上述實(shí)驗(yàn)情況可知，本數(shù)據(jù)同步中間件在MySQL單向同步的INSERT場景下表現(xiàn)良好，主要是經(jīng)過數(shù)據(jù)合并后對同表的INSERT語句使用了預(yù)編譯方式，將INSERT語句批量提交到目標(biāo)MySQL中，避免了目標(biāo)MySQL多次解析、優(yōu)化相同的SQL語句。

6 結(jié)語

本文基于MySQL設(shè)計(jì)并實(shí)現(xiàn)了一個(gè)數(shù)據(jù)同步中間件，用于將MySQL中的增量數(shù)據(jù)同步至目標(biāo)MySQL或其他異構(gòu)數(shù)據(jù)存儲(chǔ)中，并支持業(yè)務(wù)自定義增量數(shù)據(jù)處理的規(guī)則腳本，幾乎滿足了各類業(yè)務(wù)的同步需求。