一、go-mysql-transfer

go-mysql-transfer是使用Go语言实现的MySQL数据库实时增量同步工具。能够实时监听MySQL二进制日志(binlog)的变动，将变更内容形成指定格式的消息，发送到接收端。在数据库和接收端之间形成一个高性能、低延迟的增量数据(Binlog)同步管道, 具有如下特点：

go-mysql-transfer具有如下特点：

1、不依赖其它组件，一键部署

2、集成多种接收端，如：Redis、MongoDB、Elasticsearch、RabbitMQ、Kafka、RocketMQ，不需要再编写客户端，开箱即用

3、内置丰富的数据解析、消息生成规则；支持Lua脚本，以处理更复杂的数据逻辑

4、支持监控告警，集成Prometheus客户端

5、高可用集群部署

6、数据同步失败重试

7、全量数据初始化

详情及安装说明请参见： MySQL Binlog 增量同步工具go-mysql-transfer实现详解

项目开源地址：
gitee (速度更快) :go-mysql-transfer
github：go-mysql-transfer

如果此工具对你有帮助，请Star支持下

二、配置

# app.yml

target: rocketmq #目标类型
#rocketmq连接配置
rocketmq_name_servers: 127.0.0.1:9876 #rocketmq命名服务地址，多个用逗号分隔
#rocketmq_group_name: transfer_test_group #rocketmq group name,默认为空
#rocketmq_instance_name: transfer_test_group_ins #rocketmq instance name,默认为空
#rocketmq_access_key: RocketMQ #访问控制 accessKey,默认为空
#rocketmq_secret_key: 12345678 #访问控制 secretKey,默认为空

三、数据转换规则

相关配置如下：

rule:
  -
    schema: eseap #数据库名称
    table: t_user #表名称
    #order_by_column: id #排序字段，存量数据同步时不能为空
    #column_lower_case:false #列名称转为小写,默认为false
    #column_upper_case:false#列名称转为大写,默认为false
    column_underscore_to_camel: true #列名称下划线转驼峰,默认为false
    # 包含的列，多值逗号分隔，如：id,name,age,area_id  为空时表示包含全部列
    #include_columns: ID,USER_NAME,PASSWORD
    #exclude_columns: BIRTHDAY,MOBIE # 排除掉的列，多值逗号分隔，如：id,name,age,area_id  默认为空
    #column_mappings: CARD_NO=sfz    #列名称映射，多个映射关系用逗号分隔，如：USER_NAME=account 表示将字段名USER_NAME映射为account
    #default_column_values: source=binlog,area_name=合肥  #默认的列-值，多个用逗号分隔，如：source=binlog,area_name=合肥
    #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；默认为json
    #value_formatter: ${ID}|${USER_NAME} #值格式化表达式，如：${ID}|${USER_NAME},${ID}表示字段id的值、${USER_NAME}表示字段name的值
    
    #rocketmq相关
    rocketmq_topic: transfer_test_topic #rocketmq topic，可以为空，默认使用表名称

示例一

RocketMQ中创建名称为transfer_test_topic的topic，topic名称一定要和rule中rocketmq_topic配置的一致

t_user表，数据如下：

同步到RocketMQ的数据如下:

示例二

t_user表、topic 同实例一

使用如下配置：

rule:
  -
    schema: eseap #数据库名称
    table: t_user #表名称
    #order_by_column: id #排序字段，存量数据同步时不能为空
    column_lower_case: true #列名称转为小写,默认为false
    #column_upper_case:false#列名称转为大写,默认为false
    #column_underscore_to_camel: true #列名称下划线转驼峰,默认为false
    # 包含的列，多值逗号分隔，如：id,name,age,area_id  为空时表示包含全部列
    #include_columns: ID,USER_NAME,PASSWORD
    #exclude_columns: BIRTHDAY,MOBIE # 排除掉的列，多值逗号分隔，如：id,name,age,area_id  默认为空
    column_mappings: USER_NAME=account    #列名称映射，多个映射关系用逗号分隔，如：USER_NAME=account 表示将字段名USER_NAME映射为account
    default_column_values: area_name=合肥  #默认的列-值，多个用逗号分隔，如：source=binlog,area_name=合肥
    #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；默认为json
    #value_formatter: ${ID}|${USER_NAME} #值格式化表达式，如：${ID}|${USER_NAME},${ID}表示字段id的值、${USER_NAME}表示字段name的值

    #rocketmq相关
    rocketmq_topic: transfer_test_topic #rocketmq topic，可以为空，默认使用表名称

column_mappings配置项表示对列名称进行重新映射

同步到RocketMQ的数据如下:

示例三

t_user表、topic 同实例一

使用如下配置：

rule:
  -
    schema: eseap #数据库名称
    table: t_user #表名称
    #order_by_column: id #排序字段，存量数据同步时不能为空
    column_lower_case: true #列名称转为小写,默认为false
    #column_upper_case:false#列名称转为大写,默认为false
    #column_underscore_to_camel: true #列名称下划线转驼峰,默认为false
    # 包含的列，多值逗号分隔，如：id,name,age,area_id  为空时表示包含全部列
    #include_columns: ID,USER_NAME,PASSWORD
    #exclude_columns: BIRTHDAY,MOBIE # 排除掉的列，多值逗号分隔，如：id,name,age,area_id  默认为空
    column_mappings: USER_NAME=account    #列名称映射，多个映射关系用逗号分隔，如：USER_NAME=account 表示将字段名USER_NAME映射为account
    default_column_values: area_name=合肥  #默认的列-值，多个用逗号分隔，如：source=binlog,area_name=合肥
    #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: v-commas  #值编码，支持json、kv-commas、v-commas；默认为json
    #value_formatter: ${ID}|${USER_NAME} #值格式化表达式，如：${ID}|${USER_NAME},${ID}表示字段id的值、${USER_NAME}表示字段name的值

    #rocketmq相关
    rocketmq_topic: transfer_test_topic #rocketmq topic，可以为空，默认使用表名称

column_mappings配置项表示对列名称进行重新映射

value_encoder配置项表示消息编码方式

同步到RocketMQ的数据如下:

四、Lua脚本

使用Lua脚本可以实现更复杂的数据处理逻辑，go-mysql-transfer支持Lua5.1语法。

示例一

RocketMQ中创建名称为transfer_test_topic的topic

t_user表，数据如下：

引入Lua脚本：

rule:
  -
    schema: eseap #数据库名称
    table: t_user #表名称
    lua_file_path: lua/t_user_rocket.lua   #lua脚本文件

Lua脚本：

local json = require("json")   -- 加载json模块
local ops = require("mqOps") --加载mq操作模块

local row = ops.rawRow()  --当前数据库的一行数据,table类型，key为列名称
local action = ops.rawAction()  --当前数据库事件,包括：insert、updare、delete

local id = row["ID"] --获取ID列的值
local userName = row["USER_NAME"] --获取USER_NAME列的值
local password = row["PASSWORD"] --获取USER_NAME列的值
local createTime = row["CREATE_TIME"] --获取CREATE_TIME列的值

local result = {}  -- 定义一个table,作为结果
result["id"] = id
result["action"] = action

if action == "delete" -- 删除事件
then
    local val = json.encode(result) -- 将result转为json
    ops.SEND("transfer_test_topic",val) -- 发送消息，第一个参数为topic(string类型)，第二个参数为消息内容
else 
    result["userName"] = userName
    result["password"] = password
    result["createTime"] = createTime
    result["source"] = "binlog" -- 数据来源
    local val = json.encode(result) -- 将result转为json
    ops.SEND("transfer_test_topic",val) -- 发送消息，第一个参数为topic(string类型)，第二个参数为消息内容
end

同步到RocketMQ的数据如下:

示例二

t_user表、topic 同实例一

使用如下脚本：

local ops = require("mqOps") --加载mq操作模块

local row = ops.rawRow()  --当前数据库的一行数据,table类型，key为列名称
local action = ops.rawAction()  --当前数据库事件,包括：insert、updare、delete

local userName = row["USER_NAME"] --获取USER_NAME列的值

if action == "insert" then -- 只监听添加事件
    local str = string.format("恭喜您：%s  注册成功",userName)  
    ops.SEND("transfer_test_topic",str) -- 发送消息，第一个参数为topic(string类型)，第二个参数为消息内容
end

同步到RocketMQ的数据如下:

mqOps模块提供的方法如下：

SEND: 发送操作,如：ops.SEND(topic,result)。参数topic为字符串类型；参数result为要发送的消息

MySQL数据实时增量同步到RocketMQ

一、go-mysql-transfer

二、配置

三、数据转换规则

四、Lua脚本