Canal Admin 架构
Canal 1.1.4 版本引入了 Canal Admin，提供了统一管理 Canal Server 的 WebUI 界面。Canal Admin 的核心概念主要有：
Instance：对应 Canal Server 里的 Instance，一个最小的订阅 MySQL 的队列。
Server：对应 Canal Server，一个 Server 里可以包含多个 Instance，Canal Server 负责订阅 MySQL 的 binlog 数据，可以将数据输出到消息队列或者为 Canal Adapter 提供消息消费。原本 Canal Server 运行所需要的 canal.properties 和 instance.properties 配置文件可以在 Canal Admin WebUI 上进行统一运维，每个 Canal Server 只需要以最基本的配置启动。 (例如 Canal Admin 的地址，以及访问配置的账号、密码即可)
集群：对应一组 Canal Server，通过 Zookeeper 协调主备实现 Canal Server HA 的高可用。
Canal 最初只支持将数据从 MySQL 同步到 Kafka，RabbitMQ 等消息队列中，从 1.1.1 版本开始，Canal 实现了一个配套落地的模块 Canal Adapter，实现对 Canal Server 订阅的 binlog 消息进行消费，支持将数据输出至 HBase，MySQL，Elasticsearch，Kudu 中。
机器规划
准备工作
源库开启 binlog，创建同步用户
CREATE USER canal IDENTIFIED BY 'canal';GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';FLUSH PRIVILEGES;
复制代码
开启 binlog，设置 binlog 模式为 row。
[mysqld]...log_bin = binlog 目录binlog_format = row...
复制代码
安装 Java
部署 Zookeeper 集群
Canal Server 和 Canal Adapter 依赖 Zookeeper 实现 HA 高可用，Zookeeper 安装部署可以参考 Zookeeper 集群搭建。
部署 Canal Admin
下载并解压安装包
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.admin-1.1.5.tar.gzmkdir -p /software/canal-admintar -xzvf canal.admin-1.1.5.tar.gz -C /software/canal-admin
复制代码
初始化 Canal Admin 元数据库
初始化 SQL 脚本里会默认创建名为 canal_manager 的数据库，canal_manager.sql 脚本存放在解压后的 conf 目录下。
#登录 MySQLmysql -h11.8.36.104:3306 -uroot -p123456#初始化元数据库source /software/canal-admin/conf/canal_manager.sql
复制代码
Canal Admin 配置文件
修改配置文件 vim /software/canal-admin/conf/application.yml：
#Canal Admin Web 界面端口server: port: 8089spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8
#元数据库连接信息spring.datasource: address: 11.8.36.104:3306 database: canal_manager username: root password: 123456 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 Server 加入 Canal Admin 使用的密码canal: adminUser: admin adminPasswd: admin
复制代码
启动 Canal Admin
sh /software/canal-admin/bin/startup.sh
复制代码
浏览器输入 http://11.8.36.104:8089 访问 Canal Admin 管理界面，默认用户名 admin，密码 123456。
添加 Canal Server 节点
下载并解压压缩包
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gzmkdir -p /software/canal-servertar -xzvf canal.deployer-1.1.5.tar.gz -C /software/canal-server
复制代码
Canal Server 配置文件
因为这里使用 Canal Admin 部署集群，所以 Canal Server 节点只需要配置 Canal Admin 的连接信息即可，真正的配置文件统一通过 Canal Admin 界面来管理。编辑 vim /software/canal-server/conf/canal_local.properties 文件：
#Canal Server 地址canal.register.ip = 11.8.36.104
#Canal Admin 连接信息canal.admin.manager = 11.8.36.104:8089canal.admin.port = 11110canal.admin.user = admin#mysql5 类型 MD5 加密结果 -- admincanal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
#自动注册canal.admin.register.auto = true#集群名canal.admin.register.cluster = canal-cluster-1#Canal Server 名字canal.admin.register.name = canal-server-1
复制代码
另一台 Canal Server 也是一样的方式配置，修改 canal.register.ip = 11.8.36.105 和 canal.admin.register.name = canal-server-2 即可。
Canal Admin 创建集群
进入 Canal Admin 集群管理页面，点击新建集群，添加集群名称和 Zookeeper 地址。
载入集群配置模板。
载入配置后先不用修改配置，点击保存。
注意载入集群配置必须在 Canal Server 添加进集群之前做，否则 Canal Server 会无法加入集群，有以下报错：
详情可以参考这个 issue。
启动 Canal Server
local 参数表示使用 canal_local.properties 配置启动 Canal Server。
sh /software/canal-server/bin/startup.sh local
复制代码
在 Canal Admin 上查看注册的 Canal Server。
MySQL 同步 MySQL
Canal Properties 配置
在集群管理界面选择相应的集群，选择操作，点击主配置对 canal.properties 进行配置。
MySQL 同步数据到 MySQL 比较麻烦，需要先将源 MySQL 的数据同步到 Canal Server 中内置的消息队列中（或者外部 Kafka,RabbitMQ 等消息队列），然后通过 Canal Adapter 去消费消息队列中的消息再写入目标 MySQL。
集群模式下，虽然有多个 Canal Server，但是只有一个是处于 running 状态，客户端连接的时候会查询 Zookeeper 节点获取并连接处于 running 状态的 Canal Server。
#Zookeeper 地址canal.zkServers = 127.0.0.1:2181,127.0.0.1:3181,127.0.0.1:4181#tcp, kafka, rocketMQ, rabbitMQcanal.serverMode = tcp# 此配置需要修改成 default-instancecanal.instance.global.spring.xml = classpath:spring/default-instance.xml
#TSDB 设置canal.instance.tsdb.enable = truecanal.instance.tsdb.url = jdbc:mysql://11.8.36.104:3306/canal_tsdbcanal.instance.tsdb.dbUsername = rootcanal.instance.tsdb.dbPassword = 123456#使用外部 MySQL 存储表结构变更信息canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
复制代码
Canal Server 默认情况下把源库对表结构修改的记录存储在本地 H2 数据库中，当 Canal Server 主备切换后会导致新的 Canal Server 无法正常同步数据，因此修改 TSDB 的设置将外部 MySQL 数据库作为 Canal Server 存储表结构变更信息的库。
create database canal_tsdb;
复制代码
以下截图为 canal.properties 配置文件中修改的内容。
Canal Instance 配置
查看源库 gtid 信息。
mysql> show master status;+------------------+-----------+--------------+------------------+------------------------------------------------+| File | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set |+------------------+-----------+--------------+------------------+------------------------------------------------+| mysql-bin.000008 | 257890708 | | | 92572381-0b9c-11ec-9544-8cdcd4b157e0:1-1066141 |+------------------+-----------+--------------+------------------+------------------------------------------------+1 row in set (0.01 sec)
mysql>
复制代码
在 Instance 管理界面，点击新建 Instance，保存为 sync_table-instance。Instance 表示一个 MySQL 的订阅队列，需要指定源库的连接信息，需要同步的表，以及同步点位。（binlog+postion 或者 gtid）
# MySQL 集群配置中的 serverId 概念，需要保证和当前 MySQL 集群中 id 唯一 (v1.0.26+版本之后 canal 会自动生成，不需要手工指定)# canal.instance.mysql.slaveId=0# 源库地址canal.instance.master.address=11.17.6.185:4679
# 方式一：binlog + postion# mysql源库起始的binlog文件# canal.instance.master.journal.name=mysql-bin.000008# mysql主库链接时起始的binlog偏移量# canal.instance.master.position=257890708 # mysql主库链接时起始的binlog的时间戳# canal.instance.master.timestamp=
# 方式二：gtid(推荐)# 启用 gtid 方式同步canal.instance.gtidon=true# gtidcanal.instance.master.gtid=92572381-0b9c-11ec-9544-8cdcd4b157e0:1-1066141
# 源库账号密码canal.instance.dbUsername=canalcanal.instance.dbPassword=canal# MySQL 数据解析编码canal.instance.connectionCharset = UTF-8
# MySQL 数据解析关注的表，Perl 正则表达式canal.instance.filter.regex=acpcanaldb.*
# MySQL 数据解析表的黑名单，过滤掉不同步的表# canal.instance.filter.black.regex=mysql\\.slave_.*
复制代码
以下截图为 instance.propertios 配置文件中修改的内容。
如果启动 Instance 的时候有以下报错:
CanalParseException: use gtid and TableMeta TSDB should be config timestamp > 0
复制代码
就设置下 canal.instance.master.timestamp= ，参数值为时间对应的纪元时间以来的毫秒数。
查看 Instance 下提供服务的 Canal Server。
ls /otter/canal/destinations/sync_table-instance/cluster[11.8.36.104:11111, 11.8.36.105:11111]#处于 running 状态的 Canal Serverget /otter/canal/destinations/sync_table-instance/running{"active":true,"address":"11.8.36.104:11111"}
复制代码
部署 Canal Adapter
下载并解压安装包
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.adapter-1.1.5.tar.gzmkdir -p /software/canal-adaptertar -xzvf canal.adapter-1.1.5.tar.gz -C /software/canal-adapter
复制代码
Canal Adapter 配置文件
由于 Canal Server 只支持将数据输出到消息队列中，因此需要额外部署 Canal Adapter 订阅 Canal Server 中的消息，然后写入目标 MySQL。编辑 vim /software/canal-adapter/conf/application.yml 文件：
#Canal Adapter Rest API 接口server: port: 8081spring: 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: 0 timeout: accessKey: secretKey: consumerProperties: #Canal Server 单机时配置 canal.tcp.server.host: #Canal Server 高可用时 Canal Adapter 通过 Zookeeper 识别哪台 Canal Server 对外提供服务 canal.tcp.zookeeper.hosts: 11.8.36.125:2181,11.8.38.116:2181,11.8.38.120:2181 canal.tcp.batch.size: 500 canal.tcp.username: canal.tcp.password: #源库连接信息，这里感觉有点多余，根据 Instance 配置也可以获取到源库信息的 srcDataSources: mySyncDS: #和 rdb 中配置文件对应 url: jdbc:mysql://11.17.6.185:4679/acpcanaldb?useUnicode=true username: canal password: canal canalAdapters: #Instance 名字 - instance: sync_table-instance # canal instance Name or mq topic name groups: #组名，Canal Adapter 根据 groupId 实现高可用，两个相同 groupId 的 Canal Adapter 中只有一个会工作，另一个作为备份 #Canal Adapter 也是通过 Zookeeper 来协调主备的 - groupId: group1 outerAdapters: - name: rdb #目录名，后面要在 rdb 目录中配置源库和目标库的映射信息 key: mysql1 #目标库连接信息 properties: jdbc.driverClassName: com.mysql.jdbc.Driver jdbc.url: jdbc:mysql://11.8.36.21:3306/acpcanaldb?useUnicode=true jdbc.username: root jdbc.password: 123456
复制代码
在 /software/canal-adapter/conf/rdb 目录下创建一个结尾为 yml 的文件：
#和 application.yml 文件中对应dataSourceKey: mySyncDS#Instance 名字destination: sync_table-instance#Canal Adapter GroupIdgroupId: group1outerAdapterKey: mysql1concurrent: true#源库和目标库结构全部对应dbMapping: mirrorDb: true database: acpcanaldb
复制代码
备 Canal Adapter 的配置和主 Canal Adapter 所有配置一致。
启动 Canal Adapter
sh /software/canal-adapter/bin/startup.sh
复制代码
同步测试
准备数据
在源库中创建表，要支持 DELETE 和 UPDATE 的同步操作，源库中的表必须有主键或者唯一约束。
create database acpcanaldb;create table acpcanaldb.sync_table(id int primary key,name varchar(20),age int);insert into acpcanaldb.sync_table values(1,'tom',18),(2,'jack',18),(3,'mike',20),(4,'cris',18),(5,'mary',19);
复制代码
查看当前源库数据：
mysql> select * from acpcanaldb.sync_table;+------+------+------+| id | name | age |+------+------+------+| 1 | tom | 18 || 2 | jack | 18 || 3 | mike | 20 || 4 | cris | 18 || 5 | mary | 19 |+------+------+------+5 rows in set (0.01 sec)
复制代码
新增数据
在源库插入数据：
insert into acpcanaldb.sync_table values (6,'peter',18);
复制代码
目标库可以正常同步数据。
修改数据
在源库修改数据：
update acpcanaldb.sync_table set name='tom2' where name='tom';
复制代码
目标库可以正常同步数据。
删除数据
在源库删除数据：
delete from acpcanaldb.sync_table where name='jack';
复制代码
目标库可以正常同步数据。
修改表结构
在源库新增一列 loc，设置默认值 china。
alter table acpcanaldb.sync_table add loc varchar(20) default 'china';
复制代码
目标库可以正常同步新增的 loc 列。
在源库删除一列 age。
alter table acpcanaldb.sync_table drop age;
复制代码
目标库也同步删除了 age 列。
在我们前面设置的 11.8.36.104 的 canal_tsdb 数据库中的 meta_history 表可以看到记录的 DDL 相关操作。
mysql> select * from canal_tsdb.meta_history;+----+---------------------+---------------------+---------------------+------------------+---------------+------------------+------------------+------------+------------+------------+---------------------------------------------------------------------------------+----------+-------+| id | gmt_create | gmt_modified | destination | binlog_file | binlog_offest | binlog_master_id | binlog_timestamp | use_schema | sql_schema | sql_table | sql_text | sql_type | extra |+----+---------------------+---------------------+---------------------+------------------+---------------+------------------+------------------+------------+------------+------------+---------------------------------------------------------------------------------+----------+-------+| 1 | 2021-09-15 02:27:53 | 2021-09-15 02:27:53 | sync_table-instance | mysql-bin.000009 | 59658313 | 3778449602 | 1631672873000 | acpcanaldb | acpcanaldb | sync_table | create table acpcanaldb.sync_table(id int primary key,name varchar(20),age int) | CREATE | NULL || 2 | 2021-09-15 02:36:34 | 2021-09-15 02:36:34 | sync_table-instance | mysql-bin.000009 | 60124865 | 3778449602 | 1631673394000 | acpcanaldb | acpcanaldb | sync_table | alter table acpcanaldb.sync_table add loc varchar(20) default 'china' | ALTER | NULL || 3 | 2021-09-15 02:38:06 | 2021-09-15 02:38:06 | sync_table-instance | mysql-bin.000009 | 60207149 | 3778449602 | 1631673486000 | acpcanaldb | acpcanaldb | sync_table | alter table acpcanaldb.sync_table drop age | ALTER | NULL |+----+---------------------+---------------------+---------------------+------------------+---------------+------------------+------------------+------------+------------+------------+-------------------------------------------------------------------------------
复制代码
事务
源数据库的事务隔离级别是 MySQL 默认的读已提交。
mysql> select @@tx_isolation;+----------------+| @@tx_isolation |+----------------+| READ-COMMITTED |+----------------+1 row in set, 1 warning (0.01 sec)
复制代码
在源库上开启事务插入 2 条数据，根据下图可以看到当源库未提交时，目标库不会同步数据，只有当源库提交事务后，目标库才会同步数据。
如果源库上的事务回滚了，对目标库没有影响，因为原本只有提交的事务才会往目标库同步。
高可用测试
高可用测试分为 3 种场景：
1.Canal Server 故障：通过 Zookeeper 协调，备 Canal Server 转为 running 状态提供服务。
2.Canal Admin 故障：管理节点故障并影响正在运行的任务。
3.Canal Adapter 故障：通过 Zookeeper 协调，备 Canal Adapter 转为 running 状态提供服务。
查看当前源库数据：
mysql> select * from acpcanaldb.sync_table;+----+---------+---------+| id | name | loc |+----+---------+---------+| 1 | tom2 | china || 3 | mike | china || 4 | cris | china || 5 | mary | china || 6 | peter | china |+----+---------+---------+6 rows in set (0.01 sec)
复制代码
Canal Server 故障
登录 Zookeeper 查看当前 Canal Server 运行情况，当前以 HA 的方式运行两个 Canal Server，只有一个 Canal Server 会提供服务，另一个作为热备。当前提供服务的 Canal Server 是 11.8.36.105:11111。
root@ydt-net-kafka1:/usr/local/kafka/bin #./zookeeper-shell.sh 11.8.36.125:2181Connecting to 11.8.36.125:2181Welcome to ZooKeeper!JLine support is disabled
WATCHER::
WatchedEvent state:SyncConnected type:None path:null#当前提供服务的 Canal Serveget /otter/canal/destinations/sync_table-instance/running{"active":true,"address":"11.8.36.105:11111"}#当前运行的两台 Canal Serverls /otter/canal/destinations/sync_table-instance/cluster[11.8.36.104:11111, 11.8.36.105:11111]
复制代码
当然在 Canal Admin 界面也可以看到当前处于 running 状态的 Canal Server。
停止 Canal Server 11.8.36.105:11111。
sh /software/canal-server/bin/stop.sh
复制代码
在 Canal Admin 界面可以看到这个 Canal Server 此时已经断开连接。
在 Zookeeper 上查询正在提供服务的 Canal Server，此时已经切换到 Canal Server 11.8.36.104:11111 了。
get /otter/canal/destinations/sync_table-instance/running{"active":true,"address":"11.8.36.104:11111"}
复制代码
在源库插入一条数据：
insert into acpcanaldb.sync_table values(7,'lisa','england');
复制代码
目标库此时仍然可以正常同步数据。
Canal Admin 故障
停止 Canal Admin：
sh /software/canal-admin/bin/stop.sh
复制代码
在源库插入一条数据：
insert into acpcanaldb.sync_table values(8,'handson','spain');
复制代码
在 Canal Server 挂掉一台并且 Canal Admin 也挂掉的时候，存活的 Canal Server 依然可以正常同步源库和目标库之间的数据。
Canal Adapter 故障
在 Zookeeper 上查看当前执行同步任务的 Canal Adapter：
get /otter/canal/destinations/sync_table-instance/1001/running{"active":true,"address":"11.8.36.104:48380","clientId":1001}
复制代码
停止 Canal Adapter 11.8.36.104：
sh /software/canal-adapter/bin/stop.sh
复制代码
查看 Zookeeper 此时 Canal Adapter 已经切换到了 11.8.36.105：
get /otter/canal/destinations/sync_table-instance/1001/running{"active":true,"address":"11.8.36.105:37000","clientId":1001}
复制代码
往源库中插入一条数据：
insert into acpcanaldb.sync_table values (11,'james','italy');
复制代码
此时源库和目标库数据可以正常同步。
MySQL 主从切换
通常在生产环境中，我们通常会部署双主模式的 MySQL，通过 keepalived 对外提供一个虚拟 IP，当主库发生故障时，虚拟 IP 漂移到备库上，由备库接管服务。关于 MySQL 双主高可用部署可以参考 MySQL + Keepalived 双主热备搭建。
查看源库当前数据：
mysql> select * from acpcanaldb.sync_table;+----+---------+---------+| id | name | loc |+----+---------+---------+| 1 | tom2 | china || 3 | mike | china || 4 | cris | china || 5 | mary | china || 6 | peter | china || 7 | lisa | england || 8 | handson | spain |+----+---------+---------+7 rows in set (0.02 sec)
复制代码
MySQL 源库发送主从切换后，往源库插入一条新数据。
insert into acpcanaldb.sync_table values(9,'james','italy');
复制代码
目标库可以正常同步数据。
MySQL 同步 Kafka
MySQL 数据同步到 Kafka 比 MySQL 同步到 MySQL 简单些，不需要 Canal Adapter，Canal Server 可以直接将 MySQL 的数据输出到 Kafka。
数据准备
在源库中创建一张表。
create table acpcanaldb.table1(id int primary key,name varchar(20),age int);
复制代码
Canal Instance 配置
和前面同步 MySQL 的 Instance 相比，这里有 2 个地方需要修改：
# 静态匹配#canal.mq.topic=acpcanaldb_table1# 动态匹配# 匹配所有表，则每个表都会发送到各自表名的topic# 例如数据库名为 acpcanaldb，表名为 table1，则 topic 名字为 acpcanaldb_table1# 这里的动态匹配有个 bug，如果源库是的表中有下划线，例如前面的 sync_table，将无法同步成功canal.mq.dynamicTopic=.*\\..*# 数据同步到消息队列Kafka版Topic的指定分区。# canal.mq.partition=0# 以下两个参数配置与canal.mq.partition 互斥。配置以下两个参数可以使数据发送至 Kafka Topic 的不同分区。# 发往 Kafka 的 3 个分区canal.mq.partitionsNum=3#库名.表名: 唯一主键，多个表之间用逗号分隔。#canal.mq.partitionHash=mytest.person:id,mytest.role:id
复制代码
关于消息队列中 Topic 和源库表的映射关系可以参考 Canal Kafka RocketMQ QuickStart。在 Instance 管理界面点击新增 Instance，保存为 sync_table-kafka。
Kafka 创建 Topic
bin/kafka-topics.sh --bootstrap-server 11.8.36.125:9092 \--create --topic acpcanaldb_table1 \--partitions 3 --replication-factor 3
复制代码
查看创建的 Topic：
bin/kafka-topics.sh --list --bootstrap-server 11.8.36.125:9092
#返回结果acpcanaldb_table1
复制代码
数据同步
先在 Kafka 上开启消费消息：
bin/kafka-console-consumer.sh --bootstrap-server 11.8.36.125:9092 \--topic acpcanaldb_table1
复制代码
往源库中插入一条数据：
insert into acpcanaldb.table1 values(2,'rose',20);
复制代码
Kafka 上可以看到消费消息：
{"data":[{"id":"2","name":"rose","age":"20"}],"database":"acpcanaldb","es":1631806308000,"id":4,"isDdl":false,"mysqlType":{"id":"int","name":"varchar(20)","age":"int"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"name":12,"age":4},"table":"table1","ts":1631806353089,"type":"INSERT"}
复制代码
监控
Canal Server 性能指标监控基于 Prometheus 的实现，可以把监控的指标信息收集到 Prometheus 中，然后通过 Grafana 展示。
部署 Prometheus
下载并解压安装包
wget https://github.com/prometheus/prometheus/releases/download/v2.30.0/prometheus-2.30.0.linux-amd64.tar.gztar -xzvf prometheus-2.30.0.linux-amd64.tar.gzcd cd prometheus-2.30.0.linux-amd64
复制代码
Prometheus 配置文件
编辑 Prometheus 配置文件 vim prometheus.yml，添加 canal 的 job。
- job_name: "canal" static_configs: - targets: ["11.8.36.104:11112","11.8.36.105:11112"]
复制代码
启动 Prometheus：
查看监控项
浏览器输入 http://11.8.37.41:9090 访问 Prometheus 界面。
部署 Grafana
下载并解压安装包
wget https://s3-us-west-2.amazonaws.com/grafana-releases/release/grafana-5.2.2.linux-amd64.tar.gztar -xzvf grafana-5.2.2.linux-amd64.tar.gzcd grafana-5.2.2
复制代码
启动 Grafana Server
配置监控图表
浏览器输入 http://11.8.37.41:3000 访问 Grafana 界面。用户名 admin，密码 admin。
添加 Prometheus 数据源。
导入模板，将 json 文本黏贴到输入框中，点击 load。
选择刚刚创建的数据源，点击 import。
效果示意图如下：
具体的监控指标可以参考 Prometheus监控。
参考资料
欢迎关注
评论