canal1.1.7使用canal-adapter进行mysql同步数据

🕗 发布于 2024-11-08 16:27 mysql 数据库 java

重要的事情说前面，canal1.1.8需要jdk11以上，大家自行选择，我这由于项目原因只能使用1.1.7兼容版的

文章参考地址：

canal 使用详解_canal使用-CSDN博客

使用canal.deployer-1.1.7和canal.adapter-1.1.7实现mysql数据同步_mysql更新数据时canal.adapter1.1.7不能同步-CSDN博客
 canal使用-CSDN博客

下载

下载地址：Releases · alibaba/canal · GitHub

下载使用这三个就可以。如果不用管理平台的话可以不下载canal-admin

canal简介

使用官方描述粘贴一下

canal [kə'næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费

早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务 trigger 获取增量变更。从 2010 年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。

基于日志增量订阅和消费的业务包括

数据库镜像
数据库实时备份
索引构建和实时维护(拆分异构索引、倒排索引等)
业务 cache 刷新
带业务逻辑的增量数据处理

当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x

MySQL主备复制原理

MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)
MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

canal 工作原理

canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
canal 解析 binary log 对象(原始为 byte 流)

前置准备

首先要检测mysql是否开启binlog,如果未开启需要修改my.cnf配置文件，进行开启

--查看是否开启bin_log日志，value = on表示开启
SHOW VARIABLES LIKE 'log_bin';

--查看最新的bin_log

show master status;
-- 查看bin_log日志文件
SHOW BINARY LOGS;
--查看bin_log写入状态
SHOW MASTER STATUS;
--查看bin_log存储格式 = row
SHOW VARIABLES LIKE 'binlog_format';

#mysql开启binlog

        # 打开binlog

        log-bin=mysql-bin

        # 选择ROW(行)模式

        binlog-format=ROW

        # 配置MySQL replaction需要定义，不要和canal的slaveId重复

        server_id=1

mysql创建一个用户，也可以使用自己已有的用户

-- 使用命令登录：mysql -u root -p
-- 创建用户用户名：canal 密码：Canal@123456
create user 'canal'@'%' identified by 'Canal@123456';
-- 授权 *.*表示所有库
grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%' identified by 'Canal@123456';
FLUSH PRIVILEGES;

这样前置初始化条件就完成了。

安装canal-admin

将文件下载下来之后解压缩得到如下文件

首先打开conf目录下的application.yml文件，可以看到标准的spring-boot配置

server:
  port: 8089
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8

spring.datasource:
  address: 127.0.0.1:3306 #数据库连接地址
  database: canal_manager #数据库名称
  username: #用户名
  password: #密码
  driver-class-name: com.mysql.jdbc.Driver
  url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
  hikari:
    maximum-pool-size: 30
    minimum-idle: 1

canal:
  adminUser: admin #管理员登录用户
  adminPasswd: 123456 #登录密码

首先在修改配置之前要先创建一个canal_manager数据库，用来存储canal的配置，canal_manager数据库的sql脚本在config/canal_manager.sql，与这个配置文件同级。运行一下就好了。运行后的表：

然后将上面的address(数据库连接地址)、username(用户名),password(密码)修改一下，改成你本地的连接。

adminUser和adminPasswd默认用户名密码是：admin和123456。

启动canal-admin：在bin/startup.bat(我是windows系统)双击启动

启动后没有报错就是成功了，在浏览器输入登录地址localhost:8089,用户名密码：admin 123456就进入管理界面了，到这canal-admin算是配置玩了，简单吧

安装canal.deployer

接下来我们可以安装canal的server端了，正常不用ali的管理视图和同步监听，只安装这一个文件就可以，但咱们要做就做全套呀，主打不写一行代码。

将canal-deployer压缩包解压得到如下的文件。

还是老样子，打开conf文件夹如下：

文件配置选择二选一

这里有一个canal.properties和canal_local.properties两个文件

如果单机配置使用canal_local.properties配置文件好点，因为配置少，如果使用canal_local.properties配置文件的话，就把canal_local.properties名称改为canal.properties，之前的canal.properties可以换个名字或者删掉。

canal_local.properties

贴一下canal_local.properties的配置让你们感受一下

# register ip
canal.register.ip =

# canal admin config
canal.admin.manager = 127.0.0.1:8089 #canal-admin的连接地址
canal.admin.port = 11110 #默认
canal.admin.user = admin #canal-admin的用户名
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441 #canal-admin的登录密码123456的密文自己去上面说的数据库canal_manager的表canal_user里查看
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster =
canal.admin.register.name =

当然使用默认的canal.properties也可以，自己选择，我们就以默认的canal.properties为例，

canal.properties配置：

单机配置只需要修改这几个参数就可以

canal.admin.manager = 127.0.0.1:8089 #canal-admin的连接地址
canal.admin.port = 11110 #默认
canal.admin.user = admin #canal-admin的用户名
canal.admin.passwd = 6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9 #canal-admin的登录密码123456的密文，取值见上面说明

# 启动端口，也是客户端连接的端口
canal.port = 11111
# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ 与canal 连接的客户端
# 如果是通过代码进行连接，这里为tcp
canal.serverMode = tcp
# canal 加载mysql 的实例
canal.destinations = example

canal.destinations理解，引用之前一个博主的说法。

canal.properties的整体配置：

#################################################
######### common argument#############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =127.0.0.1
canal.port = 11111
canal.metrics.pull.port = 11112
# canal instance user/passwd
# canal.user = canal
# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458

# canal admin config
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9
# admin auto register
canal.admin.register.auto = true
#canal.admin.register.cluster =
#canal.admin.register.name =

canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ
canal.serverMode = tcp
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000
## memory store RingBuffer size, should be Math.pow(2,n)
canal.instance.memory.buffer.size = 16384
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024 
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE
canal.instance.memory.rawEntry = true

## detecing config
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.sql = select 1
canal.instance.detecting.interval.time = 3
canal.instance.detecting.retry.threshold = 3
canal.instance.detecting.heartbeatHaEnable = false

# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
canal.instance.transaction.size =  1024
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = 60

# network config
canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.network.soTimeout = 30

# binlog filter config
canal.instance.filter.druid.ddl = true
canal.instance.filter.query.dcl = false
canal.instance.filter.query.dml = false
canal.instance.filter.query.ddl = false
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
canal.instance.filter.transaction.entry = false
canal.instance.filter.dml.insert = false
canal.instance.filter.dml.update = false
canal.instance.filter.dml.delete = false

# binlog format/image check
canal.instance.binlog.format = ROW,STATEMENT,MIXED 
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB

# binlog ddl isolation
canal.instance.get.ddl.isolation = false

# parallel parser config
canal.instance.parser.parallel = true
## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
#canal.instance.parser.parallelThreadSize = 16
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256

# table meta tsdb info
canal.instance.tsdb.enable = true
canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername = canal
canal.instance.tsdb.dbPassword = canal
# dump snapshot interval, default 24 hour
canal.instance.tsdb.snapshot.interval = 24
# purge snapshot expire , default 360 hour(15 days)
canal.instance.tsdb.snapshot.expire = 360

#################################################
######### destinations#############
#################################################
canal.destinations = example
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = 5
# set this value to 'true' means that when binlog pos not found, skip to latest.
# WARN: pls keep 'false' in production env, or if you know what you want.
canal.auto.reset.latest.pos.mode = false

canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml

canal.instance.global.mode = spring
canal.instance.global.lazy = false
canal.instance.global.manager.address = ${canal.admin.manager}
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
canal.instance.global.spring.xml = classpath:spring/file-instance.xml
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml

##################################################
#########       MQ Properties      #############
##################################################
# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =
canal.aliyun.uid=

canal.mq.flatMessage = true
canal.mq.canalBatchSize = 50
canal.mq.canalGetTimeout = 100
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.accessChannel = local

canal.mq.database.hash = true
canal.mq.send.thread.size = 30
canal.mq.build.thread.size = 8

##################################################
#########      Kafka      #############
##################################################
kafka.bootstrap.servers = 127.0.0.1:9092
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0

kafka.kerberos.enable = false
kafka.kerberos.krb5.file = ../conf/kerberos/krb5.conf
kafka.kerberos.jaas.file = ../conf/kerberos/jaas.conf

# sasl demo
# kafka.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \\n username=\"alice\" \\npassword="alice-secret\";
# kafka.sasl.mechanism = SCRAM-SHA-512
# kafka.security.protocol = SASL_PLAINTEXT

##################################################
#########     RocketMQ     #############
##################################################
rocketmq.producer.group = test
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic =
rocketmq.namespace =
rocketmq.namesrv.addr = 127.0.0.1:9876
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
rocketmq.tag = 

##################################################
#########     RabbitMQ     #############
##################################################
rabbitmq.host =
rabbitmq.virtual.host =
rabbitmq.exchange =
rabbitmq.username =
rabbitmq.password =
rabbitmq.deliveryMode =


##################################################
#########       Pulsar         #############
##################################################
pulsarmq.serverUrl =
pulsarmq.roleToken =
pulsarmq.topicTenantPrefix =

配置instance.properties

路径在conf/example/instance.properties,因为我们在canal.properties指定了canal.destinations = example，所以要在对应的example文件夹下配置instance。这里引用了其他博主的文件描述很详细，我就拿过来了

## mysql serverId , v1.0.26+ will autoGen
## v1.0.26版本后会自动生成slaveId，所以可以不用配置
# canal.instance.mysql.slaveId=0

# 数据库地址
canal.instance.master.address=127.0.0.1:3306
# binlog日志名称
canal.instance.master.journal.name=mysql-bin.000001
# mysql主库链接时起始的binlog偏移量
canal.instance.master.position=154
# mysql主库链接时起始的binlog的时间戳
canal.instance.master.timestamp=
canal.instance.master.gtid=

# username/password
# 在MySQL服务器授权的账号密码
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
# 字符集
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
# canal.instance.defaultDatabaseName 默认那个库，这里指定为test（需要在MySQL中建立一个test库）

# table regex .*\\..*表示监听所有表也可以写具体的表名，用，隔开
canal.instance.filter.regex=.*\\..*
# mysql 数据解析表的黑名单，多个表用，隔开
canal.instance.filter.black.regex=

canal.instance.filter.regex规则事例，主要就是监听哪些表进行复制：

https://github.com/alibaba/canal/wiki/AdminGuide#properties%E9%85%8D%E7%BD%AE%E6%96%87%E4%BB%B6

1. 所有表：.* or .*\\..*
2. canal schema下所有表： canal\\..*
3. canal下的以canal打头的表：canal\\.canal.*
4. canal schema下的一张表：canal\\.test1

5. 多个规则组合使用：canal\\..*,mysql.test1,mysql.test2 (逗号分隔)

这个配置完成之后就可以启动了

双击bin/startup.bat目录下的脚本文件，如果没有报错的话就启动成功了

然后在去刷新我们的canal-admin管理页面，可以看到这个server实例已经注入进来了。

通过canal-admin管理页面管理server的话，需要在instance管理里增加对应的instance配置，至于为什么没有从canal-deployer的example读取过来不了解。

在instance管理里点击‘新建instance’，在页面里点击导入模板修改配置，或者将之前example文件夹下的instance.properties复制过来就可以。然后起一个名字，选择你的实例保存。

成功添加成功之后列表就会显示你新增的配置

安装canal.adapter

将canal.adapter-1.1.7解压得到如下的问价夹

老样子打开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 指定连接类型  对应consumerProperties下的配置
  flatMessage: true
  zookeeperHosts:
  syncBatchSize: 1000 #批量数量
  retries: 10  #循环次数 -1表示无线循环
  timeout:
  accessKey:
  secretKey:
  consumerProperties:
    # canal tcp consumer
    canal.tcp.server.host: 127.0.0.1:11111   #canal.deployer的地址和端口
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    #canal.tcp.username: canal
    #canal.tcp.password: canal
    # 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://127.0.0.1:3306/zm?useUnicode=true&characterEncoding=UTF-8&useSSL=false  # 对应canal.adapter-1.1.7\conf\rdb下的配置文件中的dataSourceKey的value值
      username: # 用户名
      password: # 密码
  canalAdapters:
  - instance: example # example对应路径canal.deployer-1.1.7\conf\example 和canal.adapter-1.1.7\conf\rdb下的配置文件中的 destination的value值
    groups:
    - groupId: g1 # g1对应canal.adapter-1.1.7\conf\rdb下的配置文件中的 groupId的value值
      outerAdapters:
      # - name: logger
      - name: rdb  #rdb对应路径canal.adapter-1.1.7\conf\rdb
        key: mysql1  # mysql1 对应 canal.adapter-1.1.7\conf\rdb下的配置文件中outerAdapterKey的value值
        #目标数据库连接信息
        properties:
          jdbc.driverClassName: com.mysql.jdbc.Driver
          jdbc.url: jdbc:mysql://目标库地址:3306/zm?useUnicode=true&characterEncoding=UTF-8&useSSL=false
          jdbc.username: #用户名
          jdbc.password: #密码
          druid.stat.enable: false
          druid.stat.slowSqlMillis: 1000

配置rdb下的mytest_user.yml文件：

dataSourceKey: defaultDS
destination: example
groupId: g1
outerAdapterKey: mysql1
concurrent: true
dbMapping:
  database: zm #源数据库
  table: zm_product #源数据库表
  targetTable: zm_product  #目标数据库表
  targetPk: #主键
    id: id
  mapAll: true #表字段一样全部映射 否则使用targetColumns进行映射
#  targetColumns:
#    id:
#    name:
#    role_id:
#    c_time:
#    test1:
#  etlCondition: "where c_time>={}" #数据查询条件
  commitBatch: 3000 # 批量提交的大小


## Mirror schema synchronize config
#dataSourceKey: defaultDS
#destination: example
#groupId: g1
#outerAdapterKey: mysql1
#concurrent: true
#dbMapping:
#  mirrorDb: true
#  database: mytest

他们说有启动失败的要修改.bat文件，我是直接成功了

启动bin/startup.bat脚本文件，启动成功