canal数据同步工具介绍与应用

canal服务

• • canal介绍
  • canal版本与环境
  • canal 服务集
  • canal应用
  • • 场景：
  • canal常见问题
  • • xml配置问题
    • 连接认证问题
    • jar版本问题
    • 连接问题

canal介绍

‌1、Canal是‌阿里巴巴开源的‌MySQL增量数据订阅和消费工具，通过模拟MySQL的‌slave与‌master交互，捕获‌binary log事件。
2、Canal的实现基于MySQL的主从复制原理，通过模拟slave的交互协议，向master发送dump请求，master收到请求后开始推送binary log给slave（实际上是Canal），Canal解析这些日志事件并将其转换为JSON格式的数据，然后发送到指定的存储目的地，如MySQL、Kafka、Elastic Search等。‌
3、Canal支持‌HA（高可用），其实现机制依赖‌Zookeeper来实现。
4、Canal可以适应不同的网络环境，包括云环境和本地环境，支持多种网络模型如虚拟局域网(VLAN)、‌VXLAN、Overlay等，并且可以与各种网络设备和云服务提供商进行集成。

canal版本与环境

单机版：适用于开发测试环境或小型生产环境。
配置简单，易于部署和管理。

HA版本
适合生产环境，特别是对数据一致性和系统稳定性有高要求的场景。
可以配置多个 Canal 实例，实现故障转移和负载均衡。
需要更复杂的配置和管理，以确保高可用性

Canal 版本与 JDK 版本的兼容性
canal安装包
canalv1.1.5版本以上就不支持jdk1.8，需要升级到jdk11

canal 服务集

canal.adapter

Canal Adapter 是 Canal 项目的一个重要组成部分，它负责将 Canal Server 采集的 MySQL binlog 数据转换成特定格式，并发送到不同的目标系统。
canal.admin
Canal Admin 是 Canal 项目的一个管理工具，它提供了一个 Web 界面，用于监控和管理 Canal 的各个组件。

canal.deployer

Canal Deployer 是 Canal 项目中的一个模块，它主要用于部署和管理 Canal Server。通过 Canal Deployer，用户可以方便地启动、停止 Canal Server，并配置相关的参数。

canal.example

Canal Example 是 Canal 项目中的一个示例目录，它包含了 Canal 的基本使用案例和配置示例。

canal应用

场景：

1、增量同步
默认是增量方式
直接查看deployer服务下的instance.properties文件

2、全量同步
修改deployer服务下的instance.properties文件
binlog文件：
canal.instance.master.journal.name=binlog文件全名称
canal.instance.master.position=0
canal.instance.master.timestamp=10位时间戳
删除服务端的meta.dat文件:这个文件记录 canal 已经消费的位置等信息

3、中间件集成同步
修改instance.properties文件
kafka
部署 Zookeeper 和 Kafka
canal.properties配置文件：
####tcp, kafka, rocketMQ, rabbitMQ
canal.serverMode = kafka
##################################################
######### 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”
rocketmq
安装 RocketMQ
canal.properties配置文件：
tcp, kafka, rocketMQ, rabbitMQ
canal.serverMode = rocketMQ
##################################################
######### 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 =
修改instance.properties文件
#mq config
canal.mq.topic=canal_topic

canal常见问题

xml配置问题

Error creating bean with name ‘metaHistoryDAO’ defined in class path
resource [spring/tsdb/h2-tsdb.xml] 去掉：canal.properties配置文件中的h2配置

连接认证问题

com.mysql.jdbc.exceptions.jdbc4.MySQLNonTransientConnectionException: Public Key Retrieval is not allowed
修改 MySQL 数据库连接字符串 URL，在末尾追加 “?allowPublicKeyRetrieval=true” 参数

jar版本问题

Mysql连接jar版本 mysql-connector-java-5.1.48.jar
切换成mysql-connector-java-8.0.28.jar

连接问题

Communications link failure。The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server.
解决：
数据库连接地址？allowPublicKeyRetrieval=true&useUnicode=true&characterEncoding=utf8&useSSL=false&tinyInt1isBit=true&serverTimezone=Asia/Shanghai&allowMultiQueries=true&rewriteBatchedStatements=true