4、Canal核心原理：Canal架构、Binlog解析原理、Canal与Kafka集成、Canal HA方案
说到CDC，Canal是绕不开的一个组件。我个人习惯把它叫做「MySQL的Binlog搬运工」。它干的事情其实很纯粹——把自己伪装成一个MySQL Slave，然后等着Master把Binlog推过来，再解析成我们能用的数据格式。
这一章，我会把Canal的底裤扒干净。从架构设计到Binlog解析，再到怎么跟Kafka搭伙过日子，最后聊聊高可用怎么搞。都是我在生产环境里踩过的坑和攒下的经验。
4.1 Canal整体架构
Canal的架构其实不复杂，但设计得很巧妙。你想想看，它要模拟MySQL主从复制，又要做数据解析和投递，这中间涉及的角色还真不少。
核心角色一览：
- Canal Server：整个系统的调度中心，负责管理Instance的生命周期
- Canal Instance：真正干活的单元，一个Instance对应一个MySQL数据源
- EventParser：伪装成Slave，跟MySQL Master建立连接，拉取Binlog
- EventSink：把解析后的Event进行过滤、转换、路由
- EventStore：内存存储，暂存解析好的Event，支持多消费者
- MetaManager：记录消费位点，保证重启后能断点续传
下面这张图是我自己画的Canal内部流转图，你看一眼就能明白数据是怎么从MySQL流到下游的。
这里有个细节我想强调一下——EventStore用的是RingBuffer，也就是环形缓冲区。为什么不用队列？因为环形缓冲区没有锁竞争，写入和读取可以并行，性能极高。我在压测环境里见过单Instance跑到10万+ TPS，RingBuffer功不可没。
4.2 Binlog解析原理
Binlog解析是Canal最核心的能力。说白了，就是把MySQL的二进制日志翻译成我们能看懂的结构化数据。
MySQL的Binlog有三种格式：
| 格式 | 特点 | Canal支持 | 我的建议 |
|---|---|---|---|
| STATEMENT | 记录原始SQL语句 | 不推荐 | 别用，主从不一致风险高 |
| ROW | 记录每一行数据的变化 | 完美支持 | 生产环境首选 |
| MIXED | 混合模式，自动切换 | 部分支持 | 能用，但不如ROW稳定 |
我个人强烈建议你用ROW格式。为什么？因为Canal解析ROW格式的Binlog，可以直接拿到「变更前」和「变更后」的整行数据。这在做数据同步时太方便了——你不需要再去MySQL查一遍，直接拿解析结果往目标库写就行。
Binlog的解析流程大致是这样的：
- 连接握手：Canal向MySQL发送COM_REGISTER_SLAVE命令，注册自己为Slave
- 请求Binlog：发送COM_BINLOG_DUMP指令，从指定位置开始拉取
- 接收事件：MySQL把Binlog Event一条条推过来
- 事件解析：Canal根据Event Type（比如WRITE_ROWS、UPDATE_ROWS、DELETE_ROWS）做反序列化
- 生成Entry：解析结果封装成Canal的Entry对象，包含Header和Entry内容
避坑指南：
我曾经遇到过一个坑——MySQL的Binlog默认是压缩的（binlog_row_image=MINIMAL），只记录被修改的列。如果你需要完整的「前镜像」和「后镜像」，记得把binlog_row_image设成FULL。不然你拿到的数据可能缺字段，下游消费端会报空指针。
解析后的Entry长什么样？我给你看个简化版的结构：
Entry {
header: {
logfileName: "mysql-bin.000123", // Binlog文件名
logfileOffset: 456789, // 偏移量
executeTime: 1700000000, // 执行时间戳
eventType: "UPDATE_ROWS" // 事件类型
},
entry: {
tableName: "orders",
databaseName: "trade_db",
rowDataList: [
{
beforeColumns: [
{ name: "id", value: "1001" },
{ name: "status", value: "PAID" },
{ name: "amount", value: "299.00" }
],
afterColumns: [
{ name: "id", value: "1001" },
{ name: "status", value: "SHIPPED" },
{ name: "amount", value: "299.00" }
]
}
]
}
}你看，beforeColumns和afterColumns一对比，变化一目了然。这就是ROW格式的魅力。
4.3 Canal与Kafka集成
在实际生产里，很少有人直接用Canal的TCP客户端去消费数据。为什么？因为TCP直连的耦合太强了——Canal挂了或者重启，消费端都得跟着处理连接状态。更常见的做法是把Canal的数据投递到Kafka，让Kafka来做削峰填谷和解耦。
Canal 1.1.5版本之后，内置了Kafka和RocketMQ的适配器。配置起来其实不复杂：
# canal.properties 核心配置
canal.serverMode = kafka
canal.mq.servers = 192.168.1.10:9092,192.168.1.11:9092
canal.mq.retries = 3
canal.mq.batchSize = 16384
canal.mq.maxRequestSize = 1048576
# instance.properties 针对某个实例的配置
canal.instance.mq.topic = canal_trade_orders
canal.mq.partitionHash = orders.id
canal.mq.dynamicTopic = true
canal.mq.flatMessage = true这里有几个参数我得跟你说道说道：
- flatMessage：设为true时，Canal会把Entry拍平成JSON字符串，Kafka消费端拿到的就是纯JSON，解析起来很方便。设为false时，投递的是protobuf序列化的二进制数据，性能更好但消费端得引入Canal的客户端依赖。
- partitionHash：指定用哪个字段做分区键。我建议用主键或者业务ID，保证同一行数据的变更事件落到同一个分区，这样消费端能保证顺序。
- dynamicTopic：如果设为true，Canal会根据库名和表名自动生成topic，比如
trade_db_orders。适合多表同步的场景。
注意：
Kafka的topic分区数一定要提前规划好。我见过有人把分区数设成1，结果Canal的写入吞吐上不去，背压导致EventStore的RingBuffer被写满，最后Canal直接OOM了。一般来说，分区数建议等于Canal Server的并发线程数，或者稍微多一点。
投递到Kafka的消息格式（flatMessage=true时）长这样：
{
"database": "trade_db",
"table": "orders",
"type": "UPDATE",
"ts": 1700000000000,
"data": [
{
"id": "1001",
"status": "SHIPPED",
"amount": "299.00"
}
],
"old": [
{
"status": "PAID"
}
]
}data字段是变更后的数据，old字段是变更前的数据（只包含有变化的列）。消费端拿到这个JSON，直接反序列化就能用，非常清爽。
4.4 Canal HA方案
单机Canal肯定不行，万一挂了数据就断了。Canal的HA方案依赖ZooKeeper来做选主和位点管理。
HA的核心思路是：多个Canal Server组成一个集群，但同一时刻只有一个Server在干活（Active），其他的处于Standby状态。Active挂了，ZK会感知到Session超时，然后触发重新选主。
配置HA其实就两步：
- 在canal.properties里开启HA模式：
canal.zkServers = 192.168.1.20:2181,192.168.1.21:2181 - 多个Canal Server使用相同的
canal.instance.destination名称
HA模式下，位点的存储会从本地文件迁移到ZK。这样Active挂了，Standby接管时能直接从ZK读取上次消费的位点，不会重复消费也不会丢数据。
HA切换流程：
- Active Canal在ZK上创建临时节点（比如
/otter/canal/destinations/orders/running） - Standby Canal监听这个节点的变化
- Active宕机，临时节点因Session超时被删除
- Standby收到通知，尝试创建自己的临时节点
- 创建成功的那个成为新的Active，开始拉取Binlog
- 切换时间一般在3-5秒内完成
不过这里有个细节要注意——MySQL那边也需要做HA。Canal连的是MySQL的VIP或者DNS域名，如果MySQL主从切换了，VIP会漂移到新的Master上。Canal检测到连接断开后，会重新连接，这时候连的就是新的Master了。
我的经验：
我曾经在生产环境遇到过一个问题——MySQL主从切换后，Canal虽然重连成功了，但Binlog的位点不对了。原因是新Master的Binlog文件名和偏移量跟旧Master不一样，Canal拿着旧位点去新Master上找，结果找不到对应的Binlog文件。
解决方案是：在Canal的MetaManager里配置一个位点回退策略。如果找不到指定位点，就回退到当前可用的最早Binlog位置，或者直接跳到最新的Binlog位置。虽然会丢一点点数据，但总比卡死强。
最后总结一下HA方案的优缺点：
| 维度 | 说明 |
|---|---|
| 优点 | 切换快（3-5秒）、位点不丢失、部署简单 |
| 缺点 | 依赖ZK、同一时刻只有单点干活、资源利用率50% |
| 适用场景 | 中小规模数据同步、对一致性要求高的场景 |
| 不适用场景 | 需要多活或水平扩展的超大规模场景（建议用Canal集群+MQ分区） |
嗯，Canal的核心原理差不多就这些。从架构到Binlog解析，再到Kafka集成和HA方案，每个环节都有不少细节。你在实际落地的时候，建议先从单机开始验证，确认Binlog解析和MQ投递都没问题，再上HA。别一上来就搞集群，出了问题排查起来很头疼。