DM 源碼閱讀系列文章（六）relay log 的實現

做者：張學程

本文爲 DM 源碼閱讀系列文章的第六篇，在 上篇文章 中咱們介紹了 binlog replication 處理單元的實現，對在增量複製過程當中 binlog event 的讀取、過濾、路由、轉換以及執行等邏輯進行了分析。

本篇文章咱們將會對 relay 數據處理單元的實現進行詳細的講解。這個單元的做用是從上游 MySQL/MariaDB 讀取 binlog event 並寫入到本地的 relay log file 中；當執行增量複製任務時，binlog replication 處理單元將讀取 relay log file 中的 event 並在進行解析後複製到下游的 TiDB 中。本篇文章的內容包括 relay log 目錄結構定義、relay log 數據的處理流程、主從切換支持、relay log 的讀取等邏輯。

值得注意的是，因爲咱們近期正在對 relay 處理單元進行重構，所以源碼中會同時包含重構先後的相關代碼實現。

relay log 目錄結構

一個已經進行過一次主從切換的 relay log 目錄結構大體以下：

<deploy_dir>/relay_log/
|-- 7e427cc0-091c-11e9-9e45-72b7c59d52d7.000001
|   |-- mysql-bin.000001
|   |-- mysql-bin.000002
|   |-- mysql-bin.000003
|   |-- mysql-bin.000004
|   `-- relay.meta
|-- 842965eb-091c-11e9-9e45-9a3bff03fa39.000002
|   |-- mysql-bin.000001
|   `-- relay.meta
`--  server-uuid.index

在 relay log 目錄下，主要包含如下幾類文件或文件夾數據：

類別	做用	文件（夾）名示例
relay log 子目錄	以單次主從切換髮生時對應於某個 MySQL/MariaDB server 爲單位組織 relay log 數據及 meta 信息	7e427cc0-091c-11e9-9e45-72b7c59d52d7.000001
relay log 數據文件	存儲實際的 binlog event 數據	mysql-bin.000001
relay meta 信息	存儲當前已從上游讀取並寫入爲 relay log 的 binlog event 對應於上游的 binlog position/GTID sets 信息	relay.meta
relay log 子目錄索引	索引各有效的 relay log 子目錄列表	server-uuid.index

relay log 處理流程

從上圖大體能夠了解 relay log 的邏輯處理流程，對應的入口代碼爲 Relay.Process，主要步驟包括：

1. 使用 binlog reader 從上游 MySQL/MariaDB 讀取 binlog event。

2. 將讀取到的 binlog event 使用 binlog transformer 進行轉換。

3. 將轉換後的 binlog event 使用 binlog writer 以 relay log file 的形式存儲在本地。

4. 當須要將數據以增量的方式同步到下游 TiDB 時，binlog replication 經過使用 relay reader 從 relay log file 中讀取 binlog event。

讀取 binlog event

relay 處理單元經過 Reader interface 從上游讀取 binlog event，其中最重要的方法爲讀取 binlog event 對象的 GetEvent。

當前對 Reader interface 的實現爲 reader，它最終經過 in 這個 br.Reader interface 從上游讀取 binlog event。reader 的使用流程爲：

1. 調用 Start 啓動讀取流程，並根據配置中是否啓用了 GTID 模式分別調用 setUpReaderByGTID 或 setUpReaderByPos 來啓動下層的 br.Reader 對象。

2. 調用 GetEvent 讀取 binlog event，具體爲 調用下層的 GetEvent 方法 獲取 binlog event。

3. 當再也不須要讀取 binlog event 時，調用 Close 關閉讀取操做。

從上面的流程能夠看出，具體的 binlog event 讀取操做使用的是另外一個下層的 br.Reader interface，當前選擇的具體實現 爲經過 TCP 鏈接進行讀取的 TCPReader。在 TCPReader 中，使用了 go-mysql 提供的 BinglogSyncer.StartSync 和 BinlogSyncer.StartSyncGTID 來啓動以 binlog position 模式或 GTID sets 模式讀取 binlog event，並經過 BinlogStreamer.GetEvent 讀取來自 TCP 的 binlog event。

轉換 binlog event

在 relay 處理單元中，對於從上游讀取到的 binlog event，咱們須要判斷是否須要寫進 relay log file 及是否須要更新對應的 relay.meta 內的斷點信息。所以在經過 Reader interface 讀取到 binlog event 後，經過調用 Transformer interface 來對 binlog event 進行相關的轉換處理。

當前對 Transformer interface 的實現爲 transformer，其主要經過在 Transform 方法中 對 binlog event 的類型進行判斷 後再進行相應處理，包括：

binlog event 類型	是否過濾	是否須要更新 relay.meta
RotateEvent	當是 fake RotateEvent 時過濾	否
QueryEvent	否	當是 DDL 時更新
XIDEvent	否	是
GenericEvent	當是 Heartbeat Event 時過濾	否
其餘類型	當 ARTIFICIAL flag 被設置時過濾	否

在 Transformer 中，咱們指望能達到如下目標：

1. 過濾上游 master server 上的 binlog file 中不存在的 binlog event，即指望 relay log file 中最終保存的 binlog event 與上游 master server 上的 binlog file 一致。

2. 僅在 DDL QueryEvent 時或 DML 事務完成時更新 relay.meta 以確保中斷恢復時能避免從 DML 事務進行中的 binlog event 處開始從上游請求 binlog event（對於 DML 相關的 binlog event，若是但願解析 INSERT/UPDATE/DELETE 等操做，則須要先獲取到對應的 TableMap event）。

寫入 relay log

在從上游讀取到 binlog event 並對其進行了相關轉換後，咱們就能夠嘗試將其寫入到本地的 relay log file 中。在 relay 處理單元中，用於將 binlog event 寫入 relay log file 的是 Writer interface，當前對應的實現爲 FileWriter，其內部會使用 out 這個 bw.FileWriter 來執行文件寫入操做，具體對 binlog event 執行寫入操做的是 WriteEvent 方法。

1. 各種型 binlog event 的判斷處理

在嘗試對 binlog event 進行寫入時，對於不一樣類型的 binlog event，須要 進行不一樣的判斷處理。

RotateEvent

在從上游讀取 binlog event 時，主要在如下狀況下可能會讀取到 RotateEvent：

1. 鏈接到上游 master server 開始讀取 binlog event 時，master 會發送一個 fake RotateEvent 告知 slave 後續 binlog event 對應的起始 binlog position。

2. 一個 master server 上的 binlog file 將要被讀取完成時，可能會包含一個 RotateEvent 以指示下一個 binlog file 的 filename 與起始 position。

所以，在處理 RotateEvent 寫入的 handleRotateEvent 方法中，主要包含如下操做：

1. 嘗試更新 FileWriter 內部記錄的當前 binlog 文件名爲 RotateEvent 內包含的文件名。

2. 判斷是不是 fake RotateEvent，若是是則跳事後續處理。

3. 與當前 relay log file 的 size 及內部 event 進行比較，判斷若是將當前 event 寫入到文件後是否會形成文件存在 hole 及該 event 是否在 relay log file 中已經存在，若是會形成 hole 則須要填充該 hole，若是已經存在則跳事後續的處理。

4. 將 event 寫入到 relay log file 中。

須要注意的是，咱們不能確保 master server 會將其 binlog file 中的全部 event 都發送給 slave（如當 MariaDB 未設置 BINLOG_SEND_ANNOTATE_ROWS_EVENT flag 時，master 就不會向 slave 發送 ANNOTATE_ROWS_EVENT），所以在寫入 event 到文件前，須要經過 handleFileHoleExist 判斷若是將 event 寫入到文件是否會存在 hole。若是存在 hode，則經過 event.GenDummyEvent 生成相應 size 的 dummy event 對 hole 進行填充。

另外須要注意的是，咱們不能確保 master server 不會將其已經發送給 slave 並寫入到了 relay log file 的 event 再次發送給 slave（如 master 在開始發送 slave 請求的 binlog event 前，會先發送 FormatDescriptionEvent 與 PreviousGTIDsEvent 等給 slave），所以在寫入 event 到文件前，須要經過 handleDuplicateEventsExist 判斷該 event 是否已經存在於 relay log file 中。

FormatDescriptionEvent

在從上游讀取 binlog event 時，主要在如下狀況下可能會讀取到 FormatDescriptionEvent：

1. 上游 master server 在發送除 RotateEvent 外的其餘 binlog event 以前，會發送一個 FormatDescriptionEvent 以使 slave 能正確 decode 後續的 binlog event。

2. 上游 master server 會將自身 binlog file 中存在的 FormatDescriptionEvent 發送給 slave，且這個 FormatDescriptionEvent 老是 binlog file 中的第 1 個 event。

所以，在處理 FormatDescriptionEvent 的 handleFormatDescriptionEvent 方法中，主要包含如下操做：

1. 關閉以前可能已經打開的 relay log file。

2. 打開該 event 須要寫入到的 relay log file 做爲當前活躍的 relay log file。

3. 檢查當前 relay log file 中是否存在 binlog file header（fe `bin`），若是不存在則爲其 寫入 binlog file header。

4. 檢查當前 relay log file 中是否存在 FormatDescriptionEvent，若是不存在則爲其 寫入該 FormatDescriptionEvent。

其餘類型 event

對於其餘類型的 binlog event，寫入操做由 handleEventDefault 進行處理，主要包含如下操做：

1. 與當前 relay log file 的 size 及內部 event 進行比較，判斷若是將當前 event 寫入到文件後是否會形成文件存在 hole 及該 event 是否在 relay log file 中已經存在，若是會形成 hole 則須要填充該 hole，若是已經存在則跳事後續的處理。

2. 將 event 寫入到 relay log file 中。

2. Recover relay log file

在寫入 binlog event 到 relay log file 時，儘管能夠經過 Flush 方法強制將緩衝中的數據刷新到磁盤文件中，但仍然可能出現 DM-worker 進程異常退出時部分數據未能刷新到磁盤文件中的狀況，形成 relay log file 內部分 event 數據缺失。

另外，對於一個事務對應的多個 binlog event，可能出現僅寫入了其中一部分 event 時 DM-worker 發生退出的狀況，形成 relay log file 中部分事務缺失部分 event。

所以，在 relay 處理單元中，咱們引入了對 relay log file 執行 Recover 的機制，用於將 relay log file 尾部不完整的 event 及事務進行踢除，對應的方法爲 FileWrite.Recover，具體實如今 doRecovering 方法中，主要操做包括：

1. 獲取 relay log file 中直到最後一個完整事務對應的 binlog position 與 GTID sets。

2. 比較 relay log file 的 size 與獲取到的 binlog position，若是相等則說明這個 relay log file 中包含的事務都是完整的，跳事後續的處理。

3. 若是 relay log file 的 size 比 binlog position 更小，則向外部報告錯誤並跳事後續的處理。

4. 若是 relay log file 的 size 比 binlog position 大，則 將 relay log file 中超出 binlog position 的部分執行 Truncate 進行截斷。

主從切換支持

爲支持將 relay 處理單元鏈接的上游 master server 在 replica group 內的不一樣 server 間進行切換（也包括 relay 處理單元鏈接的上游 VIP 指向的實際 server 發生了改變），relay 處理單元會嘗試將從不一樣上游 server 讀取到的 binlog event 保存到不一樣的 relay log 子目錄中，目錄與文件結構能夠參考前文的 relay log 目錄結構。

爲支持上述功能，relay 處理單元在讀取 binlog event 前主要執行如下操做：

1. 比較當前上游 server 的 UUID 信息與 relay.meta 信息，判斷當前鏈接到的是不是前一次鏈接過的 server。

2. 若是不是前一次鏈接過的 server，則說明切換到了新的 server，所以建立新的 relay log 子目錄並更新對應的 meta 信息。

讀取 relay log

relay 處理單元用於從上游讀取 binlog event 並將其寫入到本地的 relay log file 中。當執行增量數據複製時，binlog replication 處理單元須要經過 streamer pkg 讀取 relay log file 並從中解析獲取須要同步的數據，其中執行讀取的對象爲 BinlogReader。

由前文介紹過的主從切換支持可知咱們會將具體的 relay log 數據存儲在可能的多個子目錄中，所以在讀取 relay log 時，咱們也 須要考慮按序依次讀取，主要操做包括：

1. 調用 parseRelay 開始從 relay log 的根目錄執行解析讀取。

2. 調用 parseDirAsPossible 開始從外部指定的或上一次調用返回的子目錄、文件及 offset 處開始讀取，並返回下一次調用時須要的子目錄、文件及 offset（便可實現切換到新的 relay log 子目錄）。

3. 對於當前須要讀取的子目錄，調用 CollectBinlogFilesCmp 收集該目錄內指定 relay log 文件及其以後的全部 relay log 文件。

4. 對於每個收集到的 relay log 文件，調用 parseFileAsPossible 嘗試對其進行解析讀取。

5. 在 parseFileAsPossible 中，反覆返回 調用 parseFile 進行 binlog event 的讀取，直到 發生錯誤 或 檢測到須要切換到新的 relay log 文件或子目錄。

6. 對因而否須要切換到新的 relay log 文件或子目錄的檢測經過在 parseFile 內 調用 needSwitchSubDir 與 調用 relaySubDirUpdated 實現。

小結

本篇文章詳細地介紹了 relay 處理單元的實現，內容包括了 relay log 的目錄結構、如何從上游 server 讀取 binlog event 並寫入到本地的 relay log file 中，以及 binlog replication 處理單元將如何讀取本地的 relay log file。到本篇文章爲止，咱們完成了對 DM 中的數據處理單元的介紹。從下一篇文章開始，咱們將開始詳細介紹 DM 內部主要功能的設計與實現原理。