📚 课程目录第3章 Canal入门

3. Canal 入门：工作原理、架构与部署

好，咱们开始聊 Canal。这是整个数据同步链路里非常关键的一环。说白了，Canal 就是 MySQL 的“日志翻译官”。它假装自己是 MySQL 的从库，去拉主库的 Binlog，然后翻译成你能消费的数据格式。

我最早接触 Canal 是在 2017 年，那时候公司要做订单库的实时异构同步。当时选型对比了好几个方案，最后还是 Canal 胜出。为什么？因为它对 MySQL 原生协议的支持最成熟，社区也活跃。嗯，咱们今天就把 Canal 的里里外外拆开看看。

3.1 Canal 工作原理

Canal 的核心机制，其实就是模拟 MySQL 主从复制协议。你想想看，MySQL 主从同步是怎么做的？主库把变更写进 Binlog，从库的 I/O 线程去拉，然后 SQL 线程回放。Canal 把自己伪装成一个从库，只拉日志，不回放。

具体流程是这样的：

Canal 向 MySQL 主库注册为一个 Slave
主库把 Binlog 推送给 Canal
Canal 解析 Binlog 的二进制格式，提取出事件（Event）
将事件转换成 Entry 对象（包含库名、表名、行数据、操作类型等）
通过 Canal 的客户端 API 或 MQ 投递给下游

关键点：Canal 只拉取和解析，不执行 SQL。所以它对主库几乎没有性能影响。我在生产环境压测过，单实例 Canal 可以支撑每秒 2-3 万条 Binlog 事件的解析，CPU 消耗大概在 20% 左右。

这里有个坑，我曾经踩过——Canal 依赖 MySQL 的 Binlog 格式必须是 ROW 模式。如果你用的是 STATEMENT 或 MIXED，Canal 也能工作，但拿不到完整的前后镜像数据。做数据同步，没有 before/after 字段，很多场景就玩不转了。

注意：MySQL 必须开启 Binlog，且格式设置为 ROW。检查命令：

SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

如果 binlog_format 不是 ROW，需要修改 my.cnf 并重启 MySQL。

3.2 Canal 架构

Canal 的架构其实不复杂，但分层很清晰。我习惯把它分成三层来看：

层级	组件	职责
接入层	Canal Server	管理连接、维护 Instance、提供网络服务
解析层	EventParser	拉取 Binlog、解析事件、生成 Entry
存储层	EventSink / MemoryStore	缓存解析后的 Entry，支持批量消费
投递层	Canal Client / MQ Producer	将 Entry 推送给下游消费者

每个 Canal Server 可以运行多个 Instance。一个 Instance 对应一个 MySQL 数据源。你想想看，如果你有 10 个数据库需要同步，就可以在同一个 Canal Server 上启动 10 个 Instance，每个 Instance 独立拉取、独立解析、独立投递。

我个人觉得，Instance 的设计是 Canal 最巧妙的地方。它让多源合并变得非常自然——每个源一个 Instance，各自维护自己的 Binlog 位点，互不干扰。

3.3 Canal 部署与配置

部署 Canal 其实不复杂。我一般分三步走：环境准备、配置修改、启动验证。

3.3.1 环境准备

Canal 依赖 Java 8 以上运行环境。另外，MySQL 那边需要给 Canal 创建一个专用账号，权限要求不高，只需要 REPLICATION SLAVE 和 REPLICATION CLIENT。

-- 创建 Canal 专用账号
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal_pass';
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

小技巧：我习惯把 Canal 的账号权限收窄到具体库，比如 GRANT ... ON mydb.*。虽然 Canal 只需要复制权限，但最小权限原则总是没错的。

3.3.2 下载与配置

从 GitHub 下载 Canal 的部署包，解压后主要关注两个配置文件：

conf/canal.properties：Canal Server 的全局配置
conf/example/instance.properties：单个 Instance 的配置

全局配置里，我通常只改端口和 MQ 相关参数。核心配置都在 Instance 层面。

# instance.properties 核心配置
# MySQL 连接信息
canal.instance.master.address=192.168.1.100:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal_pass

# Binlog 位点配置（首次启动可以不配，Canal 自动从最新位点开始）
canal.instance.master.journal.name=
canal.instance.master.position=

# 解析过滤规则
canal.instance.filter.regex=.*\\..*
canal.instance.filter.black.regex=mysql\\.slave_.*

这里有个细节，canal.instance.filter.regex 的默认值是 .*\\..*，表示监听所有库所有表。如果你只想监听某个库，可以改成 mydb\\..*。我曾经在生产环境遇到过因为过滤规则写错，导致 Canal 把系统库的 Binlog 也拉进来了，白白浪费带宽和 CPU。

3.3.3 启动与验证

启动 Canal 很简单：

# 进入 Canal 目录
cd canal.deployer-1.1.6

# 启动
./bin/startup.sh

# 查看日志
tail -f logs/canal/canal.log
tail -f logs/example/example.log

启动成功后，你会看到类似这样的日志：

2024-01-15 10:23:45.123 [main] INFO  c.a.o.c.c.CanalLauncher - ## start the canal server.
2024-01-15 10:23:46.456 [main] INFO  c.a.o.c.c.CanalController - ## start the canal server[172.16.1.10:11111]
2024-01-15 10:23:47.789 [destination = example , address = 192.168.1.100:3306] INFO  c.a.o.c.i.core.AbstractCanalInstance - start successful

看到 start successful 就说明 Canal 已经连上 MySQL 并开始拉取 Binlog 了。

常见问题：如果启动报错 java.io.IOException: caching_sha2_password，说明 MySQL 8.0 的默认认证插件不兼容。解决方案有两种：

修改 MySQL 用户认证方式：ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'canal_pass';
在 Canal 的启动参数中增加 -Dcanal.instance.mysql.auth.mode=native

3.4 客户端接入示例

Canal 提供了 Java 客户端，使用起来非常直观。我贴一段最简代码：

// Canal Java 客户端示例
CanalConnector connector = CanalConnectors.newSingleConnector(
    new InetSocketAddress("127.0.0.1", 11111),
    "example",  // Instance 名称
    "",         // 用户名（空表示不鉴权）
    ""          // 密码
);

connector.connect();
connector.subscribe(".*\\..*");

while (true) {
    Message message = connector.getWithoutAck(100); // 批量拉取
    long batchId = message.getId();
    if (batchId == -1 || message.getEntries().isEmpty()) {
        Thread.sleep(100);
        continue;
    }
    for (Entry entry : message.getEntries()) {
        if (entry.getEntryType() == EntryType.ROWDATA) {
            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
            // 处理每一行变更
            for (RowData rowData : rowChange.getRowDatasList()) {
                System.out.println("操作类型: " + rowChange.getEventType());
                System.out.println("变更前: " + rowData.getBeforeColumnsList());
                System.out.println("变更后: " + rowData.getAfterColumnsList());
            }
        }
    }
    connector.ack(batchId); // 确认消费
}

这段代码的核心逻辑就是：拉取 → 解析 → 处理 → 确认。注意 ack 这一步不能省，否则 Canal 会认为客户端还没消费完，不会继续推送新数据。

经验之谈：我在生产环境里，一般不会直接用 Canal 的 Java Client 直连消费。更常见的做法是把 Canal 的数据投递到 Kafka 或 RocketMQ，然后由下游的消费者去拉取。这样解耦更彻底，也方便做多路分发。后面章节我们会详细讲 MQ 集成。

3.5 本章小结

Canal 的核心就三件事：伪装成 MySQL 从库拉 Binlog、解析成 Entry、投递给下游。架构上通过 Instance 实现了多源隔离，部署配置也不复杂。嗯，到这里你应该已经能跑通一个最基本的 Canal 实例了。

记住几个关键点：Binlog 必须是 ROW 模式、MySQL 账号要有复制权限、Instance 配置里的过滤规则要写对。这些坑我都踩过，你注意避开就好。