摘要：作者張學(xué)程本文為源碼閱讀系列文章的第六篇，在上篇文章中我們介紹了處理單元的實(shí)現(xiàn)，對(duì)在增量復(fù)制過(guò)程中的讀取過(guò)濾路由轉(zhuǎn)換以及執(zhí)行等邏輯進(jìn)行了分析。值得注意的是，由于我們近期正在對(duì)處理單元進(jìn)行重構(gòu)，因此源碼中會(huì)同時(shí)包含重構(gòu)前后的相關(guān)代碼實(shí)現(xiàn)。

作者：張學(xué)程

本文為 DM 源碼閱讀系列文章的第六篇，在 上篇文章 中我們介紹了 binlog replication 處理單元的實(shí)現(xiàn)，對(duì)在增量復(fù)制過(guò)程中 binlog event 的讀取、過(guò)濾、路由、轉(zhuǎn)換以及執(zhí)行等邏輯進(jìn)行了分析。

本篇文章我們將會(huì)對(duì) relay 數(shù)據(jù)處理單元的實(shí)現(xiàn)進(jìn)行詳細(xì)的講解。這個(gè)單元的作用是從上游 MySQL/MariaDB 讀取 binlog event 并寫入到本地的 relay log file 中；當(dāng)執(zhí)行增量復(fù)制任務(wù)時(shí)，binlog replication 處理單元將讀取 relay log file 中的 event 并在進(jìn)行解析后復(fù)制到下游的 TiDB 中。本篇文章的內(nèi)容包括 relay log 目錄結(jié)構(gòu)定義、relay log 數(shù)據(jù)的處理流程、主從切換支持、relay log 的讀取等邏輯。

值得注意的是，由于我們近期正在對(duì) relay 處理單元進(jìn)行重構(gòu)，因此源碼中會(huì)同時(shí)包含重構(gòu)前后的相關(guān)代碼實(shí)現(xiàn)。

一個(gè)已經(jīng)進(jìn)行過(guò)一次主從切換的 relay log 目錄結(jié)構(gòu)大致如下：

/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 目錄下，主要包含以下幾類文件或文件夾數(shù)據(jù)：

從上圖大致可以了解 relay log 的邏輯處理流程，對(duì)應(yīng)的入口代碼為 Relay.Process，主要步驟包括：

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

將讀取到的 binlog event 使用 binlog transformer 進(jìn)行轉(zhuǎn)換。

將轉(zhuǎn)換后的 binlog event 使用 binlog writer 以 relay log file 的形式存儲(chǔ)在本地。

當(dāng)需要將數(shù)據(jù)以增量的方式同步到下游 TiDB 時(shí)，binlog replication 通過(guò)使用 relay reader 從 relay log file 中讀取 binlog event。

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

當(dāng)前對(duì) Reader interface 的實(shí)現(xiàn)為 reader，它最終通過(guò) in 這個(gè) br.Reader interface 從上游讀取 binlog event。reader 的使用流程為：

調(diào)用 Start 啟動(dòng)讀取流程，并根據(jù)配置中是否啟用了 GTID 模式分別調(diào)用 setUpReaderByGTID 或 setUpReaderByPos 來(lái)啟動(dòng)下層的 br.Reader 對(duì)象。

調(diào)用 GetEvent 讀取 binlog event，具體為 調(diào)用下層的 GetEvent 方法 獲取 binlog event。

當(dāng)不再需要讀取 binlog event 時(shí)，調(diào)用 Close 關(guān)閉讀取操作。

從上面的流程可以看出，具體的 binlog event 讀取操作使用的是另一個(gè)下層的 br.Reader interface，當(dāng)前選擇的具體實(shí)現(xiàn) 為通過(guò) TCP 連接進(jìn)行讀取的 TCPReader。在 TCPReader 中，使用了 go-mysql 提供的 BinglogSyncer.StartSync 和 BinlogSyncer.StartSyncGTID 來(lái)啟動(dòng)以 binlog position 模式或 GTID sets 模式讀取 binlog event，并通過(guò) BinlogStreamer.GetEvent 讀取來(lái)自 TCP 的 binlog event。

在 relay 處理單元中，對(duì)于從上游讀取到的 binlog event，我們需要判斷是否需要寫進(jìn) relay log file 及是否需要更新對(duì)應(yīng)的 relay.meta 內(nèi)的斷點(diǎn)信息。因此在通過(guò) Reader interface 讀取到 binlog event 后，通過(guò)調(diào)用 Transformer interface 來(lái)對(duì) binlog event 進(jìn)行相關(guān)的轉(zhuǎn)換處理。

當(dāng)前對(duì) Transformer interface 的實(shí)現(xiàn)為 transformer，其主要通過(guò)在 Transform 方法中 對(duì) binlog event 的類型進(jìn)行判斷 后再進(jìn)行相應(yīng)處理，包括：

在 Transformer 中，我們期望能達(dá)到以下目標(biāo)：

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

僅在 DDL QueryEvent 時(shí)或 DML 事務(wù)完成時(shí)更新 relay.meta 以確保中斷恢復(fù)時(shí)能避免從 DML 事務(wù)進(jìn)行中的 binlog event 處開(kāi)始從上游請(qǐng)求 binlog event（對(duì)于 DML 相關(guān)的 binlog event，如果希望解析 INSERT/UPDATE/DELETE 等操作，則需要先獲取到對(duì)應(yīng)的 TableMap event）。

在從上游讀取到 binlog event 并對(duì)其進(jìn)行了相關(guān)轉(zhuǎn)換后，我們就可以嘗試將其寫入到本地的 relay log file 中。在 relay 處理單元中，用于將 binlog event 寫入 relay log file 的是 Writer interface，當(dāng)前對(duì)應(yīng)的實(shí)現(xiàn)為 FileWriter，其內(nèi)部會(huì)使用 out 這個(gè) bw.FileWriter 來(lái)執(zhí)行文件寫入操作，具體對(duì) binlog event 執(zhí)行寫入操作的是 WriteEvent 方法。

在嘗試對(duì) binlog event 進(jìn)行寫入時(shí)，對(duì)于不同類型的 binlog event，需要 進(jìn)行不同的判斷處理。

在從上游讀取 binlog event 時(shí)，主要在以下情況下可能會(huì)讀取到 RotateEvent：

連接到上游 master server 開(kāi)始讀取 binlog event 時(shí)，master 會(huì)發(fā)送一個(gè) fake RotateEvent 告知 slave 后續(xù) binlog event 對(duì)應(yīng)的起始 binlog position。

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

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

嘗試更新 FileWriter 內(nèi)部記錄的當(dāng)前 binlog 文件名為 RotateEvent 內(nèi)包含的文件名。

判斷是否是 fake RotateEvent，如果是則跳過(guò)后續(xù)處理。

與當(dāng)前 relay log file 的 size 及內(nèi)部 event 進(jìn)行比較，判斷如果將當(dāng)前 event 寫入到文件后是否會(huì)造成文件存在 hole 及該 event 是否在 relay log file 中已經(jīng)存在，如果會(huì)造成 hole 則需要填充該 hole，如果已經(jīng)存在則跳過(guò)后續(xù)的處理。

將 event 寫入到 relay log file 中。

需要注意的是，我們不能確保 master server 會(huì)將其 binlog file 中的所有 event 都發(fā)送給 slave（如當(dāng) MariaDB 未設(shè)置 BINLOG_SEND_ANNOTATE_ROWS_EVENT flag 時(shí)，master 就不會(huì)向 slave 發(fā)送 ANNOTATE_ROWS_EVENT），因此在寫入 event 到文件前，需要通過(guò) handleFileHoleExist 判斷如果將 event 寫入到文件是否會(huì)存在 hole。如果存在 hode，則通過(guò) event.GenDummyEvent 生成相應(yīng) size 的 dummy event 對(duì) hole 進(jìn)行填充。

另外需要注意的是，我們不能確保 master server 不會(huì)將其已經(jīng)發(fā)送給 slave 并寫入到了 relay log file 的 event 再次發(fā)送給 slave（如 master 在開(kāi)始發(fā)送 slave 請(qǐng)求的 binlog event 前，會(huì)先發(fā)送 FormatDescriptionEvent 與 PreviousGTIDsEvent 等給 slave），因此在寫入 event 到文件前，需要通過(guò) handleDuplicateEventsExist 判斷該 event 是否已經(jīng)存在于 relay log file 中。

在從上游讀取 binlog event 時(shí)，主要在以下情況下可能會(huì)讀取到 FormatDescriptionEvent：

上游 master server 在發(fā)送除 RotateEvent 外的其他 binlog event 之前，會(huì)發(fā)送一個(gè) FormatDescriptionEvent 以使 slave 能正確 decode 后續(xù)的 binlog event。

上游 master server 會(huì)將自身 binlog file 中存在的 FormatDescriptionEvent 發(fā)送給 slave，且這個(gè) FormatDescriptionEvent 總是 binlog file 中的第 1 個(gè) event。

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

關(guān)閉之前可能已經(jīng)打開(kāi)的 relay log file。

打開(kāi)該 event 需要寫入到的 relay log file 作為當(dāng)前活躍的 relay log file。

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

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

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

與當(dāng)前 relay log file 的 size 及內(nèi)部 event 進(jìn)行比較，判斷如果將當(dāng)前 event 寫入到文件后是否會(huì)造成文件存在 hole 及該 event 是否在 relay log file 中已經(jīng)存在，如果會(huì)造成 hole 則需要填充該 hole，如果已經(jīng)存在則跳過(guò)后續(xù)的處理。

將 event 寫入到 relay log file 中。

在寫入 binlog event 到 relay log file 時(shí)，盡管可以通過(guò) Flush 方法強(qiáng)制將緩沖中的數(shù)據(jù)刷新到磁盤文件中，但仍然可能出現(xiàn) DM-worker 進(jìn)程異常退出時(shí)部分?jǐn)?shù)據(jù)未能刷新到磁盤文件中的情況，造成 relay log file 內(nèi)部分 event 數(shù)據(jù)缺失。

另外，對(duì)于一個(gè)事務(wù)對(duì)應(yīng)的多個(gè) binlog event，可能出現(xiàn)僅寫入了其中一部分 event 時(shí) DM-worker 發(fā)生退出的情況，造成 relay log file 中部分事務(wù)缺失部分 event。

因此，在 relay 處理單元中，我們引入了對(duì) relay log file 執(zhí)行 Recover 的機(jī)制，用于將 relay log file 尾部不完整的 event 及事務(wù)進(jìn)行踢除，對(duì)應(yīng)的方法為 FileWrite.Recover，具體實(shí)現(xiàn)在 doRecovering 方法中，主要操作包括：

獲取 relay log file 中直到最后一個(gè)完整事務(wù)對(duì)應(yīng)的 binlog position 與 GTID sets。

比較 relay log file 的 size 與獲取到的 binlog position，如果相等則說(shuō)明這個(gè) relay log file 中包含的事務(wù)都是完整的，跳過(guò)后續(xù)的處理。

如果 relay log file 的 size 比 binlog position 更小，則向外部報(bào)告錯(cuò)誤并跳過(guò)后續(xù)的處理。

如果 relay log file 的 size 比 binlog position 大，則 將 relay log file 中超出 binlog position 的部分執(zhí)行 Truncate 進(jìn)行截?cái)唷?/p> 主從切換支持

為支持將 relay 處理單元連接的上游 master server 在 replica group 內(nèi)的不同 server 間進(jìn)行切換（也包括 relay 處理單元連接的上游 VIP 指向的實(shí)際 server 發(fā)生了改變），relay 處理單元會(huì)嘗試將從不同上游 server 讀取到的 binlog event 保存到不同的 relay log 子目錄中，目錄與文件結(jié)構(gòu)可以參考前文的 relay log 目錄結(jié)構(gòu)。

為支持上述功能，relay 處理單元在讀取 binlog event 前主要執(zhí)行以下操作：

比較當(dāng)前上游 server 的 UUID 信息與 relay.meta 信息，判斷當(dāng)前連接到的是否是前一次連接過(guò)的 server。

如果不是前一次連接過(guò)的 server，則說(shuō)明切換到了新的 server，因此創(chuàng)建新的 relay log 子目錄并更新對(duì)應(yīng)的 meta 信息。

relay 處理單元用于從上游讀取 binlog event 并將其寫入到本地的 relay log file 中。當(dāng)執(zhí)行增量數(shù)據(jù)復(fù)制時(shí)，binlog replication 處理單元需要通過(guò) streamer pkg 讀取 relay log file 并從中解析獲取需要同步的數(shù)據(jù)，其中執(zhí)行讀取的對(duì)象為 BinlogReader。

由前文介紹過(guò)的主從切換支持可知我們會(huì)將具體的 relay log 數(shù)據(jù)存儲(chǔ)在可能的多個(gè)子目錄中，因此在讀取 relay log 時(shí)，我們也 需要考慮按序依次讀取，主要操作包括：

調(diào)用 parseRelay 開(kāi)始從 relay log 的根目錄執(zhí)行解析讀取。

調(diào)用 parseDirAsPossible 開(kāi)始從外部指定的或上一次調(diào)用返回的子目錄、文件及 offset 處開(kāi)始讀取，并返回下一次調(diào)用時(shí)需要的子目錄、文件及 offset（即可實(shí)現(xiàn)切換到新的 relay log 子目錄）。

對(duì)于當(dāng)前需要讀取的子目錄，調(diào)用 CollectBinlogFilesCmp 收集該目錄內(nèi)指定 relay log 文件及其之后的所有 relay log 文件。

對(duì)于每一個(gè)收集到的 relay log 文件，調(diào)用 parseFileAsPossible 嘗試對(duì)其進(jìn)行解析讀取。

在 parseFileAsPossible 中，反復(fù)返回 調(diào)用 parseFile 進(jìn)行 binlog event 的讀取，直到 發(fā)生錯(cuò)誤 或 檢測(cè)到需要切換到新的 relay log 文件或子目錄。

對(duì)于是否需要切換到新的 relay log 文件或子目錄的檢測(cè)通過(guò)在 parseFile 內(nèi) 調(diào)用 needSwitchSubDir 與 調(diào)用 relaySubDirUpdated 實(shí)現(xiàn)。

本篇文章詳細(xì)地介紹了 relay 處理單元的實(shí)現(xiàn)，內(nèi)容包括了 relay log 的目錄結(jié)構(gòu)、如何從上游 server 讀取 binlog event 并寫入到本地的 relay log file 中，以及 binlog replication 處理單元將如何讀取本地的 relay log file。到本篇文章為止，我們完成了對(duì) DM 中的數(shù)據(jù)處理單元的介紹。從下一篇文章開(kāi)始，我們將開(kāi)始詳細(xì)介紹 DM 內(nèi)部主要功能的設(shè)計(jì)與實(shí)現(xiàn)原理。