后起之秀 | MySQL Binlog增量同步工具go-mysql-transfer實現(xiàn)詳解

一、 概述

工作需要研究了下阿里開源的MySQL Binlog增量訂閱消費組件canal，其功能強大、運行穩(wěn)定，但是有些方面不是太符合需求，主要有如下三點：

1. 需要自己編寫客戶端來消費canal解析到的數(shù)據(jù)

2. server-client模式，需要同時部署server和client兩個組件，我們的項目中有6個業(yè)務數(shù)據(jù)庫要實時同步到redis，意味著要多部署12個組件，硬件和運維成本都會增加。

3. 從server端到client端需要經(jīng)過一次網(wǎng)絡傳輸和序列化反序列化操作，然后再同步到接收端，感覺沒有直接懟到接收端更高效。

go-mysql-transfer是使用Go語言實現(xiàn)的MySQL數(shù)據(jù)庫實時增量同步工具， 參考Canal但是規(guī)避了上述三點。旨在實現(xiàn)一個高性能、低延遲、簡潔易用的Binlog增量數(shù)據(jù)同步管道, 具有如下特點：

1. 不依賴其它組件，一鍵部署

2. 集成多種接收端，如：Redis、MongoDB、Elasticsearch、RocketMQ、Kafka、RabbitMQ，不需要再編寫客戶端，開箱即用

3. 內(nèi)置豐富的數(shù)據(jù)解析、消息生成規(guī)則；支持Lua腳本，以處理更復雜的數(shù)據(jù)邏輯

4. 支持監(jiān)控告警，集成Prometheus客戶端

5. 高可用集群部署

6. 數(shù)據(jù)同步失敗重試

7. 全量數(shù)據(jù)初始化

二、 與同類工具比較

三、 設計實現(xiàn)

1、實現(xiàn)原理

go-mysql-transfer將自己偽裝成MySQL的Slave，向Master發(fā)送dump協(xié)議獲取binlog，解析binlog并生成消息，實時發(fā)送給接收端。

2、數(shù)據(jù)轉(zhuǎn)換規(guī)則

將從binlog解析出來的數(shù)據(jù)，經(jīng)過簡單的處理轉(zhuǎn)換發(fā)送到接收端。使用內(nèi)置豐富數(shù)數(shù)據(jù)轉(zhuǎn)換規(guī)則，可完成大部分同步工作。

例如將表t_user同步到reids，配置如下規(guī)則：

rule:
  -
    schema: eseap #數(shù)據(jù)庫名稱
    table: t_user #表名稱
    column_underscore_to_camel: true #列名稱下劃線轉(zhuǎn)駝峰,默認為false
    datetime_formatter: yyyy-MM-dd HH:mm:ss #datetime、timestamp類型格式化，不填寫默認yyyy-MM-dd HH:mm:ss
    value_encoder: json  #值編碼類型，支持json、kv-commas、v-commas
    redis_structure: string # redis數(shù)據(jù)類型。支持string、hash、list、set類型(與redis的數(shù)據(jù)類型一致)
    redis_key_prefix: USER_ #key前綴
    redis_key_column: USER_NAME #使用哪個列的值作為key，不填寫默認使用主鍵

t_user表，數(shù)據(jù)如下：

同步到Redis后，數(shù)據(jù)如下：

更多規(guī)則配置和同步案例 請見后續(xù)的"使用說明"章節(jié)。

3、數(shù)據(jù)轉(zhuǎn)換腳本

Lua 是一種輕量小巧的腳本語言， 其設計目的是為了嵌入應用程序中，從而為應用程序提供靈活的擴展和定制功能。開發(fā)者只需要花費少量時間就能大致掌握Lua的語法，照虎畫貓寫出可用的腳本。

基于Lua的高擴展性，可以實現(xiàn)更為復雜的數(shù)據(jù)解析、消息生成邏輯，定制需要的數(shù)據(jù)格式。

使用方式：

rule:
  -
    schema: eseap
    table: t_user
    lua_file_path: lua/t_user_string.lua   #lua腳本文件

示例腳本：

local json = require("json")    -- 加載json模塊
local ops = require("redisOps") -- 加載redis操作模塊

local row = ops.rawRow()  --當前變動的一行數(shù)據(jù),table類型，key為列名稱
local action = ops.rawAction()  --當前數(shù)據(jù)庫的操作事件,包括：insert、updare、delete

local id = row["ID"] --獲取ID列的值
local userName = row["USER_NAME"] --獲取USER_NAME列的值
local key = "user_"..id -- 定義key

if action == "delete" -- 刪除事件
then
    ops.DEL(key)  -- 刪除KEY
else 
    local password = row["PASSWORD"] --獲取USER_NAME列的值
    local createTime = row["CREATE_TIME"] --獲取CREATE_TIME列的值
    local result= {}  -- 定義結(jié)果
    result["id"] = id
    result["userName"] = userName
    result["password"] = password
    result["createTime"] = createTime
    result["source"] = "binlog" -- 數(shù)據(jù)來源
    local val = json.encode(result) -- 將result轉(zhuǎn)為json
    ops.SET(key,val)  -- 對應Redis的SET命令，第一個參數(shù)為key(string類型)，第二個參數(shù)為value
end

t_user表，數(shù)據(jù)如下：

同步到Redis后，數(shù)據(jù)如下：

更多Lua腳本使用說明 和同步案例 請見后續(xù)的"使用說明"章節(jié)。

4、監(jiān)控告警

Prometheus是流行開源監(jiān)控報警系統(tǒng)和TSDB，其指標采集組件被稱作exporter。go-mysql-transfer本身就是一個exporter。向Prometheus提供應用狀態(tài)、接收端狀態(tài)、insert數(shù)量、update數(shù)量、delete數(shù)量、delay延時等指標。

go-mysql-transfer內(nèi)置Prometheus exporter可以監(jiān)控系統(tǒng)的運行狀況，并進行健康告警。

相關配置：

enable_exporter: true #啟用prometheus exporter，默認false
exporter_addr: 9595 #prometheus exporter端口，默認9595

直接訪問127.0.0.1:9595可以看到導出的指標值，如何與Prometheus集成，請參見Prometheus相關教程。

指標說明：transfer_leader_state：當前節(jié)點是否為leader，0=否、1=是 transfer_destination_state：接收端狀態(tài)， 0=掉線、1=正常 transfer_inserted_num：插入數(shù)據(jù)的數(shù)量 transfer_updated_num：修改數(shù)據(jù)的數(shù)量 transfer_deleted_num：刪除數(shù)據(jù)的數(shù)量 transfer_delay：與MySQL Master的時延

5、高可用

可以選擇依賴zookeeper或者etcdr構建高可用集群，一個集群中只存在一個leader節(jié)點，其余皆為follower節(jié)點。只有l(wèi)eader節(jié)點響應binglog的dump事件，follower節(jié)點為蟄伏狀態(tài)，不發(fā)送dump命令，因此多個follower也不會加重Master的負擔。當leader節(jié)點出現(xiàn)故障，follower節(jié)點迅速替補上去，實現(xiàn)秒級故障切換。

相關配置：

cluster: # 集群配置
  name: myTransfer #集群名稱，具有相同name的節(jié)點放入同一個集群
  # ZooKeeper地址，多個用逗號分隔
  zk_addrs: 192.168.1.10:2181,192.168.1.11:2182,192.168.1.12:2183
  #zk_authentication: 123456 #digest類型的訪問秘鑰，如：user:password，默認為空
  #etcd_addrs: 192.168.1.10:2379 #etcd連接地址，多個用逗號分隔
  #etcd_user: test #etcd用戶名
  #etcd_password: 123456 #etcd密碼

6、失敗重試

網(wǎng)絡抖動、接收方故障都會導致數(shù)據(jù)同步失敗，需要有重試機制，才能保證不漏掉數(shù)據(jù)，使得每一條數(shù)據(jù)都能送達。

通常有兩種重試實現(xiàn)方式，一種方式是記錄下故障時刻binglog的position(位移)，等故障恢復后，從position處重新dump 數(shù)據(jù)，發(fā)送給接收端。

一種方式是將同步失敗的數(shù)據(jù)在本地落盤，形成隊列。當探測到接收端可用時，逐條預出列嘗試發(fā)送，發(fā)送成功最終出列。確保不丟數(shù)據(jù)，隊列先進先出的特性也可保證數(shù)據(jù)順序性，正確性。

go-mysql-transfer采用的是后者，目的是減少發(fā)送dump命令的次數(shù)，減輕Master的負擔。因為binglog記錄的整個Master數(shù)據(jù)庫的日志，其增長速度很快。如果只需要拿幾條數(shù)據(jù)，而dump很多數(shù)據(jù)，有點得不償失。

7、全量數(shù)據(jù)初始化

如果數(shù)據(jù)庫原本存在無法通過binlog進行增量同步的數(shù)據(jù)，可以使用命令行工具-stock完成始化同步。stock基于 SELECT * FROM {table}的方式分批查詢出數(shù)據(jù)，根據(jù)規(guī)則或者Lua腳本生成指定格式的消息，批量發(fā)送到接收端。執(zhí)行命令 go-mysql-transfer -stoc，在控制臺可以直觀的看到數(shù)據(jù)同步狀態(tài)，如下：

四、安裝

二進制安裝包

直接下載編譯好的安裝包: https://github.com/wj596/go-mysql-transfer/releases

源碼編譯

1、依賴Golang 1.14 及以上版本 2、設置GO111MODULE=on 3、拉取源碼 go get -d github.com/wj596/go-mysql-transfer 3、進入目錄，執(zhí)行 go build 編譯

五、部署運行

開啟MySQL的binlog

#Linux在my.cnf文件
#Windows在my.ini文件
log-bin=mysql-bin # 開啟 binlog
binlog-format=ROW # 選擇 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定義，不要和 go-mysql-transfer 的 slave_id 重復

命令行運行 1、修改app.yml 2、Windows直接運行 go-mysql-transfer.exe 3、Linux執(zhí)行 nohup go-mysql-transfer &

docker運行

1、拉取源碼 go get -d github.com/wj596/go-mysql-transfer 2、修改配置文件 app.yml 中相關配置 3、構建鏡像 docker image build -t go-mysql-transfer -f Dockerfile . 4、運行 docker run -d --name go-mysql-transfer -p 9595:9595 go-mysql-transfer:latest

六、性能測試

1、測試環(huán)境

平臺：虛擬機 CPU：E7-4890 4核8線程 內(nèi)存：8G 硬盤：機械硬盤 OS:Windows Sever 2012 R2 MySQL: 5.5 Rides: 4.0.2

2、測試數(shù)據(jù)

t_user表，14個字段，1個字段包含中文，數(shù)據(jù)量527206條

3、測試配置

規(guī)則：

    schema: eseap
    table: t_user
    order_by_column: id #排序字段，全量數(shù)據(jù)初始化時不能為空
    #column_lower_case:false #列名稱轉(zhuǎn)為小寫,默認為false
    #column_upper_case:false#列名稱轉(zhuǎn)為大寫,默認為false
    column_underscore_to_camel: true #列名稱下劃線轉(zhuǎn)駝峰,默認為false
    # 包含的列，多值逗號分隔，如：id,name,age,area_id  為空時表示包含全部列
    #include_column: ID,USER_NAME,PASSWORD
    date_formatter: yyyy-MM-dd #date類型格式化， 不填寫默認yyyy-MM-dd
    datetime_formatter: yyyy-MM-dd HH:mm:ss #datetime、timestamp類型格式化，不填寫默認yyyy-MM-dd HH:mm:ss
    value_encoder: json  #值編碼，支持json、kv-commas、v-commas
    redis_structure: string # 數(shù)據(jù)類型。支持string、hash、list、set類型(與redis的數(shù)據(jù)類型一直)
    redis_key_prefix: USER_ #key的前綴
    redis_key_column: ID #使用哪個列的值作為key，不填寫默認使用主鍵

腳本：

local json = require("json")    -- 加載json模塊
local ops = require("redisOps") -- 加載redis操作模塊

local row = ops.rawRow()  --當前變動的一行數(shù)據(jù),table類型，key為列名稱
local action = ops.rawAction()  --當前數(shù)據(jù)庫的操作事件,包括：insert、updare、delete

local id = row["ID"] --獲取ID列的值
local userName = row["USER_NAME"] --獲取USER_NAME列的值
local key = "user_"..id -- 定義key

if action == "delete" -- 刪除事件
then
    ops.DEL(key)  -- 刪除KEY
else 
    local password = row["PASSWORD"] --獲取USER_NAME列的值
    local createTime = row["CREATE_TIME"] --獲取CREATE_TIME列的值
    local result= {}  -- 定義結(jié)果
    result["id"] = id
    result["userName"] = userName
    result["password"] = password
    result["createTime"] = createTime
    result["source"] = "binlog" -- 數(shù)據(jù)來源
    local val = json.encode(result) -- 將result轉(zhuǎn)為json
    ops.SET(key,val)  -- 對應Redis的SET命令，第一個參數(shù)為key(string類型)，第二個參數(shù)為value
end

3、測試用例一

使用規(guī)則，將52萬條數(shù)據(jù)全量初始化同步到Redis，結(jié)果如下：

3次運行的中間值為4.6秒

4、測試用例二

使用Lua腳本，將52萬條數(shù)據(jù)全量初始化同步到Redis，結(jié)果如下：

3次運行的中間值為9.5秒

5、測試用例三

使用規(guī)則，將binlog中52萬條增量數(shù)據(jù)同步到Redis。結(jié)果如下：每秒增量同步(TPS)32950條

6、測試用例四

使用Lua腳本，將binlog中52萬條增量數(shù)據(jù)同步到Redis。結(jié)果如下：

每秒增量同步(TPS)15819條

7、測試用例五

100個線程不停向MySQL寫數(shù)據(jù)，使用規(guī)則將數(shù)據(jù)實時增量同步到Redis，TPS保持在4000以上，資源占用情況如下：

100個線程不停向MySQL寫數(shù)據(jù)，使用Lua腳本將數(shù)據(jù)實時增量同步到Redis，TPS保持在2000以上，資源占用情況如下：

以上測試結(jié)果，會隨著測試環(huán)境的不同而改變，僅作為參考。