标签：canal binlog 同步 name MySQL instance yjs mysql

MySQL自动同步开源工具

在现代的数据处理中，数据同步是非常重要的一个环节。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 模式下的性能优化问题。

4. mysql小知识：

MySQL 的 binlog（Binary Log）记录规则并不是按照库（数据库）为单位的，而是按照事务（或者语句）为单位的。也就是说，MySQL 的 binlog 会记录所有对数据库进行更改的事务或语句，而不管它们涉及到多少个数据库或表。
MySQL Binlog 记录的基本规则

事务级别记录：

在使用 ROW 格式的 binlog 时，MySQL 会记录每一个具体的行级别操作（如 INSERT、UPDATE、DELETE）。这些记录是基于事务的，即一个事务中的所有操作都会被记录在 binlog 中，直到事务提交时才会写入到 binlog 文件。

语句级别记录：

在使用 STATEMENT 格式的 binlog 时，MySQL 会记录每一个改变数据的 SQL 语句。语句级别的 binlog 记录方式也不区分具体的库，它会记录所有库中的数据更改语句。

混合模式（MIXED）：

在混合模式下，MySQL 会根据具体的 SQL 语句自动选择使用 ROW 或 STATEMENT 模式记录 binlog。
多库操作的处理

跨库事务：

如果一个事务涉及到多个数据库（库A和库B），那么所有这些数据库的操作都会被记录在同一个 binlog 事务记录中。这意味着 binlog 不会按库来拆分或记录日志。

跨库查询：

类似地，如果你执行的查询语句跨越多个数据库（如 SELECT * FROM db1.table1, db2.table2 WHERE ...），并且该查询导致数据更改，那么这条语句也会完整地记录在 binlog 中，而不会分开记录。

Binlog 文件的切分

按文件大小：
binlog 文件通常按照大小进行切分。当一个 binlog 文件达到 max_binlog_size 参数指定的大小时，MySQL 会生成一个新的 binlog 文件。这与是否涉及多个库无关。
手动切分：
你也可以通过 FLUSH LOGS 命令手动切分 binlog 文件，这会强制 MySQL 切换到一个新的 binlog 文件。
如何筛选特定库的 binlog 记录
因为 binlog 并不是按库分离记录的，如果你只想查看某个特定库的操作记录，需要在分析 binlog 时手动筛选。可以使用以下工具和方法：

mysqlbinlog 工具：

可以通过 --database 参数来过滤特定数据库的 binlog 记录。
SQL 解析：
通过 SHOW BINLOG EVENTS 命令可以查看 binlog 中的具体事件，然后手动筛选与某个特定库相关的事件。

日志分析工具：

使用 pt-query-digest 等工具分析 binlog 数据，并过滤出指定数据库的操作。

总结：

MySQL 的 binlog 记录是基于事务或语句的，并不会按库来拆分或组织记录。所有涉及到的库的操作会一起记录在相同的 binlog 文件中。要查看特定库的 binlog 记录，必须在 binlog 文件中筛选出相关的记录。

dev01机器测试同步方案

测试环境进行：dev01 + dev数据库、sit数据库(test1.yjs_schema)
docker：centos+openjdk-1.8.0_412
https://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
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化：

镜像：

peerup-registry-vpc.cn-beijing.cr.aliyuncs.com/zd_base/canal-yjsschema:v1.0.1

需要指定的变量：

点击查看代码

mysql_source=peerup-dev.rwlb.rds.aliyuncs.com
mysql_source=peerup-dev.rwlb.rds.aliyuncs.com
mysql_target=peerup-sit.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,binlog,同步,name,MySQL,instance,yjs,mysql
From： https://www.cnblogs.com/yuandlb/p/18358971

基于 canal+mysql 实现 yjs-schema 数据实时同步

MySQL自动同步开源工具

1. MaxScale

2. SymmetricDS

3. Canal

canal测试方案：

开启 binlog + gtid 模式：

1. 同步方式的基础

2. 数据恢复与同步的可靠性

3. 关于性能损耗：

4. mysql小知识：

事务级别记录：

语句级别记录：

混合模式（MIXED）：

跨库事务：

跨库查询：

Binlog 文件的切分

mysqlbinlog 工具：

日志分析工具：

总结：

dev01机器测试同步方案

canal-deployer --- 源库

canal-adapter --- 目标库

服务管理：

docker化：

镜像：

需要指定的变量：

健康监测：

全量同步：

相关文章

赞助商

阅读排行