canal 是Alibaba开源的一款产品，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。
基于日志增量订阅和消费的业务包括
- 数据库镜像
- 数据库实时备份
- 索引构建和实时维护(拆分异构索引、倒排索引等)
- 业务 cache 刷新
- 带业务逻辑的增量数据处理
canal github地址：https://github.com/alibaba/canal
工作原理
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 流)
安装部署
下载文件
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.adapter-1.1.5.tar.gz
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.admin-1.1.5.tar.gz
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz
MySQL配置
配置binlog的format为ROW
创建用户访问
| CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; grant all privileges on canal_manager.* to canal@'%'; -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ; FLUSH PRIVILEGES; |
安装canal-admin
| mkdir /opt/canal-admin && tar zxvf canal.admin-1.1.5.tar.gz -C /opt/canal-admin/ cd /opt/canal-admin && vim conf/application.yml |
配置文件
| [root@cdctest canal-admin]# cat conf/application.yml server: port: 8999 spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 spring.datasource: canal: |
初始化数据
| root@(none)02:48:41>use canal_manager; Database changed root@canal_manager02:48:49>source conf/canal_manager.sql root@canal_manager02:49:00>show tables; +-------------------------+ | Tables_in_canal_manager | +-------------------------+ | canal_adapter_config | | canal_cluster | | canal_config | | canal_instance_config | | canal_node_server | | canal_user | +-------------------------+ 6 rows in set (0.01 sec) |
启动web服务
| cd /opt/canal-admin && bin/startup.sh |
查看服务
| [root@cdctest canal-admin]# netstat -ntlp |grep 8999 tcp 0 0 0.0.0.0:8999 0.0.0.0:* LISTEN 28832/java |
启动如果有错，查看日志
| tail -f logs/admin.log |
登陆页面
| Canal Admin 的 web 访问地址：http://127.0.0.1:8999/ 登录用户名：admin 登录密码：123456 |
安装Canal Deploy
解压缩
| mkdir /opt/canal && tar zxvf canal.deployer-1.1.5.tar.gz -C /opt/canal |
修改配置文件
| [root@cdctest canal]# cd conf [root@cdctest conf]# cat canal.properties # register ip canal.register.ip = # canal admin config |
| 注意 passwd 后面的字符串是 admin 在 MySQL 里的密文。这个密码跟前面 Canal Admin 配置文件里的密码保持一致。如果前面密码改了，这里也要相应修改，密文的值可以通过 MySQL 的 password 方法获取。 |
| MariaDB [canal_manager]> select password('admin'); +-------------------------------------------+ | password('admin') | +-------------------------------------------+ | *4ACFE3202A5FF5CF467898FC58AAB1D61502**** | +-------------------------------------------+ 1 row in set (0.00 sec) |
启动服务
| sh bin/startup.sh |
图形化部署
如果使用 Canal Admin 管理 Canal server，则登录 admin 的管理界面
http://127.0.0.1:8999/#/canalServer/nodeServers
选择 Canal Server > Server 管理，单击 新建 Server。
在弹出的 新建Server信息 界面，填入相应信息(如果启动了canal-deploy，会自动添加进来)，单击 确定 按钮。
之后，选择 Canal Server > Instance 管理 ，单击 新建 Instance。
单击 载入模板 ，显示一个配置文件，跟前面看到的类似。修改配置文件中的跟源端 canal 和数据库有关的信息。
Instance中以下两个配置分别表示同步过滤的条件
| canal.instance.filter.regex=.*\\..* canal.instance.filter.black.regex= |
查看日志
server日志：
| vi logs/canal/canal.log |
instance日志：
| tail -f logs/canal/canal.log tail -f logs/example/example.log tail -f logs/mariadb/mariadb.log |
安装canal-adapter
解压目录
| mkdir /opt/adapter && tar zxvf canal.adapter-1.1.5.tar.gz -C /opt/adapter/ |
修改启动器配置文件
首先指定 adapter 源端类型，通过 mode 指定。这里选择 tcp。后面就要指定 canal.tcp 相关属性，包括 canal server 的 IP 和 端口，数据库的连接用户和密码。之后指定 adapter 目标端连接信息。instance 是源端实例名称，在 canal 部署的时候定义的。
说明 如果没有用 Canal Admin 部署，沿用的是 example 这个名称；如果用了 Canal Admin 部署 instance，前面命名的是 mysqldb。
key 是自定义，名字后面有用。jdbc 相关属性是目标端 OceanBase MySQL 的连接方式，可以使用 MySQL 自带的驱动。
| [root@cdctest adapter]# cat 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: canalAdapters: |
RDB 映射文件
修改 conf/rdb/mytest_user.yml 文件。
映射有两种：一是按表映射；二是整库映射。下面示例是整库映射。
| [root@cdctest adapter]# cat conf/rdb/mytest_user.yml ## Mirror schema synchronize config dataSourceKey: defaultDS destination: mysqldb groupId: g1 outerAdapterKey: obmysql concurrent: true dbMapping: mirrorDb: true database: canaltest commitBatch: 500 |
其中，destination 指定的是 canal instance 名称；outerAdapterKey 是前面定义的 key；mirrorDb 指定数据库级别 DDL 和 DML 镜像同步。
导入的类型以目标表的元类型为准, 将自动进行类型转换。
启动RDB
如果使用了 OceanBase 的驱动，则将目标库 OceanBase 驱动包放入 lib文件夹。
启动 canal-adapter 启动器。
| bin/startup.sh |
查看RDB日志
| tail -f logs/adapter/adapter.log 2023-12-27 15:41:24.508 [pool-8-thread-1] INFO c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":null,"database":"canaltest","destination":"mysqldb","es":1670492484000,"groupId":"g1","isDdl":true,"old":null,"pkNames":null,"sql":"create table t1(id int, name varchar(20))","table":"t1","ts":1670492484508,"type":"CREATE"} 2023-12-27 15:41:24.509 [pool-8-thread-1] DEBUG c.a.o.c.c.adapter.rdb.service.RdbMirrorDbSyncService - DDL: {"data":null,"database":"canaltest","destination":"mysqldb","es":1670492484000,"groupId":"g1","isDdl":true,"old":null,"pkNames":null,"sql":"create table t1(id int, name varchar(20))","table":"t1","ts":1670492484508,"type":"CREATE"} 2023-12-27 15:42:02.099 [pool-8-thread-1] INFO c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":null,"database":"","destination":"mysqldb","es":1670492522000,"groupId":"g1","isDdl":false,"old":null,"pkNames":[],"sql":"insert into t1 values(1,'zx')","table":"t1","ts":1670492522099,"type":"QUERY"} 2023-12-27 15:42:02.099 [pool-8-thread-1] INFO c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":1,"name":"zx"}],"database":"canaltest","destination":"mysqldb","es":1670492522000,"groupId":"g1","isDdl":false,"old":null,"pkNames":[],"sql":"","table":"t1","ts":1670492522099,"type":"INSERT"} 2023-12-27 15:42:02.240 [pool-5-thread-1] DEBUG c.a.o.canal.client.adapter.rdb.service.RdbSyncService - DML: {"data":{"id":1,"name":"zx"},"database":"canaltest","destination":"mysqldb","old":null,"table":"t1","type":"INSERT"} |
