摘要：我們將抽象成了一個(gè)簡單的數(shù)據(jù)庫，為的或者，為的元數(shù)據(jù)，的數(shù)據(jù)則存在數(shù)據(jù)文件中。元數(shù)據(jù)中提供了數(shù)據(jù)存儲(chǔ)的文件和位置，可以通過這些信息讀取文件的指定位置獲取到數(shù)據(jù)。

作者：王相

TiDB-Binlog 組件用于收集 TiDB 的 binlog，并提供實(shí)時(shí)備份和同步功能。該組件在功能上類似于 MySQL 的主從復(fù)制，MySQL 的主從復(fù)制依賴于記錄的 binlog 文件，TiDB-Binlog 組件也是如此，主要的不同點(diǎn)是 TiDB 是分布式的，因此需要收集各個(gè) TiDB 實(shí)例產(chǎn)生的 binlog，并按照事務(wù)提交的時(shí)間排序后才能同步到下游。如果你需要部署 TiDB 集群的從庫，或者想訂閱 TiDB 數(shù)據(jù)的變更輸出到其他的系統(tǒng)中，TiDB-Binlog 則是必不可少的工具。

TiDB-Binlog 這個(gè)組件已經(jīng)發(fā)布了 2 年多時(shí)間，經(jīng)歷過幾次架構(gòu)演進(jìn)，去年十月到現(xiàn)在大規(guī)模使用的是 Kafka 版本，架構(gòu)圖如下：

Kafka 版本的 TiDB-Binlog 主要包括兩個(gè)組件：

Pump：一個(gè)守護(hù)進(jìn)程，在每個(gè) TiDB 主機(jī)的后臺(tái)運(yùn)行。其主要功能是實(shí)時(shí)記錄 TiDB 產(chǎn)生的 binlog 并順序?qū)懭?Kafka 中。

Drainer： 從 Kafka 中收集 binlog，并按照 TiDB 中事務(wù)的提交順序轉(zhuǎn)化為指定數(shù)據(jù)庫兼容的 SQL 語句或者指定格式的數(shù)據(jù)，最后同步到目的數(shù)據(jù)庫或者寫到順序文件。

這個(gè)架構(gòu)的工作原理為：

TiDB 需要與 Pump 綁定，即 TiDB 實(shí)例只能將它生成的 binlog 發(fā)送到一個(gè)指定的 Pump 中；

Pump 將 binlog 先寫到本地文件，再異步地寫入到 Kafka；

Drainer 從 Kafka 中讀出 binlog，對(duì) binlog 進(jìn)行排序，對(duì) binlog 解析后生成 SQL 或指定格式的數(shù)據(jù)再同步到下游。

根據(jù)用戶的反饋，以及我們自己做的一些測(cè)試，發(fā)現(xiàn)該版本主要存在一些問題。

首先，TiDB 的負(fù)載可能不均衡，部分 TiDB 業(yè)務(wù)較多，產(chǎn)生的 binlog 也比較多，對(duì)應(yīng)的 Pump 的負(fù)載高，導(dǎo)致數(shù)據(jù)同步延遲高。

其次，依賴 Kafka 集群，增加了運(yùn)維成本；而且 TiDB 產(chǎn)生的單條 binlog 的大小可達(dá) 2G（例如批量刪除數(shù)據(jù)、批量寫入數(shù)據(jù)），需要配置 Kafka 的消息大小相關(guān)設(shè)置，而 Kafka 并不太適合單條數(shù)據(jù)較大的場(chǎng)景。

最后，Drainer 需要讀取 Kafka 中的 binlog、對(duì) binlog 進(jìn)行排序、解析 binlog，同步數(shù)據(jù)到下游等工作，可以看出 Drainer 的工作較多，而且 Drainer 是一個(gè)單點(diǎn)，所以往往同步數(shù)據(jù)的瓶頸都在 Drainer。

以上這些問題我們很難在已有的框架下進(jìn)行優(yōu)化，因此我們對(duì) TiDB-Binlog 進(jìn)行了重構(gòu)，最新版本的 TiDB-Binlog 的總體架構(gòu)如下圖所示：

新版本 TiDB-Binlog 不再使用 Kafka 存儲(chǔ) binlog，仍然保留了 Pump 和 Drainer 兩個(gè)組件，但是對(duì)功能進(jìn)行了調(diào)整：

Pump 用于實(shí)時(shí)記錄 TiDB 產(chǎn)生的 binlog，并將 binlog 按照事務(wù)的提交時(shí)間進(jìn)行排序，再提供給 Drainer 進(jìn)行消費(fèi)。

Drainer 從各個(gè) Pump 中收集 binlog 進(jìn)行歸并，再將 binlog 轉(zhuǎn)化成 SQL 或者指定格式的數(shù)據(jù)，最終同步到下游。

該版本的主要優(yōu)點(diǎn)為：

多個(gè) Pump 形成一個(gè)集群，可以水平擴(kuò)容，各個(gè) Pump 可以均勻地承擔(dān)業(yè)務(wù)的壓力。

TiDB 通過內(nèi)置的 Pump Client 將 binlog 分發(fā)到各個(gè) Pump，即使有部分 Pump 出現(xiàn)故障也不影響 TiDB 的業(yè)務(wù)。

Pump 內(nèi)部實(shí)現(xiàn)了簡單的 kv 來存儲(chǔ) binlog，方便對(duì) binlog 數(shù)據(jù)的管理。

原來 Drainer 的 binlog 排序邏輯移到了 Pump 來做，而 Pump 是可擴(kuò)展的，這樣就能提高整體的同步性能。

Drainer 不再需要像原來一樣讀取一批 binlog 到內(nèi)存里進(jìn)行堆排序，只需要依次讀取各個(gè) Pump 的 binlog 進(jìn)行歸并排序，這樣可以大大節(jié)省內(nèi)存的使用，同時(shí)也更容易做內(nèi)存控制。

由于該版本最大的特點(diǎn)是多個(gè) Pump 組成了一個(gè)集群（cluster），因此該版本命名為 cluster 版本。下面我們以最新的 cluster 版本的架構(gòu)來介紹 TiDB-Binlog 的實(shí)現(xiàn)原理。

首先我們先介紹一下 TiDB 中的 binlog，TiDB 的事務(wù)采用 2pc 算法，一個(gè)成功的事務(wù)會(huì)寫兩條 binlog，包括一條 Prewrite binlog 和 一條 Commit binlog；如果事務(wù)失敗，會(huì)發(fā)一條 Rollback binlog。

binlog 的結(jié)構(gòu)定義為：

// Binlog 記錄事務(wù)中所有的變更，可以用 Binlog 構(gòu)建 SQL
message Binlog {
    // Binlog 的類型，包括 Prewrite、Commit、Rollback 等
    optional BinlogType  tp = 1 [(gogoproto.nullable) = false];
    
    // Prewrite, Commit 和 Rollback 類型的 binlog 的 start_ts，記錄事務(wù)開始的 ts
    optional int64  start_ts = 2 [(gogoproto.nullable) = false];
    
    // commit_ts 記錄事務(wù)結(jié)束的 ts，只記錄在 commit 類型的 binlog 中
    optional int64  commit_ts = 3 [(gogoproto.nullable) = false];
    
    // prewrite key 只記錄在 Prewrite 類型的 binlog 中，
    // 是一個(gè)事務(wù)的主鍵，用于查詢?cè)撌聞?wù)是否提交
    optional bytes  prewrite_key = 4;
    
    // prewrite_value 記錄在 Prewrite 類型的 binlog 中，用于記錄每一行數(shù)據(jù)的改變
    optional bytes  prewrite_value = 5;
    
    // ddl_query 記錄 ddl 語句
    optional bytes  ddl_query = 6;
    
    // ddl_job_id 記錄 ddl 的 job id
    optional int64  ddl_job_id  = 7 [(gogoproto.nullable) = false];
}

binlog 及相關(guān)的數(shù)據(jù)結(jié)構(gòu)定義見: binlog.proto

其中 start_ts 為事務(wù)開始時(shí)的 ts，commit_ts 為事務(wù)提交的 ts。ts 是由物理時(shí)間和邏輯時(shí)間轉(zhuǎn)化而成的，在 TiDB 中是唯一的，由 PD 來統(tǒng)一提供。在開始一個(gè)事務(wù)時(shí)，TiDB 會(huì)請(qǐng)求 PD，獲取一個(gè) ts 作為事務(wù)的 start_ts，在事務(wù)提交時(shí)則再次請(qǐng)求 PD 獲取一個(gè) ts 作為 commit_ts。 我們?cè)?Pump 和 Drainer 中就是根據(jù) binlog 的 commit_ts 來對(duì) binlog 進(jìn)行排序的。

TiDB 的 binlog 記錄為 row 模式，即保存每一行數(shù)據(jù)的改變。數(shù)據(jù)的變化記錄在 ?prewrite_value 字段中，該字段的數(shù)據(jù)主要由序列化后的 TableMutation 結(jié)構(gòu)的數(shù)據(jù)組成。TableMutation 的結(jié)構(gòu)如下所示：

// TableMutation 存儲(chǔ)表中數(shù)據(jù)的變化
message TableMutation {
    // 表的 id，唯一標(biāo)識(shí)一個(gè)表
    optional int64 table_id = 1 [(gogoproto.nullable) = false];
    
    // 保存插入的每行數(shù)據(jù)
    repeated bytes inserted_rows = 2;
    
    // 保存修改前和修改后的每行的數(shù)據(jù)
    repeated bytes updated_rows = 3;
    
    // 已廢棄
    repeated int64 deleted_ids = 4;
    
    // 已廢棄
    repeated bytes deleted_pks = 5;
    
    // 刪除行的數(shù)據(jù)
    repeated bytes deleted_rows  = 6;
    
    // 記錄數(shù)據(jù)變更的順序
    repeated MutationType sequence = 7;
}

下面以一個(gè)例子來說明 binlog 中是怎么存儲(chǔ)數(shù)據(jù)的變化的。

例如 table 的結(jié)構(gòu)為：

create table test (id int, name varchar(24), primary key id)

按照順序執(zhí)行如下 SQL：

begin;
insert into test(id, name) values(1, "a");
insert into test(id, name) values(2, "b");
update test set name = "c" where id = 1;
update test set name = "d" where id = 2;
delete from test where id = 2;
insert into test(id, name) values(2, "c");
commit;

則生成的 TableMutation 的數(shù)據(jù)如下所示：

inserted_rows:
1, "a"
2, "b"
2, "c"
 
updated_rows:
1, "a", 1, "c"
2, "b", 2, "d"
 
deleted_rows:
2, "d"
 
sequence:
Insert, Insert, Update, Update, DeleteRow, Insert

可以從例子中看出，sequence 中保存的數(shù)據(jù)變更類型的順序?yàn)閳?zhí)行 SQL 的順序，具體變更的數(shù)據(jù)內(nèi)容則保存到了相應(yīng)的變量中。

Drainer 在把 binlog 數(shù)據(jù)同步到下游前，就需要把上面的這些數(shù)據(jù)還原成 SQL，再同步到下游。

另外需要說明的是，TiDB 在寫 binlog 時(shí)，會(huì)同時(shí)向 TiKV 發(fā)起寫數(shù)據(jù)請(qǐng)求和向 Pump 發(fā)送 Prewrite binlog，如果 TiKV 和 Pump 其中一個(gè)請(qǐng)求失敗，則該事務(wù)失敗。當(dāng) Prewrite 成功后，TiDB 向 TiKV 發(fā)起 Commit 消息，并異步地向 Pump 發(fā)送一條 Commit binlog。由于 TiDB 是同時(shí)向 TiKV 和 Pump 發(fā)送請(qǐng)求的，所以只要保證 Pump 處理 Prewrite binlog 請(qǐng)求的時(shí)間小于等于 TiKV 執(zhí)行 Prewrite 的時(shí)間，開啟 binlog 就不會(huì)對(duì)事務(wù)的延遲造成影響。

從上面的介紹中我們知道由多個(gè) Pump 組成一個(gè)集群，共同承擔(dān)寫 binlog 的請(qǐng)求，那么就需要保證 TiDB 能夠?qū)?binlog 的請(qǐng)求盡可能均勻地分發(fā)到各個(gè) Pump，并且需要識(shí)別不可用的 Pump，及時(shí)獲取到新加入集群中 Pump 信息。這部分的工作是在 Pump Client 中實(shí)現(xiàn)的。

Pump Client 以包的形式集成在 TiDB 中，代碼鏈接：pump_client。

Pump Client 維護(hù) Pump 集群的信息，Pump 的信息主要來自于 PD 中保存的 Pump 的狀態(tài)信息，狀態(tài)信息的定義如下（代碼鏈接：Status）：

type Status struct {
    // Pump/Drainer 實(shí)例的唯一標(biāo)識(shí)
    NodeID string `json:"nodeId"`
    
    // Pump/Drainer 的服務(wù)地址
    Addr string `json:"host"`
    
    // Pump/Drainer 的狀態(tài)，值可以為 online、pausing、paused、closing、offline
    State string `json:"state"`
    
    // Pump/Drainer 是否 alive（目前沒有使用該字段）
    IsAlive bool `json:"isAlive"`
    
    // Pump的分?jǐn)?shù)，該分?jǐn)?shù)是由節(jié)點(diǎn)的負(fù)載、磁盤使用率、存儲(chǔ)的數(shù)據(jù)量大小等因素計(jì)算得來的，
    // 這樣 Pump Client 可以根據(jù)分?jǐn)?shù)來選取合適的 Pump 發(fā)送 binlog（待實(shí)現(xiàn)）
    Score int64 `json:"score"`
    
    // Pump 的標(biāo)簽，可以通過 label 對(duì) TiDB 和 Pump 進(jìn)行分組，
    // TiDB 只能將 binlog 發(fā)送到相同 label 的 Pump（待實(shí)現(xiàn)）
    Label *Label `json:"label"`
    
    // Pump： 保存的 binlog 的最大的 commit_ts
    // Drainer：已消費(fèi)的 binlog 的最大的 commit_ts
    MaxCommitTS int64 `json:"maxCommitTS"`
    
    // 該狀態(tài)信息的更新時(shí)間對(duì)應(yīng)的 ts.
    UpdateTS int64 `json:"updateTS"`
}

Pump Client 根據(jù) Pump 上報(bào)到 PD 的信息以及寫 binlog 請(qǐng)求的實(shí)際情況將 Pump 劃分為可用 Pump 與不可用 Pump 兩個(gè)部分。

劃分的方法包括：

初始化時(shí)從 PD 中獲取所有 Pump 的信息，將狀態(tài)為 online 的 Pump 加入到可用 Pump 列表中，其他 Pump 加入到非可用列表中。

Pump 每隔固定的時(shí)間會(huì)發(fā)送心跳到 PD，并更新自己的狀態(tài)。Pump Client 監(jiān)控 PD 中 Pump 上傳的狀態(tài)信息，及時(shí)更新內(nèi)存中維護(hù)的 Pump 信息，如果狀態(tài)由非 online 轉(zhuǎn)換為 online 則將該 Pump 加入到可用 Pump 列表；反之加入到非可用列表中。

在寫 binlog 到 Pump 時(shí)，如果該 Pump 在重試多次后仍然寫 binlog 失敗，則把該 Pump 加入到非可用 Pump 列表中。

定時(shí)發(fā)送探活請(qǐng)求（數(shù)據(jù)為空的 binlog 寫請(qǐng)求）到非可用 Pump 列表中的狀態(tài)為 online 的 Pump，如果返回成功，則把該 Pump 重新加入到可用 Pump 列表中。

通過上面的這些措施，Pump Client 就可以及時(shí)地更新所維護(hù)的 Pump 集群信息，保證將 binlog 發(fā)送到可用的 Pump 中。

另外一個(gè)問題是，怎么保證 Pump Client 可以將 binlog 寫請(qǐng)求均勻地分發(fā)到各個(gè) Pump？我們目前提供了幾種路由策略：

range： 按照順序依次選取 Pump 發(fā)送 binlog，即第一次選取第一個(gè) Pump，第二次選取第二個(gè) Pump...

hash：對(duì) binlog 的 start_ts 進(jìn)行 hash，然后選取 hash 值對(duì)應(yīng)的 Pump。

score：根據(jù) Pump 上報(bào)的分?jǐn)?shù)按照加權(quán)平均算法選取 Pump 發(fā)送 binlog（待實(shí)現(xiàn)）。

需要注意的地方是，以上的策略只是針對(duì) Prewrite binlog，對(duì)于 Commit binlog，Pump Client 會(huì)將它發(fā)送到對(duì)應(yīng)的 Prewrite binlog 所選擇的 Pump，這樣做是因?yàn)樵?Pump 中需要將包含 Prewrite binlog 和 Commit binlog 的完整 binlog（即執(zhí)行成功的事務(wù)的 binlog）提供給 Drainer，將 Commit binlog 發(fā)送到其他 Pump 沒有意義。

Pump Client 向 Pump 提交寫 binlog 的請(qǐng)求接口為 pump.proto 中的 WriteBinlog，使用 grpc 發(fā)送 binlog 請(qǐng)求。

Pump 主要用來承擔(dān) binlog 的寫請(qǐng)求，維護(hù) binlog 數(shù)據(jù)，并將有序的 binlog 提供給 Drainer。我們將 Pump 抽象成了一個(gè)簡單的 kv 數(shù)據(jù)庫，key 為 binlog 的 start _ts（Priwrite binlog） 或者 commit_ts（Commit binlog），value 為 binlog 的元數(shù)據(jù)，binlog 的數(shù)據(jù)則存在數(shù)據(jù)文件中。Drainer 像查數(shù)據(jù)庫一樣的來獲取所需要的 binlog。

Pump 內(nèi)置了 leveldb 用于存儲(chǔ) binlog 的元信息。在 Pump 收到 binlog 的寫請(qǐng)求時(shí)，會(huì)首先將 binlog 數(shù)據(jù)以 append 的形式寫到文件中，然后將 binlog 的 ts、類型、數(shù)據(jù)長度、所保存的文件以及在文件中的位置信息保存在 leveldb 中，如果為 Prewrite binlog，則以 start_ts作為 key；如果是 Commit binlog，則以 commit_ts 作為 key。

當(dāng) Drainer 向 Pump 請(qǐng)求獲取指定 ts 之后的 binlog 時(shí)，Pump 則查詢 leveldb 中大于該 ts 的 binlog 的元數(shù)據(jù)，如果當(dāng)前數(shù)據(jù)為 Prewrite binlog，則必須找到對(duì)應(yīng)的 Commit binlog；如果為 Commit binlog 則繼續(xù)向前推進(jìn)。這里有個(gè)問題，在 binlog 一節(jié)中提到，如果 TiKV 成功寫入了數(shù)據(jù)，并且 Pump 成功接收到了 Prewrite binlog，則該事務(wù)就提交成功了，那么如果在 TiDB 發(fā)送 Commit binlog 到 Pump 前發(fā)生了一些異常（例如 TiDB 異常退出，或者強(qiáng)制終止了 TiDB 進(jìn)程），導(dǎo)致 Pump 沒有接收到 Commit binlog，那么 Pump 中就會(huì)一直找不到某些 Prewrite binlog 對(duì)應(yīng)的 Commit binlog。這里我們?cè)?Pump 中做了處理，如果某個(gè) Prewrite binlog 超過了十分鐘都沒有找到對(duì)應(yīng)的 Commit binlog，則通過 binlog 數(shù)據(jù)中的 prewrite_key 去查詢 TiKV 該事務(wù)是否提交，如果已經(jīng)提交成功，則 TiKV 會(huì)返回該事務(wù)的 commit_ts；否則 Pump 就丟棄該條 Prewrite binlog。

binlog 元數(shù)據(jù)中提供了數(shù)據(jù)存儲(chǔ)的文件和位置，可以通過這些信息讀取 binlog 文件的指定位置獲取到數(shù)據(jù)。因?yàn)?binlog 數(shù)據(jù)基本上是按順序?qū)懭氲轿募械模虼宋覀冎恍枰樞虻刈x binlog 文件即可，這樣就保證了不會(huì)因?yàn)轭l繁地讀取文件而影響 Pump 的性能。最終，Pump 以 commit_ts 為排序標(biāo)準(zhǔn)將 binlog 數(shù)據(jù)傳輸給 Drainer。Drainer 向 Pump 請(qǐng)求 binlog 數(shù)據(jù)的接口為 pump.proto 中的 PullBinlogs，以 grpc streaming 的形式傳輸 binlog 數(shù)據(jù)。

值得一提的是，Pump 中有一個(gè) fake binlog 機(jī)制。Pump 會(huì)定時(shí)（默認(rèn)三秒）向本地存儲(chǔ)中寫入一條數(shù)據(jù)為空的 binlog，在生成該 binlog 前，會(huì)向 PD 中獲取一個(gè) ts，作為該 binlog 的 start_ts 與 commit_ts，這種 binlog 我們叫作 fake binlog。這樣做的原因在 Drainer 中介紹。

Drainer 從各個(gè) Pump 中獲取 binlog，歸并后按照順序解析 binlog、生成 SQL 或者指定格式的數(shù)據(jù)，然后再同步到下游。

既然要從各個(gè) Pump 獲取數(shù)據(jù)，Drainer 就需要維護(hù) Pump 集群的信息，及時(shí)獲取到新增加的 Pump，并識(shí)別出不可用的 Pump，這部分功能與 Pump Client 類似，Drainer 也是通過 PD 中存儲(chǔ)的 Pump 的狀態(tài)信息來維護(hù) Pump 信息。另外需要注意的是，如果新增加了一個(gè) Pump，必須讓該 Pump 通知 Drainer 自己上線了，這么做是為了保證不會(huì)丟數(shù)據(jù)。例如：

集群中已經(jīng)存在 Pump1 和 Pump2，Drainer 讀取 Pump1 和 Pump2 的數(shù)據(jù)并進(jìn)行歸并：

Pump1 存儲(chǔ)的 binlog 為｛ 1，3，5，7，9 ｝，Pump2 存儲(chǔ)的 binlog 為｛2，4，6，10｝。Drainer 從兩個(gè) Pump 獲取 binlog，假設(shè)當(dāng)前已經(jīng)讀取到了｛1，2，3，4，5，6，7｝這些 binlog，已處理的 binlog 的位置為 7。此時(shí) Pump3 加入集群，從 Pump3 上報(bào)自己的上線信息到 PD，到 Drainer 從 PD 中獲取到 Pump3 信息需要一定的時(shí)間，如果 Pump3 沒有通知 Drainer 就直接提供寫 binlog 服務(wù)，寫入了 binlog｛8，12｝，Drainer 在此期間繼續(xù)讀取 Pump1 和 Pump2 的 binlog，假設(shè)讀取到了 9，之后才識(shí)別到了 Pump3 并將 Pump3 加入到歸并排序中，此時(shí) Pump3 的 binlog 8 就丟失了。為了避免這種情況，需要讓 Pump3 通知 Drainer 自己已經(jīng)上線，Drainer 收到通知后將 Pump3 加入到歸并排序，并返回成功給 Pump3，然后 Pump3 才能提供寫 binlog 的服務(wù)。

Drainer 通過如上所示的方式對(duì) binlog 進(jìn)行歸并排序，并推進(jìn)同步的位置。那么可能會(huì)存在這種情況：某個(gè) Pump 由于一些特殊的原因一直沒有收到 binlog 數(shù)據(jù)，那么 Drainer 中的歸并排序就無法繼續(xù)下去，正如我們用兩條腿走路，其中一只腿不動(dòng)就不能繼續(xù)前進(jìn)。我們使用 Pump 一節(jié)中提到的 fake binlog 的機(jī)制來避免這種問題，Pump 每隔指定的時(shí)間就生成一條 fake binlog，即使某些 Pump 一直沒有數(shù)據(jù)寫入，也可以保證歸并排序正常向前推進(jìn)。

Drainer 將所有 Pump 的數(shù)據(jù)按照 commit_ts 進(jìn)行歸并排序后，將 binlog 數(shù)據(jù)傳遞給 Drainer 中的數(shù)據(jù)解析及同步模塊。通過上面的 binlog 格式的介紹，我們可以看出 binlog 文件中并沒有存儲(chǔ)表結(jié)構(gòu)的信息，因此需要在 Drainer 中維護(hù)所有庫和表的結(jié)構(gòu)信息。在啟動(dòng) Drainer 時(shí)，Drainer 會(huì)請(qǐng)求 TiKV，獲取到所有歷史的 DDL job 的信息，對(duì)這些 DDL job 進(jìn)行過濾，使用 Drainer 啟動(dòng)時(shí)指定的 initial-commit-ts（或者 checkpoint 中保存的 commit_ts）之前的 DDL 在內(nèi)存中構(gòu)建庫和表結(jié)構(gòu)信息。這樣 Drainer 就有了一份 ts 對(duì)應(yīng)時(shí)間點(diǎn)的庫和表的快照，在讀取到 DDL 類型的 binlog 時(shí)，則更新庫和表的信息；讀取到 DML 類型的 binlog 時(shí)，則根據(jù)庫和表的信息來生成 SQL。

在生成 SQL 之后，就可以同步到下游了。為了提高 Drainer 同步的速度，Drainer 中使用多個(gè)協(xié)程來執(zhí)行 SQL。在生成 SQL 時(shí)，我們會(huì)使用主鍵／唯一鍵的值作為該條 SQL 的 key，通過對(duì) key 進(jìn)行 hash 來將 SQL 發(fā)送到對(duì)應(yīng)的協(xié)程中。當(dāng)每個(gè)協(xié)程收集到了足夠多的 SQL，或者超過了一定的時(shí)間，則將這一批的 SQL 在一個(gè)事務(wù)中提交到下游。

但是有些 SQL 是相關(guān)的，如果被分到了不同的協(xié)程，那 SQL 的執(zhí)行順序就不能得到保證，造成數(shù)據(jù)的不一致。例如：

SQL1: delete from test.test where id = 1;

SQL2: replace into test.test (id, name ) values(1, "a");

按照順序執(zhí)行后表中存在 id ＝ 1 該行數(shù)據(jù)，如果這兩條 SQL 分別分配到了協(xié)程 1 和協(xié)程 2 中，并且協(xié)程 2 先執(zhí)行了 SQL，則表中不再存在 id ＝ 1 的數(shù)據(jù)。為了避免這種情況的發(fā)生，Drainer 中加入了沖突檢測(cè)的機(jī)制，如果檢測(cè)出來兩條 SQL 存在沖突（修改了同一行數(shù)據(jù)），則暫時(shí)不將后面的 SQL 發(fā)送到協(xié)程，而是生成一個(gè) Flush 類型的 job 發(fā)送到所有的協(xié)程， 每個(gè)協(xié)程在遇到 Flush job 時(shí)就會(huì)馬上執(zhí)行所緩存的 SQL。接著才會(huì)把該條有沖突的 SQL 發(fā)送到對(duì)應(yīng)的協(xié)程中。下面給出一個(gè)例子說明一下沖突檢測(cè)的機(jī)制：

有以下這些 SQL，其中 id 為表的主鍵：

SQL1: update itest set id = 4, name = "c", age = 15 where id = 3;    key: 3, 4

SQL2: ?update itest set id = 5, name = "b", age = 14 where id = 2;   key：5, 2

SQL3：delete from itest where id = 3;                                key: 3

首先將 SQL1 發(fā)送到指定的協(xié)程，這時(shí)所有的 keys 為［3，4］；

SQL2 的 key［5，2］與 keys 中的［3，4］都沒有沖突，將 SQL2 發(fā)送到指定的協(xié)程，這時(shí) keys 為［3，4，5，2］；

SQL3 的 key［3］與 keys 中的［3］存在沖突，發(fā)送 Flush job 到所有協(xié)程，SQL1 和 SQL2 被執(zhí)行，清空 keys；

將 SQL3 發(fā)送到指定的協(xié)程，同時(shí)更新 keys 為［3］。

Drainer 通過以上這些機(jī)制來高效地同步數(shù)據(jù)，并且保證數(shù)據(jù)的一致。