作者：和元
在现代的数据处理中，数据同步是非常重要的一个环节。MySQL 作为一个广泛应用的数据库管理系统，自动同步数据也是一个比较常见的需求。今天我们将介绍一些开源工具，可以帮助我们实现 MySQL 数据库的自动同步。
数据库同步工具简介：
1. MaxScale
MaxScale是一个开源的MySQL Proxy工具，它提供了许多强大的功能，其中就包括数据同步。我们可以通过MaxScale实现MySQL数据库之间的自动同步。
MaxScale的优点是功能强大，支持复杂的数据同步场景，但配置复杂度较高，需要一定的学习成本。
2. SymmetricDS
SymmetricDS是一个开源的数据同步工具，支持多种数据库之间的数据同步。我们可以通过SymmetricDS实现MySQL数据库之间的自动同步。
SymmetricDS的优点是易于配置，适合快速搭建数据同步解决方案，但功能相对较弱，适用于简单的数据同步场景。
3. Canal
Canal是阿里巴巴开源的数据同步工具，可以实现MySQL数据库的数据实时同步。Canal采用基于日志的方式进行数据同步，保证了数据的准确性和实时性。
Canal的优点是数据同步实时性高，能够快速同步数据变更，但相对较复杂，需要一定的技术背景来使用。
综上，基于目前公司情况，选择使用canal来进行yjs-schema（此处指的是数据库名称，根据业务需求改变）数据实时同步。
canal 同步测试方案：
参考文档：https://www.cnblogs.com/cndarren/p/16318728.html
canal官方文档：https://github.com/alibaba/canal/wiki/简介
canal全量同步：https://help.aliyun.com/zh/tablestore/use-cases/use-canal-to-synchronize-data#section-xmo-fys-8w3
开启 binlog + gtid 模式：
1. 同步方式的基础
基于 binlog 的同步： 这种方式是 Canal 最基础的同步方式，通过解析 MySQL binlog 文件中的数据变更来进行同步。 同步位点（Position）： Canal 使用 binlog 文件名和位置（offset）来标记同步的位点。当 Canal 重新启动时，会从上次记录的位点继续读取 binlog。 基于 binlog + GTID 的同步： GTID（全局事务标识符）是一种 MySQL 提供的更高级的事务标识方式，它为每个事务分配一个唯一的标识符。 同步位点（GTID）： 在这种方式下，Canal 使用 GTID 来标记同步的进度，而不是依赖 binlog 文件名和位置。当 Canal 重新启动时，会从上次处理的 GTID 集继续同步数据。
2. 数据恢复与同步的可靠性
基于 binlog 的同步： 恢复能力：如果 Canal 异常停止，需要依赖上次记录的 binlog 位点（文件名+位置）来恢复同步。这种方式在某些情况下可能会导致数据丢失或重复同步。 可靠性：如果 MySQL 主库和从库之间存在延迟，或者在 binlog 清理时丢失了部分日志，可能会导致 Canal 无法准确恢复同步状态。 基于 binlog + GTID 的同步： 恢复能力：由于 GTID 是全局唯一的，Canal 可以更可靠地从上次停止的位置恢复同步。即使 MySQL 切换了 binlog 文件或发生了主从切换，Canal 仍然可以基于 GTID 准确恢复同步。 可靠性：这种方式下，Canal 的同步更加可靠，可以避免 binlog 文件滚动、主从切换等情况引起的同步中断或数据丢失。
3. 关于性能损耗：
Canal 开启 GTID 模式后，虽然会引入一些性能损耗，但一般情况下，这些损耗是非常小的，不会显著影响 Canal 的性能。GTID 模式带来的数据一致性和容错性通常远远超过了它带来的性能开销。因此，除非在极端高并发或超大规模的集群环境下，才需要特别关注 GTID 模式下的性能优化问题。
dev01机器测试同步方案
测试环境进行：dev01 + dev数据库、sit数据库(test1.yjs_schema) docker：centos+openjdk-1.8.0_412https://peerup-test.oss-cn-beijing.aliyuncs.com/dlb/canal.adapter-1.1.7.tar.gz
https://peerup-test.oss-cn-beijing.aliyuncs.com/dlb/canal.deployer-1.1.7.tar.gz
wget 到 centos 容器的 /opt 目录
canal-deployer --- 源库
canal-deployer 的 conf/example/instance.properties 文件的内容：
################################################# ## mysql serverId , v1.0.26+ will autoGen # canal.instance.mysql.slaveId=0 # enable gtid use true/false 开启gtid模式 canal.instance.gtidon=true # position info canal.instance.master.address=$mysql_source:3306 canal.instance.master.journal.name= canal.instance.master.position= canal.instance.master.timestamp= canal.instance.master.gtid= # rds oss binlog canal.instance.rds.accesskey= canal.instance.rds.secretkey= canal.instance.rds.instanceId= # table meta tsdb info canal.instance.tsdb.enable=true #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb #canal.instance.tsdb.dbUsername=canal #canal.instance.tsdb.dbPassword=canal #canal.instance.standby.address = #canal.instance.standby.journal.name = #canal.instance.standby.position = #canal.instance.standby.timestamp = #canal.instance.standby.gtid= # username/password canal.instance.dbUsername=$user canal.instance.dbPassword=$passwd canal.instance.connectionCharset = UTF-8 # enable druid Decrypt database password canal.instance.enableDruid=false #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ== # table regex canal.instance.filter.regex=$db_name.$table_name # table black regex canal.instance.filter.black.regex=mysql\\.slave_.* # table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2) #canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch # table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2) #canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch # mq config canal.mq.topic=example # dynamic topic route by schema or table regex #canal.mq.dynamicTopic=mytest1.user,topic2:mytest2\\..*,.*\\..* canal.mq.partition=0 # hash partition config #canal.mq.enableDynamicQueuePartition=false #canal.mq.partitionsNum=3 #canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6 #canal.mq.partitionHash=test.table:id^name,.*\\..* # # multi stream for polardbx canal.instance.multi.stream.on=false ################################################# 查看全部
canal-adapter --- 目标库
canal-adapter 的 conf/application.yml 文件的内容：
server: port: 8081 spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null canal.conf: mode: tcp #tcp kafka rocketMQ rabbitMQ flatMessage: true zookeeperHosts: syncBatchSize: 1000 retries: -1 timeout: accessKey: secretKey: consumerProperties: # canal tcp consumer canal.tcp.server.host: 127.0.0.1:11111 canal.tcp.zookeeper.hosts: canal.tcp.batch.size: 500 canal.tcp.username: canal.tcp.password: # kafka consumer kafka.bootstrap.servers: 127.0.0.1:9092 kafka.enable.auto.commit: false kafka.auto.commit.interval.ms: 1000 kafka.auto.offset.reset: latest kafka.request.timeout.ms: 40000 kafka.session.timeout.ms: 30000 kafka.isolation.level: read_committed kafka.max.poll.records: 1000 # rocketMQ consumer rocketmq.namespace: rocketmq.namesrv.addr: 127.0.0.1:9876 rocketmq.batch.size: 1000 rocketmq.enable.message.trace: false rocketmq.customized.trace.topic: rocketmq.access.channel: rocketmq.subscribe.filter: # rabbitMQ consumer rabbitmq.host: rabbitmq.virtual.host: rabbitmq.username: rabbitmq.password: rabbitmq.resource.ownerId: srcDataSources: defaultDS: url: jdbc:mysql://$mysql_source:3306/$db_name?useUnicode=true&characterEncoding=utf8&autoReconnect=true&useSSL=false username: $user password: $passwd canalAdapters: - instance: example # canal instance Name or mq topic name groups: - groupId: g1 outerAdapters: - name: logger - name: rdb key: mysql1 properties: jdbc.driverClassName: com.mysql.jdbc.Driver jdbc.url: jdbc:mysql://$mysql_target:3306/$db_name?useUnicode=true&characterEncoding=utf8&autoReconnect=true&useSSL=false jdbc.username: $user jdbc.password: $passwd threads: 5
修改 canal-adapter 的 conf/rdb/yjs_schema.yml 文件的以下内容：
dataSourceKey: defaultDS destination: example groupId: g1 outerAdapterKey: mysql1 concurrent: false dbMapping: mirrorDb: true database: $db_name table: $table_name targetTable: $table_name targetPk: $pkid: $pkid mapAll: true
服务管理：
#!/bin/bash usage(){ cat <<-"EOF" +------------------------------+ | canal 使用说明 | +------------------------------+ | $0 start 启动canal | | $0 stop 停止canal | | $0 restart 重启canal | | $0 status 查看canal | +------------------------------+ EOF } case $1 in start)sh /opt/deployer/bin/startup.sh && sh /opt/adapter/bin/startup.sh ;; stop) sh /opt/deployer/bin/stop.sh && sh /opt/adapter/bin/stop.sh ;; restart) sh /opt/deployer/bin/restart.sh && sh /opt/adapter/bin/restart.sh ;; status) ps aux|egrep 'deployer|adapter'|grep -v grep ;; *) usage ;; esac
docker化：
镜像：
docker commit ${test_docker_name} ${docker_images_name}:${tag}
需要指定的变量：
mysql_source=test1.rwlb.rds.aliyuncs.com mysql_target=test2.rwlb.rds.aliyuncs.com db_name=test1 table_name=yjs_schema pkid=schema_version #主键，可以配置多个
健康监测：
curl http://localhost:8081/destinations
返回结果：
[{"destination":"example","status":"on"}]
全量同步：
curl "localhost:8081/etl/rdb/mysql1/yjs_schema.yml" -X POST
注意事项：
- canal同步两个 mysql 数据库时，db_name 必须一致，及数据库名字必须一致
欢迎访问阿里云云数据库 RDS MySQL 详情页了解更多信息：https://www.aliyun.com/product/rds/mysql
