当前位置：首页 > news >正文

Canal

news 2025/9/11 4:15:11

canal译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。

1.canal 工作原理

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

2.安装/配置MySQL

1.安装mysql

已安装略过此步骤，未安装，可以参考linux 安装MySQL-CSDN博客进行安装。

2.开启binlog日志

vi /etc/my.cnf

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 行 模式
#binlog-do-db 根据自己的情况进行修改，指定具体要同步的数据库，如果不配置
则表示所有数据库均开启 Binlog
#binlog-do-db=demo
#binlog-do-db=dmp

注意：需要重启MySQL实例使配置生效，执行命令：sudo systemctl restart mysqld

3.新建canal账号并授权

create user canal@'%' identified with mysql_native_password by 'wsx-123';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;

4.查看binlog日志

1)查看binlog模式：show variables like '%binlog_format%';

2)查看binlog日志文件列表：show binary logs;

3)查看当前正在写入的binlog文件： show master status;

3.安装Canal

3.1下载 canal

wget https://github.com/alibaba/canal/releases/download/canal-1.1.6/canal.adapter-1.1.6.tar.gz

3.2 解压canal

mkdir ~/opt/canal

tar -zxf canal.deployer-1.1.6.tar.gz -C ~/opt/canal

3.3 修改canal 的基本通用配置

vi conf/canal.properties

#canal常用端口
canal.admin.port = 11110
canal.port = 11111
canal.metrics.pull.port = 11112

# 输出 model，默认 tcp，改为输出到: kafka, rocketMQ, rabbitMQ, pulsarMQ
canal.serverMode = tcp

# 指定实例,多个实例使用逗号分隔: canal.destinations = example1,example2
#如果需要多个实例处理不同的MySQL数据的话，直接拷贝出多个conf/example，并对其重新命名，命名和配置文件中指定的名称一致即可
canal.destinations = example

3.4 修改canal 实例配置

vi conf/example/instance.properties

## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=0

# mysql数据库地址:ip+端口
canal.instance.master.address=127.0.0.1:3306

# mysql数据库用户名和密码
canal.instance.dbUsername=canal
canal.instance.dbPassword=wsx-123

#代表数据库的编码方式对应到 java 中的编码类型，比如 UTF-8，GBK , ISO-8859-1
canal.instance.connectionCharset = UTF-8

# 指定库和表，这里的 .* 表示 canal.instance.master.address 下面的所有数据库
canal.instance.filter.regex=.*\\..*

#指定具体数据库和该库所有表
#canal.instance.defaultDatabaseName=demo
#table regex
#canal.instance.filter.regex = demo\\..*