摘要：實(shí)際上中的數(shù)據(jù)同步處理單元分為兩類全局共享單例。獨(dú)享數(shù)據(jù)同步處理單元使用邏輯相關(guān)代碼在。數(shù)據(jù)同步處理單元運(yùn)行狀態(tài)監(jiān)控。后續(xù)會(huì)分三篇文章詳細(xì)地介紹數(shù)據(jù)同步處理單元的實(shí)現(xiàn)，包括全量同步實(shí)現(xiàn)增量同步實(shí)現(xiàn)實(shí)現(xiàn)

作者：lan

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

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

Task 數(shù)據(jù)同步流程初始化操作步驟：

DM-master 接收到 task，將 task 拆分成 subtask 后 分發(fā)給對(duì)應(yīng)的各個(gè) DM-worker；

DM-worker 接收到 subtask 后 創(chuàng)建一個(gè) subtask 對(duì)象，然后 初始化數(shù)據(jù)同步流程。

從 初始化數(shù)據(jù)同步流程 的代碼中我們可以看到，根據(jù) task 配置項(xiàng) task-mode 的不同，DM-worker 會(huì)初始化不同的數(shù)據(jù)同步流程：

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

實(shí)際上 DM-worker 中的數(shù)據(jù)同步處理單元分為兩類：

全局共享單例。dm-worker 啟動(dòng)的時(shí)候只初始化一次這類數(shù)據(jù)同步處理單元，所有的 subtask 都可以使用這類數(shù)據(jù)同步處理單元的服務(wù)；relay log 屬于這種類型。

subtask 獨(dú)享。dm-worker 會(huì)為每個(gè) subtask 初始化一系列的數(shù)據(jù)同步處理單元；dump、load、binlog replication（sync）屬于這種類型。

兩類數(shù)據(jù)同步處理單元的使用邏輯不同，這篇文檔會(huì)著重講一下 subtask 獨(dú)享的數(shù)據(jù)同步處理單元的使用邏輯，不會(huì)囊括更多的 relay log 相關(guān)的內(nèi)容，后面會(huì)有多帶帶一篇文章詳細(xì)介紹它。

relay log 相關(guān)使用代碼在 dm/worker/relay.go 、具體功能實(shí)現(xiàn)代碼在 relay/relay.go，有興趣的同學(xué)也可以先行閱讀一下相關(guān)代碼，relay log 的代碼注釋也是比較豐富，并且簡(jiǎn)單易懂。

subtask 獨(dú)享數(shù)據(jù)同步處理單元使用邏輯相關(guān)代碼在 dm/worker/subtask.go。subtask 對(duì)象包含的主要屬性有：

units：初始化后要運(yùn)行的數(shù)據(jù)同步處理單元。

currUnit：當(dāng)前正在運(yùn)行的數(shù)據(jù)同步處理單元。

prevUnit：上一個(gè)運(yùn)行的數(shù)據(jù)同步處理單元。

stage：subtask 的運(yùn)行階段狀態(tài)， 包含 New、Running、Paused，Stopped，Finished，具體定義的代碼在 dm/proto/dmworker.proto。

result：subtask 當(dāng)前數(shù)據(jù)同步處理單元的運(yùn)行結(jié)果，對(duì)應(yīng)著 stage = Paused/Stopped/Finished 的詳細(xì)信息。

主要的邏輯有：

初始化 subtask 對(duì)象實(shí)例的時(shí)候會(huì) 編排數(shù)據(jù)同步處理單元的運(yùn)行先后順序。所有的數(shù)據(jù)同步處理單元都實(shí)現(xiàn)了 dm/unit interface，所以接下來的運(yùn)行中就不需要關(guān)心具體的數(shù)據(jù)同步處理單元的類型，可以按照統(tǒng)一的 interface 方法來運(yùn)行數(shù)據(jù)同步處理單元，以及對(duì)其進(jìn)行狀態(tài)監(jiān)控。

初始化各個(gè)數(shù)據(jù)同步處理單元。subtask 在運(yùn)行前集中地初始化所有的數(shù)據(jù)同步處理單元，我們計(jì)劃之后優(yōu)化成在各個(gè)數(shù)據(jù)同步處理單元運(yùn)行前再進(jìn)行初始化，這樣子減少資源的提前或者無效的占用。

數(shù)據(jù)同步處理單元運(yùn)行狀態(tài)監(jiān)控。通過監(jiān)控當(dāng)前運(yùn)行的數(shù)據(jù)同步處理單元的結(jié)果，將 subtask 的 stage 設(shè)置為 Paused/Stopped/Finished。

如果 當(dāng)前的數(shù)據(jù)同步處理單元工作已經(jīng)完成，則會(huì)根據(jù) units 來 選取下一個(gè)需要運(yùn)行的數(shù)據(jù)同步處理單元，如果沒有需要的數(shù)據(jù)同步處理單元，那么會(huì)將 subtask 的 stage 設(shè)置為 Finished。這里有個(gè)注意點(diǎn)，因?yàn)?binlog replication 單元永遠(yuǎn)不會(huì)結(jié)束，所以不會(huì)進(jìn)入 Finished 的狀態(tài)。

如果 返回的 result 里面包含有錯(cuò)誤信息，則會(huì)將 subtask 的 stage 設(shè)置為 Paused，并且打印具體的錯(cuò)誤信息。

如果是用戶手動(dòng)暫?；蛘咄Ｖ?，則會(huì)將 subtask 的 stage 設(shè)置為 Paused/Stopped。這里有個(gè)注意點(diǎn)，這個(gè)時(shí)候 stage=Paused 是沒有錯(cuò)誤信息的。

數(shù)據(jù)同步處理單元之間的運(yùn)行交接處理邏輯。部分?jǐn)?shù)據(jù)同步處理單元在開始工作的時(shí)候需要滿足一些前置條件，例如 binlog replication（sync）的運(yùn)行需要等待 relay log 處理單元已經(jīng)儲(chǔ)存下來其開始同步需要的 binlog 文件，否則 subtask 將處于 stage=Paused 的暫停等待狀態(tài)。

本篇文章主要介紹了數(shù)據(jù)同步處理單元實(shí)現(xiàn)了什么功能，數(shù)據(jù)同步流程、運(yùn)行邏輯，以及數(shù)據(jù)同步處理單元的 interface 設(shè)計(jì)。后續(xù)會(huì)分三篇文章詳細(xì)地介紹數(shù)據(jù)同步處理單元的實(shí)現(xiàn)，包括：

dump/load 全量同步實(shí)現(xiàn)

binlog replication 增量同步實(shí)現(xiàn)

relay log 實(shí)現(xiàn)