后起之秀 | MySQL Binlog增量同步工具go-mysql-transfer实现详解

本文主要是介绍后起之秀 | MySQL Binlog增量同步工具go-mysql-transfer实现详解，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

点击上方蓝色字体，选择“设为星标”

回复”资源“获取更多资源

一、概述

工作需要研究了下阿里开源的MySQL Binlog增量订阅消费组件canal，其功能强大、运行稳定，但是有些方面不是太符合需求，主要有如下三点：

需要自己编写客户端来消费canal解析到的数据
server-client模式，需要同时部署server和client两个组件，我们的项目中有6个业务数据库要实时同步到redis，意味着要多部署12个组件，硬件和运维成本都会增加。
从server端到client端需要经过一次网络传输和序列化反序列化操作，然后再同步到接收端，感觉没有直接怼到接收端更高效。

go-mysql-transfer是使用Go语言实现的MySQL数据库实时增量同步工具，参考Canal但是规避了上述三点。旨在实现一个高性能、低延迟、简洁易用的Binlog增量数据同步管道, 具有如下特点：

不依赖其它组件，一键部署
集成多种接收端，如：Redis、MongoDB、Elasticsearch、RocketMQ、Kafka、RabbitMQ，不需要再编写客户端，开箱即用
内置丰富的数据解析、消息生成规则；支持Lua脚本，以处理更复杂的数据逻辑
支持监控告警，集成Prometheus客户端
高可用集群部署
数据同步失败重试
全量数据初始化

二、与同类工具比较

三、设计实现

1、实现原理

go-mysql-transfer将自己伪装成MySQL的Slave，向Master发送dump协议获取binlog，解析binlog并生成消息，实时发送给接收端。

2、数据转换规则

将从binlog解析出来的数据，经过简单的处理转换发送到接收端。使用内置丰富数数据转换规则，可完成大部分同步工作。

例如将表t_user同步到reids，配置如下规则：

rule:-schema: eseap #数据库名称table: t_user #表名称column_underscore_to_camel: true #列名称下划线转驼峰,默认为falsedatetime_formatter: yyyy-MM-dd HH:mm:ss #datetime、timestamp类型格式化，不填写默认yyyy-MM-dd HH:mm:ssvalue_encoder: json  #值编码类型，支持json、kv-commas、v-commasredis_structure: string # redis数据类型。支持string、hash、list、set类型(与redis的数据类型一致)redis_key_prefix: USER_ #key前缀redis_key_column: USER_NAME #使用哪个列的值作为key，不填写默认使用主键

t_user表，数据如下：

同步到Redis后，数据如下：

更多规则配置和同步案例请见后续的"使用说明"章节。

3、数据转换脚本

Lua 是一种轻量小巧的脚本语言，其设计目的是为了嵌入应用程序中，从而为应用程序提供灵活的扩展和定制功能。开发者只需要花费少量时间就能大致掌握Lua的语法，照虎画猫写出可用的脚本。

基于Lua的高扩展性，可以实现更为复杂的数据解析、消息生成逻辑，定制需要的数据格式。

使用方式：

rule:-schema: eseaptable: t_userlua_file_path: lua/t_user_string.lua   #lua脚本文件

示例脚本：

local json = require("json")    -- 加载json模块
local ops = require("redisOps") -- 加载redis操作模块local row = ops.rawRow()  --当前变动的一行数据,table类型，key为列名称
local action = ops.rawAction()  --当前数据库的操作事件,包括：insert、updare、deletelocal id = row["ID"] --获取ID列的值
local userName = row["USER_NAME"] --获取USER_NAME列的值
local key = "user_"..id -- 定义keyif action == "delete" -- 删除事件
thenops.DEL(key)  -- 删除KEY
else local password = row["PASSWORD"] --获取USER_NAME列的值local createTime = row["CREATE_TIME"] --获取CREATE_TIME列的值local result= {}  -- 定义结果result["id"] = idresult["userName"] = userNameresult["password"] = passwordresult["createTime"] = createTimeresult["source"] = "binlog" -- 数据来源local val = json.encode(result) -- 将result转为jsonops.SET(key,val)  -- 对应Redis的SET命令，第一个参数为key(string类型)，第二个参数为value
end

t_user表，数据如下：

同步到Redis后，数据如下：

更多Lua脚本使用说明和同步案例请见后续的"使用说明"章节。

4、监控告警

Prometheus是流行开源监控报警系统和TSDB，其指标采集组件被称作exporter。go-mysql-transfer本身就是一个exporter。向Prometheus提供应用状态、接收端状态、insert数量、update数量、delete数量、delay延时等指标。

go-mysql-transfer内置Prometheus exporter可以监控系统的运行状况，并进行健康告警。

四、安装

二进制安装包

直接下载编译好的安装包: 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、进入目录，执行 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执行 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、测试环境

平台：虚拟机 CPU：E7-4890 4核8线程内存：8G 硬盘：机械硬盘 OS:Windows Sever 2012 R2 MySQL: 5.5 Rides: 4.0.2

2、测试数据

t_user表，14个字段，1个字段包含中文，数据量527206条

3、测试配置

规则：

    schema: eseaptable: t_userorder_by_column: id #排序字段，全量数据初始化时不能为空#column_lower_case:false #列名称转为小写,默认为false#column_upper_case:false#列名称转为大写,默认为falsecolumn_underscore_to_camel: true #列名称下划线转驼峰,默认为false# 包含的列，多值逗号分隔，如：id,name,age,area_id  为空时表示包含全部列#include_column: ID,USER_NAME,PASSWORDdate_formatter: yyyy-MM-dd #date类型格式化， 不填写默认yyyy-MM-dddatetime_formatter: yyyy-MM-dd HH:mm:ss #datetime、timestamp类型格式化，不填写默认yyyy-MM-dd HH:mm:ssvalue_encoder: json  #值编码，支持json、kv-commas、v-commasredis_structure: string # 数据类型。支持string、hash、list、set类型(与redis的数据类型一直)redis_key_prefix: USER_ #key的前缀redis_key_column: ID #使用哪个列的值作为key，不填写默认使用主键

脚本：

local json = require("json")    -- 加载json模块
local ops = require("redisOps") -- 加载redis操作模块local row = ops.rawRow()  --当前变动的一行数据,table类型，key为列名称
local action = ops.rawAction()  --当前数据库的操作事件,包括：insert、updare、deletelocal id = row["ID"] --获取ID列的值
local userName = row["USER_NAME"] --获取USER_NAME列的值
local key = "user_"..id -- 定义keyif action == "delete" -- 删除事件
thenops.DEL(key)  -- 删除KEY
else local password = row["PASSWORD"] --获取USER_NAME列的值local createTime = row["CREATE_TIME"] --获取CREATE_TIME列的值local result= {}  -- 定义结果result["id"] = idresult["userName"] = userNameresult["password"] = passwordresult["createTime"] = createTimeresult["source"] = "binlog" -- 数据来源local val = json.encode(result) -- 将result转为jsonops.SET(key,val)  -- 对应Redis的SET命令，第一个参数为key(string类型)，第二个参数为value
end

3、测试用例一

使用规则，将52万条数据全量初始化同步到Redis，结果如下：