摘要：實際上中的數據同步處理單元分為兩類全局共享單例。獨享數據同步處理單元使用邏輯相關代碼在。數據同步處理單元運行狀態監控。后續會分三篇文章詳細地介紹數據同步處理單元的實現，包括全量同步實現增量同步實現實現

作者：lan

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

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

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

DM-master 接收到 task，將 task 拆分成 subtask 后 分發給對應的各個 DM-worker；

DM-worker 接收到 subtask 后 創建一個 subtask 對象，然后 初始化數據同步流程。

從 初始化數據同步流程 的代碼中我們可以看到，根據 task 配置項 task-mode 的不同，DM-worker 會初始化不同的數據同步流程：

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 實現