DM 源碼閱讀系列文章（五）Binlog replication 實現

做者：lan

本文爲 DM 源碼閱讀系列文章的第五篇。上篇文章 介紹了 dump 和 load 兩個數據同步處理單元的設計實現，對核心 interface 實現、數據導入併發模型、數據導入暫停或中斷的恢復進行了分析。本篇文章將詳細地介紹 DM 核心處理單元 Binlog replication，內容包含 binlog 讀取、過濾、路由、轉換，以及執行等邏輯。文內涉及到 shard merge 相關邏輯功能，如 column mapping、shard DDL 同步處理，會在 shard merge 篇單獨詳細講解，這裏就不贅述了。

Binlog replication 處理流程

從上圖能夠大體瞭解到 Binlog replication 的邏輯處理流程，對應的 邏輯入口代碼。

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

2. 對 binlog events 進行處理轉換（transformation），這裏能夠作三類操做：

   操做	說明
   Filter	根據 庫/表同步黑白名單 對庫/表進行過濾；根據 binlog event 類型過濾。
   Routing	根據 庫/表 路由規則 對庫/表名進行轉換，用於合庫合表。
   Convert	將 binlog 轉換爲 job 對象，發送到 executor。

3. executor 對 job 進行衝突檢測，而後根據固定規則分發給對應的 worker 執行。
4. 按期保存 binlog position/gtid 到 checkpoint。

Binlog 讀取

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

• 從遠程的 MySQL/MariaDB
• 從 DM-worker 的本地 relay log

兩種方式都提供了一樣的讀取方法，處理核心都是 go-mysql。該庫主要提供了兩個功能：

• 註冊爲 MySQL/MariaDB 的 slave server ，從 MySQL/MariaDB 順序讀取 raw binlog events。
• 解析 raw binlog events。

更多的處理細節會在下篇關於 relay log 的文章中進行介紹，火燒眉毛的小夥伴能夠先翻閱一下相關代碼實現。

Binlog 轉換

處理程序拿到解析好的 binlog event 後，根據 binlog 的類型來對 binlog 進行分類處理。Binlog replication 主要關心如下類型的 binlog event ：

類型	說明
rotate event	消費完一個 binlog 文件，開始消費下一個 binlog 文件，用於更新 checkpoint 的 binlog position。
row event	包含 insert/update/delete DML 數據。
query event	包含 DDL 或者 statement DML 等數據。
xid event	表明一個 transaction 的 commit，通過 go-mysql 的處理後帶有對應 transaction 結束位置的 binlog position 和 gtid ，能夠用來保存 checkpoint。

Binlog replication 數據處理單元會對每一類 binlog event 進行如下的處理步驟，具體實現的處理順序可能略有差別，以代碼實現爲準。

過濾

Binlog replication 會從兩個維度對 binlog event 來進行過濾：

• 根據 同步庫/表黑白名單，過濾掉對應庫/表的全部 binlog event。
• 根據 binlog event 過濾規則，過濾掉對應庫/表指定的 binlog event。

row event 過濾處理 和 query event 過濾處理 的實如今邏輯上面存在一些差別：

• row event 包含 庫名和表名 信息；query event 須要經過 tidb parser 解析 event 裏面包含的 query statement 來獲取須要的庫名，表名以及其餘信息。
• tidb parser 不是徹底 100% 兼容 MySQL 語法，當遇到 parser 不支持的 query statement 時候，解析就會報錯，從而沒法獲取到對應的庫名和表名信息。Binlog replication 提供了一些 內置的不支持的 query statement 正則表達式，配合 使用 [schema-pattern: *, table-pattern: *] 的 binlog event 過濾規則，來跳過 parser 不支持的 query statement。
• query event 裏面也會包含 statement format binlog event，此時 Binlog replication 就能夠利用 parser 解析出來具體的 statement 類型，對不支持的 statement format binlog event 做出相應的處理： 對於須要同步的表，進行報錯處理；不須要同步的表，忽略繼續同步。

路由

binlog 過濾完成以後，對於須要同步的表就會根據過濾步驟得到的庫名和表名，經過 路由規則 轉換獲得須要同步到的目標庫名和表名，在接下來的轉換步驟來使用目標庫名和表名來轉換出正確的 DML 和 DDL statement。

轉換

row event 轉換處理和 query event 轉換處理的實現存在一些差別，這裏分開來說述。

row event 轉換處理經過三個轉換函數生成對應的 statements：

• generate insert sqls ：將 write rows event 轉換爲 replace into statements。

• generate update sqls：

  • safe mode = true，將 update rows event 轉換爲 delete + replace statements。
  • safe mode = false，將 update row event 轉換爲 update statements。
• generate delete sqls：將 delete rows event 轉換爲 delete statements。

query event 轉換處理：

• 由於 TiDB 目前不支持一條 DDL 語句包含多個 DDL 操做，query event 轉換處理會首先嚐試將 包含多個 DDL 變動操做的單條 DDL 語句 拆分紅 只包含一個 DDL 操做的多條 DDL 語句（具體代碼實現）。
• 使用 parser 將 DDL statement 對應的 ast 結構裏面的庫名和表名替換成對應的目標庫名和表名（具體代碼實現）。

經過轉換處理以後，將不一樣的 binlog event 包裝成不一樣的 job 發送到 executor 執行：

• row event -> insert/update/delete job
• query event -> ddl job
• xid event -> xid job

Job 執行

衝突檢測

binlog 順序同步模型要求按照 binlog 順序一個一個來同步 binlog event，這樣的順序同步勢必不能知足高 QPS 低同步延遲的同步需求，而且不是全部的 binlog 涉及到的操做都存在衝突。Binlog replication 採用衝突檢測機制，鑑別出來須要順序執行的 jobs，在確保這些 jobs 的順序執行的基礎上，最大程度地保持其餘 job 的併發執行來知足性能方面的要求。

衝突檢測流程以下：

• 遇到 DDL job，等待前面已經分發出來的全部 DML jobs 執行完成後，而後單獨執行該 DDL job，執行完成以後保存 checkpoint 信息。
• 遇到 DML job，會 先檢測而且嘗試解決衝突。若是檢測到衝突（即存在兩個 executor 的 worker 的 jobs 都須要與當前的 job 保持順序執行），會發送一個 flush job 來等待已經分發的全部 DML jobs 執行完成，而後再將 job 分發到對應的 worker，而且記錄該分發信息到內存。在沒有衝突的狀況下，若是不須要與已經分發出去的 job 保持順序的話，發送 job 到任意 worker 上；若是須要保持順序的話，那麼根據內存儲存的歷史分發信息，發送 job 到對應的 worker 上。

衝突檢測實現比較簡單，根據轉換步驟得到每條 statement 對應的 primary/unique key 信息，來進行交集檢測，若是存在交集那麼認定是須要順序的執行兩條 statement，請參考 具體實現代碼。

執行

job 分發到對應的 worker 後，worker 根據必定的規則來批量執行這些 job，以下：

• 遇到 DDL 當即執行。
• 遇到 flush 或者積累的 job 數量超過 配置的 batch 數量 當即執行。
• 沒有新的 job 分發進來，清空當前已經積累的 jobs 或者 sleep 10 ms。

根據上面三個規則能夠很快地將已經分發的 jobs 應用到下游 TiDB。

小結

本篇文章詳細地介紹 DM 核心處理單元 Binlog replication，內容包含 binlog 讀取、過濾、路由、轉換，以及執行等邏輯。下一篇咱們會對 relay log 數據處理單元的設計進行詳細的講解。