4、Canal入门：Canal是什么、Canal工作原理、Canal的安装与启动、Canal连接MySQL测试
好，咱们进入第四章。前面几章我们把CDC的概念、场景和整体架构都聊透了。从这一章开始，要动真格的了——上手第一个核心组件：Canal。
Canal这东西，说白了就是阿里巴巴开源的一个MySQL binlog解析工具。我最早接触它是在2018年，那时候公司要做订单数据的实时同步到Elasticsearch，调研了一圈，最后选了Canal。嗯，这一用就是好几年，踩过坑，也填过坑，今天把这些经验都揉碎了讲给你。
4.1 Canal是什么
Canal，全称是「Canal Alibaba」，中文意思是「运河」。名字挺形象——它就像一条数据运河，把MySQL的数据源源不断地运送到下游系统。
它的核心能力就一句话：伪装成MySQL的从库，拉取主库的binlog，然后解析成你可以消费的数据格式。
你想想看，MySQL主从复制大家都知道吧？主库写binlog，从库拉binlog然后回放。Canal干的活，本质上跟从库一模一样。只不过从库把binlog拿来写数据，Canal把binlog拿来给你做业务处理。
核心定位： Canal是一个中间件，它不存储数据，只做数据的搬运和解析。它支持的数据格式包括JSON、Avro、Protobuf等，最常用的是JSON。
我个人习惯把Canal放在数据链路的最前端，紧挨着MySQL。它后面可以接Kafka、RocketMQ、Elasticsearch，甚至直接写文件。灵活性非常高。
4.2 Canal工作原理
理解Canal的工作原理，其实就理解了MySQL主从复制的精髓。我画了一张图，你看完就明白了。
整个流程其实就三步：
- 伪装：Canal把自己伪装成一个MySQL从库，向主库发送dump请求。
- 拉取：主库把binlog推送给Canal，Canal接收并缓存。
- 解析：Canal把binlog的二进制格式解析成结构化事件，比如INSERT、UPDATE、DELETE。
这里有个关键点——binlog的格式。MySQL的binlog有三种格式：STATEMENT、ROW、MIXED。Canal要求MySQL必须开启ROW模式，因为只有ROW模式才能记录每一行数据的变化细节。STATEMENT模式只记录SQL语句，解析出来你根本不知道具体改了哪些数据。
注意： 如果你的MySQL binlog格式是STATEMENT或MIXED，Canal无法正常工作。我曾经接手过一个项目，对方说Canal连不上，我一看binlog_format还是STATEMENT……改完重启就好了。
4.3 Canal的安装与启动
安装Canal其实不复杂，但有几个前置条件必须满足。我按步骤来，你跟着做就行。
4.3.1 前置条件
| 组件 | 版本要求 | 说明 |
|---|---|---|
| MySQL | 5.7+ 或 8.0+ | 必须开启binlog，格式为ROW |
| JDK | 1.8+ | Canal基于Java开发 |
| 操作系统 | Linux / macOS / Windows | 推荐Linux，生产环境稳定 |
4.3.2 MySQL配置
首先，确保你的MySQL已经开启了binlog。检查一下my.cnf（或my.ini）配置文件：
[mysqld]
# 开启binlog
log-bin=mysql-bin
# binlog格式必须为ROW
binlog-format=ROW
# 服务器ID，主从复制需要唯一
server-id=1
# 需要同步的数据库（可选，不配置则同步所有）
binlog-do-db=test_db
配置完重启MySQL，然后登录验证：
mysql> SHOW VARIABLES LIKE 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin | ON |
+---------------+-------+
mysql> SHOW VARIABLES LIKE 'binlog_format';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW |
+---------------+-------+
嗯，看到ON和ROW就对了。
4.3.3 创建Canal用户
Canal需要连接MySQL，所以得给它一个账号。我个人习惯单独创建一个用户，权限最小化：
mysql> CREATE USER 'canal'@'%' IDENTIFIED BY 'canal123';
mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
mysql> FLUSH PRIVILEGES;
这里注意，REPLICATION SLAVE和REPLICATION CLIENT是必须的，少了任何一个Canal都拉不到binlog。SELECT权限是为了读取表结构信息。
4.3.4 下载与安装Canal
去Canal的GitHub Releases页面下载最新稳定版。我写这篇文章时，最新稳定版是1.1.7。下载后解压：
# 下载
wget https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.deployer-1.1.7.tar.gz
# 解压
tar -zxvf canal.deployer-1.1.7.tar.gz -C /usr/local/canal
解压后的目录结构是这样的：
/usr/local/canal/
├── bin/ # 启动脚本
├── conf/ # 配置文件
├── lib/ # 依赖jar包
├── logs/ # 日志目录
└── plugin/ # 插件目录
4.3.5 配置Canal
核心配置文件在conf/canal.properties和conf/example/instance.properties。前者是全局配置，后者是实例配置。我们主要改实例配置：
# conf/example/instance.properties
# MySQL地址
canal.instance.master.address=127.0.0.1:3306
# 数据库账号
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal123
# 需要订阅的数据库和表
# 格式：dbName.tableName，支持正则
canal.instance.filter.regex=test_db\\..*
# 解析后投递的目标
# tcp: 直接TCP输出（默认）
# kafka: 投递到Kafka
# rocketmq: 投递到RocketMQ
canal.serverMode=tcp
这里有个坑——正则表达式中的反斜杠要转义。我第一次配置时写成了test_db\..*，结果Canal启动后什么数据都抓不到，查了半天日志才发现是正则写错了。
4.3.6 启动Canal
启动很简单，执行脚本就行：
# 启动
/usr/local/canal/bin/startup.sh
# 查看日志
tail -f /usr/local/canal/logs/canal/canal.log
看到类似这样的日志，说明启动成功：
2024-01-15 10:30:00.123 [main] INFO c.a.o.c.CanalLauncher - ## start the canal server.
2024-01-15 10:30:01.456 [main] INFO c.a.o.c.CanalLauncher - ## the canal server is running now ......
小技巧： 启动后可以用jps命令查看Java进程，如果看到CanalLauncher，说明Canal已经跑起来了。
4.4 Canal连接MySQL测试
安装启动都搞定了，接下来验证一下Canal能不能正常拉取数据。Canal自带了一个简单的客户端示例，我们可以用它来测试。
4.4.1 编写测试客户端
创建一个Maven项目，引入Canal客户端依赖：
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.7</version>
</dependency>
然后写一个简单的Java类：
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import java.net.InetSocketAddress;
import java.util.List;
public class CanalClientTest {
public static void main(String[] args) {
// 1. 创建连接
CanalConnector connector = CanalConnectors.newSingleConnector(
new InetSocketAddress("127.0.0.1", 11111),
"example", // instance名称
"", // 用户名（默认空）
"" // 密码（默认空）
);
// 2. 连接并订阅
connector.connect();
connector.subscribe("test_db\\..*");
// 3. 循环拉取数据
while (true) {
// 每次拉取100条，超时时间1秒
Message message = connector.getWithoutAck(100, 1000);
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
continue;
}
// 4. 解析数据
for (Entry entry : message.getEntries()) {
if (entry.getEntryType() == EntryType.ROWDATA) {
RowChange rowChange;
try {
rowChange = RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("解析失败", e);
}
// 获取事件类型
String eventType = rowChange.getEventType().name();
System.out.println("事件类型: " + eventType);
// 遍历行数据
for (RowData rowData : rowChange.getRowDatasList()) {
if (eventType.equals("INSERT") || eventType.equals("UPDATE")) {
System.out.println("新增数据: " + rowData.getAfterColumnsList());
}
if (eventType.equals("DELETE") || eventType.equals("UPDATE")) {
System.out.println("旧数据: " + rowData.getBeforeColumnsList());
}
}
}
}
// 5. 确认消费
connector.ack(batchId);
}
}
}
4.4.2 运行测试
运行这个客户端，然后在MySQL里执行一些操作：
mysql> USE test_db;
mysql> CREATE TABLE user (
id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(50),
age INT
);
mysql> INSERT INTO user(name, age) VALUES('张三', 25);
mysql> UPDATE user SET age = 26 WHERE name = '张三';
mysql> DELETE FROM user WHERE name = '张三';
你会看到客户端控制台输出类似这样的内容：
事件类型: INSERT
新增数据: [id:1, name:张三, age:25]
事件类型: UPDATE
新增数据: [id:1, name:张三, age:26]
旧数据: [id:1, name:张三, age:25]
事件类型: DELETE
旧数据: [id:1, name:张三, age:26]
看到这个输出，说明Canal已经成功抓取到了MySQL的binlog变化。恭喜，你的第一个CDC链路跑通了！
验证要点：
- INSERT事件：只有
afterColumnsList有数据 - UPDATE事件：
beforeColumnsList和afterColumnsList都有数据 - DELETE事件：只有
beforeColumnsList有数据
4.5 常见问题与避坑
最后，分享几个我在项目中遇到过的坑，你遇到了可以直接对照排查。
问题1：Canal启动后日志报错「No suitable driver」
原因：MySQL驱动版本不匹配。Canal 1.1.7默认带的驱动是5.x，如果你连的是MySQL 8.0，需要手动替换驱动为8.x版本。去lib目录下把mysql-connector-java-5.x.jar换成mysql-connector-java-8.x.jar。
问题2：Canal连接MySQL报错「Access denied for user」
原因：用户权限不足。检查一下REPLICATION SLAVE和REPLICATION CLIENT权限是否已经授予。我曾经遇到过只给了SELECT权限，结果Canal一直连不上。
问题3：Canal能启动但抓不到数据
原因：最常见的是binlog格式不是ROW，或者正则表达式写错了。先检查MySQL的binlog_format，再检查instance.properties里的canal.instance.filter.regex。我建议先用test_db\\..*这种最简单的正则测试，通了再细化。