DM 源碼閱讀系列文章（三）數據同步處理單元介紹

做者：lan

本文爲 DM 源碼閱讀系列文章的第三篇，上篇文章 介紹了 DM 的總體架構，DM 組件 DM-master 和 DM-worker 的入口代碼，以及二者之間的數據交互模型。本篇文章詳細地介紹 DM 數據同步處理單元（DM-worker 內部用來同步數據的邏輯單元），包括數據同步處理單元實現了什麼功能，數據同步流程、運行邏輯，以及數據同步處理單元的 interface 設計。

數據同步處理單元

從上圖能夠了解到目前 DM 包含 relay log、dump、load、binlog replication（sync） 4 個數據同步處理單元，涵蓋了如下數據同步處理的功能：

處理單元	功能
relay log	持久化 MySQL/MariaDB Binlog 到磁盤
dump	從 MySQL/MariaDB dump 全量數據
load	加載全量數據到 TiDB cluster
binlog replication（sync）	複製 relay log 存儲的 Binlog 到 TiDB cluster

數據同步流程

Task 數據同步流程初始化操做步驟：

1. DM-master 接收到 task，將 task 拆分紅 subtask 後 分發給對應的各個 DM-worker；

2. DM-worker 接收到 subtask 後 建立一個 subtask 對象，而後 初始化數據同步流程。

從 初始化數據同步流程 的代碼中咱們能夠看到，根據 task 配置項 task-mode 的不一樣，DM-worker 會初始化不一樣的數據同步流程：

task-mode	同步流程	須要的數據同步處理單元
all	全量同步 -> 增量數據同步	relay log、dump、load、binlog replication（sync）
full	全量同步	dump、load
incremental	增量同步	relay log，binlog replication（sync）

運行邏輯

DM 數據同步處理單元 interface 定義在 dm/unit，relay log、dump、load、binlog replication（sync）都實現了該 interface（golang interface 介紹）。

實際上 DM-worker 中的數據同步處理單元分爲兩類：

• 全局共享單例。dm-worker 啓動的時候只初始化一次這類數據同步處理單元，全部的 subtask 均可以使用這類數據同步處理單元的服務；relay log 屬於這種類型。

• subtask 獨享。dm-worker 會爲每一個 subtask 初始化一系列的數據同步處理單元；dump、load、binlog replication（sync）屬於這種類型。

兩類數據同步處理單元的使用邏輯不一樣，這篇文檔會着重講一下 subtask 獨享的數據同步處理單元的使用邏輯，不會囊括更多的 relay log 相關的內容，後面會有單獨一篇文章詳細介紹它。

relay log 相關使用代碼在 dm/worker/relay.go 、具體功能實現代碼在 relay/relay.go，有興趣的同窗也能夠先行閱讀一下相關代碼，relay log 的代碼註釋也是比較豐富，而且簡單易懂。

subtask 獨享數據同步處理單元使用邏輯相關代碼在 dm/worker/subtask.go。subtask 對象包含的主要屬性有：

• units：初始化後要運行的數據同步處理單元。

• currUnit：當前正在運行的數據同步處理單元。

• prevUnit：上一個運行的數據同步處理單元。

• stage：subtask 的運行階段狀態， 包含 New、Running、Paused，Stopped，Finished，具體定義的代碼在 dm/proto/dmworker.proto。

• result：subtask 當前數據同步處理單元的運行結果，對應着 stage = Paused/Stopped/Finished 的詳細信息。

主要的邏輯有：

• 初始化 subtask 對象實例的時候會 編排數據同步處理單元的運行前後順序。全部的數據同步處理單元都實現了 dm/unit interface，因此接下來的運行中就不須要關心具體的數據同步處理單元的類型，能夠按照統一的 interface 方法來運行數據同步處理單元，以及對其進行狀態監控。

• 初始化各個數據同步處理單元。subtask 在運行前集中地初始化全部的數據同步處理單元，咱們計劃以後優化成在各個數據同步處理單元運行前再進行初始化，這樣子減小資源的提早或者無效的佔用。

• 數據同步處理單元運行狀態監控。經過監控當前運行的數據同步處理單元的結果，將 subtask 的 stage 設置爲 Paused/Stopped/Finished。

  • 若是 當前的數據同步處理單元工做已經完成，則會根據 units 來 選取下一個須要運行的數據同步處理單元，若是沒有須要的數據同步處理單元，那麼會將 subtask 的 stage 設置爲 Finished。這裏有個注意點，由於 binlog replication 單元永遠不會結束，因此不會進入 Finished 的狀態。

  • 若是 返回的 result 裏面包含有錯誤信息，則會將 subtask 的 stage 設置爲 Paused，而且打印具體的錯誤信息。

  • 若是是用戶手動暫停或者中止，則會將 subtask 的 stage 設置爲 Paused/Stopped。這裏有個注意點，這個時候 stage=Paused 是沒有錯誤信息的。

• 數據同步處理單元之間的運行交接處理邏輯。部分數據同步處理單元在開始工做的時候須要知足一些前置條件，例如 binlog replication（sync）的運行須要等待 relay log 處理單元已經儲存下來其開始同步須要的 binlog 文件，不然 subtask 將處於 stage=Paused 的暫停等待狀態。

小結

本篇文章主要介紹了數據同步處理單元實現了什麼功能，數據同步流程、運行邏輯，以及數據同步處理單元的 interface 設計。後續會分三篇文章詳細地介紹數據同步處理單元的實現，包括：

• dump/load 全量同步實現
• binlog replication 增量同步實現
• relay log 實現