摘要：作者張學程本文為源碼閱讀系列文章的第六篇，在上篇文章中我們介紹了處理單元的實現，對在增量復制過程中的讀取過濾路由轉換以及執行等邏輯進行了分析。值得注意的是，由于我們近期正在對處理單元進行重構，因此源碼中會同時包含重構前后的相關代碼實現。

作者：張學程

本文為 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/
|-- 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 的邏輯處理流程，對應的入口代碼為 Relay.Process，主要步驟包括：

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

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

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

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

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

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

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

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

當不再需要讀取 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。

在 relay 處理單元中，對于從上游讀取到的 binlog event，我們需要判斷是否需要寫進 relay log file 及是否需要更新對應的 relay.meta 內的斷點信息。因此在通過 Reader interface 讀取到 binlog event 后，通過調用 Transformer interface 來對 binlog event 進行相關的轉換處理。

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

在 Transformer 中，我們期望能達到以下目標：

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

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

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

在嘗試對 binlog event 進行寫入時，對于不同類型的 binlog event，需要 進行不同的判斷處理。

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

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

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

因此，在處理 RotateEvent 寫入的 handleRotateEvent 方法中，主要包含以下操作：

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

判斷是否是 fake RotateEvent，如果是則跳過后續處理。

與當前 relay log file 的 size 及內部 event 進行比較，判斷如果將當前 event 寫入到文件后是否會造成文件存在 hole 及該 event 是否在 relay log file 中已經存在，如果會造成 hole 則需要填充該 hole，如果已經存在則跳過后續的處理。

將 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 中。

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

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

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

因此，在處理 FormatDescriptionEvent 的 handleFormatDescriptionEvent 方法中，主要包含以下操作：

關閉之前可能已經打開的 relay log file。

打開該 event 需要寫入到的 relay log file 作為當前活躍的 relay log file。

檢查當前 relay log file 中是否存在 binlog file header（ fe `bin` ），如果不存在則為其 寫入 binlog file header。

檢查當前 relay log file 中是否存在 FormatDescriptionEvent，如果不存在則為其 寫入該 FormatDescriptionEvent。

對于其他類型的 binlog event，寫入操作由 handleEventDefault 進行處理，主要包含以下操作：

與當前 relay log file 的 size 及內部 event 進行比較，判斷如果將當前 event 寫入到文件后是否會造成文件存在 hole 及該 event 是否在 relay log file 中已經存在，如果會造成 hole 則需要填充該 hole，如果已經存在則跳過后續的處理。

將 event 寫入到 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 方法中，主要操作包括：

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

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

如果 relay log file 的 size 比 binlog position 更小，則向外部報告錯誤并跳過后續的處理。

如果 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 前主要執行以下操作：

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

如果不是前一次連接過的 server，則說明切換到了新的 server，因此創建新的 relay log 子目錄并更新對應的 meta 信息。

relay 處理單元用于從上游讀取 binlog event 并將其寫入到本地的 relay log file 中。當執行增量數據復制時，binlog replication 處理單元需要通過 streamer pkg 讀取 relay log file 并從中解析獲取需要同步的數據，其中執行讀取的對象為 BinlogReader。

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

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

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

對于當前需要讀取的子目錄，調用 CollectBinlogFilesCmp 收集該目錄內指定 relay log 文件及其之后的所有 relay log 文件。

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

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

對于是否需要切換到新的 relay log 文件或子目錄的檢測通過在 parseFile 內 調用 needSwitchSubDir 與 調用 relaySubDirUpdated 實現。

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