📚 课程目录第4章 Canal实战

4. Canal实战：从MySQL到Kafka、ES，再到高可用部署

好，咱们进入Canal的实战环节。说实话，Canal是我在实时数据同步中用得最多的工具之一。它原理不复杂，但坑确实不少。今天我把几个核心场景串起来讲——连接MySQL、同步到Kafka、同步到Elasticsearch，最后再聊聊高可用怎么搭。

本章核心脉络：Canal伪装成MySQL从库 → 解析binlog → 投递到Kafka/ES → 多实例保证不丢数据。

4.1 Canal连接MySQL：先让Canal能“看到”binlog

这一步是基础。Canal本质上是一个MySQL的从库，它需要MySQL开启binlog，并且binlog格式必须是ROW模式。我见过不少团队因为binlog格式是STATEMENT或MIXED，导致Canal解析出来的数据对不上——尤其是update语句，没有前后镜像，根本没法用。

避坑提醒：MySQL的binlog_format一定要设为ROW，且binlog_row_image建议设为FULL。否则Canal拿不到完整的before/after数据，做数据对比时会很痛苦。

MySQL侧配置（my.cnf）：

[mysqld]
log-bin=mysql-bin
binlog_format=ROW
binlog_row_image=FULL
server_id=1

Canal侧配置（conf/canal.properties）：

canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset=UTF-8
canal.instance.tsdb.enable=true

我个人习惯在MySQL里单独创建一个Canal专用账号，只给SELECT, REPLICATION SLAVE, REPLICATION CLIENT权限。这样即使Canal被攻破，也拿不到写权限。

小技巧：启动Canal后，先看日志里有没有“success to connect to mysql”和“dump binlog success”这两行。缺任何一行都说明连接有问题。

4.2 Canal同步数据到Kafka：解耦与削峰

把Canal的数据直接推到Kafka，是生产环境最常用的模式。为什么？因为Kafka能扛住突发流量，下游消费端可以按需拉取，不会把Canal压垮。

我曾经遇到一个场景：业务大促期间，MySQL的写入量瞬间飙升到平时的10倍。如果Canal直接推给ES，ES的写入队列直接打满，丢数据是小事，ES集群都差点挂了。后来改成Canal → Kafka → ES，Kafka做缓冲，问题就解决了。

Canal投递到Kafka的配置（conf/canal.properties）：

canal.serverMode=kafka
kafka.bootstrap.servers=192.168.1.10:9092,192.168.1.11:9092
kafka.acks=all
kafka.compression.type=snappy
canal.mq.topic=canal-topic

这里有个关键参数——kafka.acks=all。我建议生产环境一定要设成all，保证消息不丢。虽然延迟会高一点点，但实时数据产品最怕的就是数据对不上。

参数	推荐值	说明
kafka.acks	all	所有副本确认后才算写入成功，保证不丢
kafka.compression.type	snappy	压缩比适中，CPU开销小，适合实时场景
canal.mq.partitionsNum	3~6	根据表数量调整，保证分区内有序
canal.mq.enableDynamicTopic	true	按表名自动创建topic，方便管理

启动Canal后，可以用Kafka自带的命令行工具验证数据是否到达：

kafka-console-consumer.sh --bootstrap-server 192.168.1.10:9092 \
  --topic canal-topic --from-beginning

如果看到类似这样的JSON输出，说明数据已经成功投递：

{
  "database": "test_db",
  "table": "orders",
  "type": "INSERT",
  "data": [{"id": 1001, "amount": 299.00, "status": "paid"}]
}

4.3 Canal同步数据到Elasticsearch：实时搜索的基石

把MySQL的数据实时同步到ES，是很多搜索、推荐系统的核心链路。Canal本身不直接写ES，它通过Kafka做中转，然后由消费端（比如Logstash或自研程序）写入ES。

我推荐用Canal Adapter来做这件事。它是Canal官方提供的适配器，支持直接写入ES，不需要自己写消费代码。

配置Canal Adapter（conf/es.yml）：

dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: orders_index
  _id: _id
  upsert: true
  sql: "select o.id as _id, o.order_no, o.amount, o.status, u.name as user_name
        from orders o
        left join users u on o.user_id = u.id"

这里有个坑——sql中的字段名必须和ES mapping中的字段名完全一致。我遇到过因为大小写不一致，导致数据写进去但查不到的情况。排查了半天，最后发现是ES里字段是userName，SQL里写的是user_name，Canal Adapter不会自动做驼峰转换。

经验之谈：Canal Adapter的sql支持多表关联，但关联表的数据变更不会触发同步。只有主表（from后面的第一张表）的binlog变更才会触发。所以如果你的业务需要关联表也触发同步，得自己写消费端。

验证ES数据是否同步成功：

curl -XGET 'http://localhost:9200/orders_index/_search?q=order_no:20240101001'

4.4 Canal的高可用部署：别让单点拖垮你

生产环境最怕什么？最怕Canal挂了没人知道，数据同步断了几个小时。高可用部署就是为了解决这个问题。

Canal的高可用依赖ZooKeeper。多个Canal实例组成一个集群，通过ZK选主，同一时刻只有一个实例在拉取binlog。如果主实例挂了，ZK会自动切换到备用实例。

部署架构：

至少2台Canal Server，配置完全一样
1个ZooKeeper集群（至少3节点，保证选举可用）
MySQL开启GTID模式（方便Canal断点续传）

Canal高可用配置（conf/canal.properties）：

canal.zkServers=192.168.1.20:2181,192.168.1.21:2181,192.168.1.22:2181
canal.instance.global.mode=cluster
canal.instance.global.lazy=false

每个Canal实例的conf/instance.properties中，canal.instance.mysql.slaveId必须不同。我习惯用IP的最后一段来区分，比如实例1用101，实例2用102。

重要提醒：Canal的高可用只保证“拉取binlog”这个环节不中断。如果ZK集群本身挂了，Canal也会失去协调能力。所以ZK集群的稳定性同样重要。

验证高可用是否生效：

启动两个Canal实例，查看日志，只有一个是primary状态
手动kill掉主实例，观察备用实例是否在几秒内变成primary
检查Kafka/ES中是否有数据中断的窗口

我曾经在生产环境做过一次演练：故意停掉主Canal，备用实例在3秒内接管，Kafka端的数据流只出现了不到5秒的延迟。嗯，这个结果是可以接受的。

个人建议：高可用部署后，一定要做一次“断网测试”。把主Canal的网络断开，看看ZK能不能正确切换。我见过不少团队配置好了高可用，但从来没验证过，结果真出问题时切换失败。

好了，Canal实战的核心内容就这些。从连接MySQL，到投递Kafka，再到写入ES，最后用高可用兜底——这条链路在实时数据产品中非常经典。你可以在自己的测试环境里搭一遍，踩踩坑，印象会更深刻。