摘要：作者本文為源碼閱讀系列文章的第五篇。上篇文章介紹了和兩個(gè)數(shù)據(jù)同步處理單元的設(shè)計(jì)實(shí)現(xiàn)，對(duì)核心實(shí)現(xiàn)數(shù)據(jù)導(dǎo)入并發(fā)模型數(shù)據(jù)導(dǎo)入暫?；蛑袛嗟幕謴?fù)進(jìn)行了分析。根據(jù)庫(kù)表路由規(guī)則對(duì)庫(kù)表名進(jìn)行轉(zhuǎn)換，用于合庫(kù)合表。遇到或者積累的數(shù)量超過配置的數(shù)量立即執(zhí)行。

作者：lan

本文為 DM 源碼閱讀系列文章的第五篇。上篇文章 介紹了 dump 和 load 兩個(gè)數(shù)據(jù)同步處理單元的設(shè)計(jì)實(shí)現(xiàn)，對(duì)核心 interface 實(shí)現(xiàn)、數(shù)據(jù)導(dǎo)入并發(fā)模型、數(shù)據(jù)導(dǎo)入暫?；蛑袛嗟幕謴?fù)進(jìn)行了分析。本篇文章將詳細(xì)地介紹 DM 核心處理單元 Binlog replication，內(nèi)容包含 binlog 讀取、過濾、路由、轉(zhuǎn)換，以及執(zhí)行等邏輯。文內(nèi)涉及到 shard merge 相關(guān)邏輯功能，如 column mapping、shard DDL 同步處理，會(huì)在 shard merge 篇多帶帶詳細(xì)講解，這里就不贅述了。

從上圖可以大致了解到 Binlog replication 的邏輯處理流程，對(duì)應(yīng)的 邏輯入口代碼。

從 relay log 或者 MySQL/MariaDB 讀取 binlog events。

對(duì) binlog events 進(jìn)行處理轉(zhuǎn)換（transformation），這里可以做三類操作：

executor 對(duì) job 進(jìn)行沖突檢測(cè)，然后根據(jù)固定規(guī)則分發(fā)給對(duì)應(yīng)的 worker 執(zhí)行。

定期保存 binlog position/gtid 到 checkpoint。

Binlog replication 支持兩種方式讀取 binlog events:

從遠(yuǎn)程的 MySQL/MariaDB

從 DM-worker 的本地 relay log

兩種方式都提供了同樣的讀取方法，處理核心都是 go-mysql。該庫(kù)主要提供了兩個(gè)功能：

注冊(cè)為 MySQL/MariaDB 的 slave server ，從 MySQL/MariaDB 順序讀取 raw binlog events。

解析 raw binlog events。

更多的處理細(xì)節(jié)會(huì)在下篇關(guān)于 relay log 的文章中進(jìn)行介紹，迫不及待的小伙伴可以先翻閱一下相關(guān)代碼實(shí)現(xiàn)。

處理程序拿到解析好的 binlog event 后，根據(jù) binlog 的類型來對(duì) binlog 進(jìn)行分類處理。Binlog replication 主要關(guān)心以下類型的 binlog event ：

Binlog replication 數(shù)據(jù)處理單元會(huì)對(duì)每一類 binlog event 進(jìn)行以下的處理步驟，具體實(shí)現(xiàn)的處理順序可能略有差異，以代碼實(shí)現(xiàn)為準(zhǔn)。

Binlog replication 會(huì)從兩個(gè)維度對(duì) binlog event 來進(jìn)行過濾：

根據(jù) 同步庫(kù)/表黑白名單，過濾掉對(duì)應(yīng)庫(kù)/表的所有 binlog event。

根據(jù) binlog event 過濾規(guī)則，過濾掉對(duì)應(yīng)庫(kù)/表指定的 binlog event。

row event 過濾處理 和 query event 過濾處理 的實(shí)現(xiàn)在邏輯上面存在一些差異：

row event 包含 庫(kù)名和表名 信息；query event 需要通過 tidb parser 解析 event 里面包含的 query statement 來獲取需要的庫(kù)名，表名以及其他信息。

tidb parser 不是完全 100% 兼容 MySQL 語(yǔ)法，當(dāng)遇到 parser 不支持的 query statement 時(shí)候，解析就會(huì)報(bào)錯(cuò)，從而無法獲取到對(duì)應(yīng)的庫(kù)名和表名信息。Binlog replication 提供了一些 內(nèi)置的不支持的 query statement 正則表達(dá)式，配合 使用 [schema-pattern: *, table-pattern: *] 的 binlog event 過濾規(guī)則，來跳過 parser 不支持的 query statement。

query event 里面也會(huì)包含 statement format binlog event，此時(shí) Binlog replication 就可以利用 parser 解析出來具體的 statement 類型，對(duì)不支持的 statement format binlog event 作出相應(yīng)的處理： 對(duì)于需要同步的表，進(jìn)行報(bào)錯(cuò)處理；不需要同步的表，忽略繼續(xù)同步。

binlog 過濾完成之后，對(duì)于需要同步的表就會(huì)根據(jù)過濾步驟獲得的庫(kù)名和表名，通過 路由規(guī)則 轉(zhuǎn)換得到需要同步到的目標(biāo)庫(kù)名和表名，在接下來的轉(zhuǎn)換步驟來使用目標(biāo)庫(kù)名和表名來轉(zhuǎn)換出正確的 DML 和 DDL statement。

row event 轉(zhuǎn)換處理和 query event 轉(zhuǎn)換處理的實(shí)現(xiàn)存在一些差異，這里分開來講述。

row event 轉(zhuǎn)換處理通過三個(gè)轉(zhuǎn)換函數(shù)生成對(duì)應(yīng)的 statements：

generate insert sqls ：將 write rows event 轉(zhuǎn)換為 replace into statements。

generate update sqls：

safe mode = true，將 update rows event 轉(zhuǎn)換為 delete + replace statements。

safe mode = false，將 update row event 轉(zhuǎn)換為 update statements。

generate delete sqls：將 delete rows event 轉(zhuǎn)換為 delete statements。

query event 轉(zhuǎn)換處理：

因?yàn)?TiDB 目前不支持一條 DDL 語(yǔ)句包含多個(gè) DDL 操作，query event 轉(zhuǎn)換處理會(huì)首先嘗試將 包含多個(gè) DDL 變更操作的單條 DDL 語(yǔ)句 拆分成 只包含一個(gè) DDL 操作的多條 DDL 語(yǔ)句（具體代碼實(shí)現(xiàn)）。

使用 parser 將 DDL statement 對(duì)應(yīng)的 ast 結(jié)構(gòu)里面的庫(kù)名和表名替換成對(duì)應(yīng)的目標(biāo)庫(kù)名和表名（具體代碼實(shí)現(xiàn)）。

通過轉(zhuǎn)換處理之后，將不同的 binlog event 包裝成不同的 job 發(fā)送到 executor 執(zhí)行：

row event -> insert/update/delete job

query event -> ddl job

xid event -> xid job

binlog 順序同步模型要求按照 binlog 順序一個(gè)一個(gè)來同步 binlog event，這樣的順序同步勢(shì)必不能滿足高 QPS 低同步延遲的同步需求，并且不是所有的 binlog 涉及到的操作都存在沖突。Binlog replication 采用沖突檢測(cè)機(jī)制，鑒別出來需要順序執(zhí)行的 jobs，在確保這些 jobs 的順序執(zhí)行的基礎(chǔ)上，最大程度地保持其他 job 的并發(fā)執(zhí)行來滿足性能方面的要求。

沖突檢測(cè)流程如下：

遇到 DDL job，等待前面已經(jīng)分發(fā)出來的所有 DML jobs 執(zhí)行完成后，然后多帶帶執(zhí)行該 DDL job，執(zhí)行完成之后保存 checkpoint 信息。

遇到 DML job，會(huì) 先檢測(cè)并且嘗試解決沖突。如果檢測(cè)到?jīng)_突（即存在兩個(gè) executor 的 worker 的 jobs 都需要與當(dāng)前的 job 保持順序執(zhí)行），會(huì)發(fā)送一個(gè) flush job 來等待已經(jīng)分發(fā)的所有 DML jobs 執(zhí)行完成，然后再將 job 分發(fā)到對(duì)應(yīng)的 worker，并且記錄該分發(fā)信息到內(nèi)存。在沒有沖突的情況下，如果不需要與已經(jīng)分發(fā)出去的 job 保持順序的話，發(fā)送 job 到任意 worker 上；如果需要保持順序的話，那么根據(jù)內(nèi)存儲(chǔ)存的歷史分發(fā)信息，發(fā)送 job 到對(duì)應(yīng)的 worker 上。

沖突檢測(cè)實(shí)現(xiàn)比較簡(jiǎn)單，根據(jù)轉(zhuǎn)換步驟獲得每條 statement 對(duì)應(yīng)的 primary/unique key 信息，來進(jìn)行交集檢測(cè)，如果存在交集那么認(rèn)定是需要順序的執(zhí)行兩條 statement，請(qǐng)參考 具體實(shí)現(xiàn)代碼。

job 分發(fā)到對(duì)應(yīng)的 worker 后，worker 根據(jù)一定的規(guī)則來批量執(zhí)行這些 job，如下：

遇到 DDL 立即執(zhí)行。

遇到 flush 或者積累的 job 數(shù)量超過 配置的 batch 數(shù)量 立即執(zhí)行。

沒有新的 job 分發(fā)進(jìn)來，清空當(dāng)前已經(jīng)積累的 jobs 或者 sleep 10 ms。

根據(jù)上面三個(gè)規(guī)則可以很快地將已經(jīng)分發(fā)的 jobs 應(yīng)用到下游 TiDB。

本篇文章詳細(xì)地介紹 DM 核心處理單元 Binlog replication，內(nèi)容包含 binlog 讀取、過濾、路由、轉(zhuǎn)換，以及執(zhí)行等邏輯。下一篇我們會(huì)對(duì) relay log 數(shù)據(jù)處理單元的設(shè)計(jì)進(jìn)行詳細(xì)的講解。