做后端开发的同学，对 MySQL 主从复制应该都不陌生：主库负责写，从库负责读，靠 binlog 把数据变更同步过去，读写分离缓解主库压力。但如果你遇到的情况是“不止要同步到另一台 MySQL，而是要同步到 Elasticsearch、Redis、消息队列，或者异构数据库”，原生复制就不够用了。这时候把阿里巴巴开源的 canal 中间件拉出来，可能是整个技术栈里性价比最高的一种方案。

这次聊的项目标题是“基于阿里巴巴 canal 产品完成 MySQL 主从数据同步”，我结合自己实际部署、接入、排查的经验，把整个落地过程从头到尾拆开讲一遍：包含 canal 的选型理由、原理、部署配置、消费端写法，以及生产环境里踩过的坑。适合正在做数据同步、实时数仓、缓存一致性这类需求的开发同学参考，也适合刚接触 canal 的运维同学快速上手。

1. 项目背景与方案选型思路

1.1 为什么传统主从复制不够用

原生 MySQL 复制解决的是从一个 MySQL 实例到另一个 MySQL 实例的数据复制问题，复制的方式是：主库开启 binlog，从库拉取 binlog 并执行其中的事件，从而完成数据重现。这个机制本身非常可靠，但它有几个明显的边界。

第一，目标端必须是 MySQL，或者至少能兼容 MySQL 协议的存储。如果你想把数据同步到 Redis，或者同步到搜索集群，原生复制根本无法直接做到。

第二，复制粒度是整库或者整表级别的，无法方便地在同步过程中做字段过滤、数据转换、多表拼接。比如只想把订单表里的订单状态同步到缓存，原生复制做不到细粒度控制。

第三，从库只能被动跟随主库的 schema 变化，不能对同步内容做应用层加工。很多业务场景需要“读取 binlog → 解析 → 转换 → 投递到下游存储”这样的灵活管道，传统主从复制给不了。

我当时接到的需求非常典型：电商系统的订单表在主库不断写入，团队希望订单状态实时同步到 Elasticsearch，同时推送一份变更消息到消息队列里做后续处理。表结构还时不时会改，新增字段不能被漏掉。这种情况下，主从复制只能算第一步，真正要解决的是“把 binlog 当数据管道”的思路。

1.2 常见的同步方案对比

围绕这个需求，当时团队总共评估过四条路：

对比下来，canal 的优势比较明显：它通过伪装成 MySQL 从库的方式读取 binlog，对主库没有任何额外写入压力，也不需要业务代码配合改。只要 MySQL 开启 binlog，创建好一个有复制权限的账号，canal 就能像一台只读从库一样，把变更事件持续拉出来。

如果你不熟悉 canal，可以把它理解成一个“复制器”：它既不存储业务数据精确副本，也不负责最终写入目标端，它只负责把 MySQL 的 binlog 事件解析成结构化消息，然后投递给下游。下游怎么消费、怎么入库，完全由你自己控制。

1.3 canal 最终胜出的原因

最终选 canal 有三个决定性因素：

第一，实时性高。canal 通过监听 binlog 的方式，数据变更后毫秒级可见，这和定时任务不同，定时任务只能做到分钟级甚至小时级。

第二，接入成本低。MySQL 端只需要开启 binlog 并创建一个授权账号，canal 服务本身可以独立部署，不影响主库，也不影响原有主从复制链路。原有的 MySQL 主从复制还可以继续保留，canal 相当于“旁路”多挂了一个订阅者。

第三，生态成熟。canal 作为阿里巴巴开源的中间件，在内部大量系统使用过，社区方案丰富，客户端和 MQ 适配层都有现成实现。相比自己从零解析 binlog，省了很多底层工作量。

当然，它也有自己的适用边界。如果数据量特别大，或者业务方对同步链路要求强一致，那么 canal 这种“解析增量、投递下游”的模式只能保证最终一致，需要配合消息队列和重试机制来保证可靠性。这一点在选型时要提前想清楚，不能拿着它当强同步工具用。

2. canal 核心原理与工作流程

2.1 canal 到底在做什么

简单来说，canal 把自己伪装成一个 MySQL 从库，向主库发送 dump 协议请求，主库则把 binlog 推送给它。canal 拿到 binlog 后，解析成结构化的数据对象，再经过过滤、转换，最终投递给客户端或者消息队列。

这个过程中有几个关键点值得展开。

MySQL 的 binlog 有几种格式：STATEMENT、ROW、MIXED。canal 要正确感知数据变更前后的状态，必须使用 ROW 模式，因为 ROW 模式的 binlog 会记录每一行的变更前镜像和变更后镜像，解析出来的信息最完整。STATEMENT 模式记录的是 SQL 语句，无法准确拿到“改了哪一行”的数据，所以 canal 官方文档明确要求 binlog_format 设置为 ROW。

同时，binlog 必须开启完整镜像，也就是 binlog_row_image 设置为 FULL，确保每次更新事件带出完整行记录。如果设置成 MINIMAL，日志体积会小一点，但解析出来的字段不全，下游想拿到 UPDATE 前后的其他字段就会缺失。

2.2 三个核心组件：EventParser、EventSink、EventStore

canal 在工作时，内部主要包括三个核心组件。

EventParser 负责和 MySQL 主库通信，通过 dump 协议请求 binlog，并解析原始 binlog 字节流，将其转换为 canal 内部的事件对象。这个组件相当于“数据入口”，它维护着解析位点，崩溃恢复时靠位点继续消费。

EventSink 负责对解析出来的事件做过滤、路由、归并等处理。比如你配置了只同步某几张表，过滤逻辑就在这一层完成；多张表的变更事件会按事务分组，保证同一个事务内的数据变更被连续投递。

EventStore 负责暂存处理后的数据事件，内部使用环形队列实现，下游消费端从这里拉取数据。环形队列的设计使得数据写入和消费解耦，canal 服务端的积压能力由此而来。

这三个组件的配合逻辑可以这样理解：EventParser 负责“扒数据”，EventSink 负责“挑数据”，EventStore 负责“存数据”，而外部客户端则负责“取数据”。一套完整的 canal 服务端，就是围绕这条流水线运转的。

2.3 关键概念：position、GTID、destination

在 canal 的配置和运维中，有几个概念会反复出现在日志和配置项里：position、GTID、destination。

position 指 binlog 的解析位点，可以理解为“上次消费到哪儿了”。canal 会把消费位点记录在元数据存储中，当服务重启或者任务恢复时，可以接着上次的位置继续消费，避免重复投递或者丢数据。canal 的位点管理和消息队列的 offset 机制比较像，理解成“书签”就对了。

GTID 是 MySQL 5.6 之后引入的全局事务标识，每个事务有唯一编号。使用 GTID 模式时，canal 可以更精准地进行故障恢复，因为事务是全局可定位且幂等的。如果你的 MySQL 支持 GTID，建议开启；canal 对 GTID 的支持已经比较成熟。

destination 在 canal 术语里指的是一个同步通道，也可以理解为一个同步实例。一个 canal server 可以同时启动多个 destination，每个 destination 对应一组独立的配置，比如连接不同的 MySQL、监听不同的库表。这种多实例的设计，让一个 canal 部署点就能服务多条业务线。

3. 环境准备与部署实操

3.1 MySQL 开启 binlog 并准备账号

部署 canal 的第一步，是确认 MySQL 满足它的运行条件。我自己一般按下面几步操作。

先检查当前 MySQL 的 binlog 配置：

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

log_bin 必须是 ON，binlog_format 必须是 ROW，binlog_row_image 必须是 FULL。如果没开启，需要修改 MySQL 配置文件，一般是 my.cnf 或者 my.ini：

[mysqld] server-id=1 log-bin=mysql-bin binlog_format=ROW binlog_row_image=FULL expire_logs_days=7

修改完需要重启 MySQL 生效。注意检查 server-id，如果主从复制链路中已经有其他从库使用过某个 server-id，canal 伪装从库时要分配一个新的 server-id，不能和现有复制链路冲突。

然后创建 canal 专用账号。canal 需要主库上的复制权限，以及读取 binlog 的权限：

CREATE USER 'canal'@'%' IDENTIFIED BY 'canal_pass'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;

很多教程只给了 REPLICATION SLAVE 权限，但我实际使用中发现，如果后续需要 canal 拉取某些元数据信息，SELECT 权限也得有，否则启动时会遇到权限不足的报错。密码尽量使用强密码，canal 账号虽然只读，但权限范围覆盖整个实例，还是要收敛访问来源。

3.2 下载 canal server 并配置 canal.properties

canal 的部署方式有两种：一种是直接下载 release 包，部署 Java 服务；另一种是用 Docker 镜像。生产环境我推荐直接用 release 包部署，方便调 JVM 参数、挂日志盘、接入监控。

从官方仓库下载对应版本的压缩包后，解压目录结构大致如下：

canal.deployer-xxx/ ├── bin/ ├── conf/ │ ├── canal.properties │ └── example/ │ └── instance.properties ├── lib/ └── logs/

先看全局配置 conf/canal.properties。这里有几个参数我每次部署都会特别关注：

canal.port = 11111 canal.destinations = example

canal.port 是服务端提供给客户端的 TCP 端口，客户端连这个端口消费数据。canal.destinations 配置的是要启动的 destination 列表，多个 destination 用逗号分隔。如果新加了一个 destination，这一步要记得同步修改。

如果计划用 MQ 模式，canal.properties 里还需要配置 MQ 的地址、topic 等参数。MQ 模式的配置项比较多，一般建议先把 TCP 模式跑通，再切 MQ，不要一上来就两个模式混着配，排查问题会很难受。

3.3 配置 instance.properties 与数据源映射

每个 destination 都有自己的配置文件，默认示例放在 conf/example/instance.properties。这个文件是 canal 连接 MySQL 的关键，核心配置项如下：

# MySQL 主库地址 canal.instance.master.address=127.0.0.1:3306 # binlog 位点信息 canal.instance.master.journal.name= canal.instance.master.position= canal.instance.master.gtid= # 连接账号 canal.instance.dbUsername=canal canal.instance.dbPassword=canal_pass # 解析规则 canal.instance.filter.regex=.*\\..* # 表过滤 canal.instance.filter.black.regex=

canal.instance.master.address 填主库地址，建议使用内网地址，尽量避免走公网，因为 binlog 数据量可能很大，公网传输不但慢，还不安全。

canal.instance.master.journal.name 和 canal.instance.master.position 是启动位点。如果留空，canal 默认从当前时间开始的 binlog 位置继续消费；如果要追历史数据，需要指定具体的 binlog 文件名和偏移量。这个位点配置一旦写错，canal 启动后会一直报找不到位点的错误，排查起来很费时间。

表过滤规则 canal.instance.filter.regex 默认值是.*\\..*，表示监听所有库的所有表。如果你只需要同步部分表，可以写成这样：

canal.instance.filter.regex=test_db\\..*

注意这里过滤规则用的是 Java 正则表达式，库名和表名之间是\\.而不是普通的点。我第一次写的时候用了一个点，结果匹配不上，折腾了半天才反应过来是转义问题。

3.4 启动验证与日志排查

配置完成后，启动 canal：

bin/startup.sh

启动后先看日志文件 logs/canal/canal.log。排查启动问题时，常见的日志内容有这么几类。

第一类是“canal is start successful”，表示启动成功，但这个日志在 canal 初始化阶段出现，不一定代表 instance 已经连接上 MySQL。

第二类是“connect to 127.0.0.1:3306 failure”，这个很直白，一般是网络不通、账号密码错误、或者 MySQL 没开远程访问。

第三类是“dump address ... has to do next step”，这表示开始接收 binlog 了，此时实例大概率已经正常工作。

如果要确认实例是否正确消费，可以看 logs/example/example.log。正常工作时，日志里会周期性出现 dump 相关的信息，表示 canal 还在持续向 MySQL 请求 binlog。

我习惯再用一个简单方式验证：在 MySQL 主库里执行一条 INSERT，然后看 canal 客户端能否收到事件。如果客户端有输出，说明整条链路已经通了。

4. 同步消费端实现

4.1 消息路由选择：TCP 客户端还是 MQ

canal 提供两种下游消费路径：TCP 直连客户端和 MQ 异步投递。

TCP 模式是指业务系统直接通过 canal-client 连接 canal server，主动拉取数据。这种模式延迟最低，控制最灵活，适合下游就是自己团队系统的场景。缺点是如果多个业务方都要同一份数据，每个业务方都得独立连接 canal server，服务端连接数压力会比较大。

MQ 模式是指 canal 将解析后的数据投递到 Kafka 或者 RocketMQ，业务方从消息队列消费。这种模式的优势非常明显：解耦、削峰、支持多订阅，而且消息队列本身有持久化和消费位点管理，可靠性更高。如果你已经有消息队列基础设施，我更推荐直接用 MQ 模式，因为 canal 服务端不再需要承担客户端连接压力，运维上也省心很多。

如果业务量不大、同步链路只有一条，TCP 模式也可以，代码写起来相对简单，一个进程启动就能消费。下面我分别给一个简单的代码示例。

4.2 写一个可用的 canal 客户端

以 Java 为例，TCP 模式需要一个 canal-client 依赖。以 Maven 项目为例：

<dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.7</version> </dependency>

然后写一个简单的消费循环：

import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.Message; import java.net.InetSocketAddress; import java.util.List; public class CanalConsumer { public static void main(String[] args) { CanalConnector connector = CanalConnectors.newSingleConnector( new InetSocketAddress("127.0.0.1", 11111), "example", "", "" ); int batchSize = 100; try { connector.connect(); connector.subscribe("test_db\\..*"); connector.rollback(); while (true) { Message message = connector.getWithoutAck(batchSize); long batchId = message.getId(); if (batchId == -1 || message.getEntries().isEmpty()) { Thread.sleep(1000); continue; } for (CanalEntry.Entry entry : message.getEntries()) { if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) { continue; } CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); CanalEntry.EventType eventType = rowChange.getEventType(); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { if (eventType == CanalEntry.EventType.INSERT) { printColumns(rowData.getAfterColumnsList()); } else if (eventType == CanalEntry.EventType.UPDATE) { System.out.println("before:"); printColumns(rowData.getBeforeColumnsList()); System.out.println("after:"); printColumns(rowData.getAfterColumnsList()); } else if (eventType == CanalEntry.EventType.DELETE) { printColumns(rowData.getBeforeColumnsList()); } } } connector.ack(batchId); } } catch (Exception e) { e.printStackTrace(); } finally { connector.disconnect(); } } private static void printColumns(List<CanalEntry.Column> columns) { for (CanalEntry.Column column : columns) { System.out.println(column.getName() + ":" + column.getValue() + " updated=" + column.getUpdated()); } } }

这段代码有几个细节需要强调。

第一，subscribe 里的表匹配规则要和服务端的过滤规则一致，否则可能出现服务端过滤完了，客户端又过滤一遍的问题。

第二，getWithoutAck 后必须调用 ack，告诉 canal 这批数据已经处理完成；如果处理失败不要 ack，canal 会重新投递。这个语义和消息队列的消费确认机制非常像。

第三，update 事件要同时处理 before 和 after，这样才能拿到变更前后的数据，很多业务比如审计日志、缓存更新，都需要借助 before 列判断是否真的发生了变化。

如果直接用 MQ 模式，canal 会直接把数据序列化后投递到指定的 topic，消费端的逻辑就不需要 canal-client 依赖了，只依赖对应的消息队列客户端即可。canal 投递到 Kafka 的消息体格式里，也包含类似上面的 entry 结构，只是经过了 protobuf 编码。消费的时候要引入相关协议包来解析。

4.3 数据格式与类型映射处理

canal 解析出来的字段值，最终都会变成字符串。这一点在接数据的时候要特别小心。

比如 MySQL 的 BIGINT、INT、VARCHAR、DATETIME，在 canal 的 Column 对象里，value 都是 String 类型。下游如果要写入 Elasticsearch 或者做计算，需要自己完成类型转换。

MySQL 的 DATETIME 在 canal 里输出的格式是“yyyy-MM-dd HH:mm:ss”，而 TIMESTAMP 也是类似格式。如果下游是 Elasticsearch，通常需要把这个字符串转成日期类型，再映射到 ES 的 date 字段。

Decimal 类型尤其容易踩坑。canal 解析出的 Decimal 值可能是科学计数法形式，比如“1.23E+10”，直接入库会丢失精度。对金额这类字段，我建议下游统一使用字符串存储，或者使用 BigDecimal 进行精确解析。

另外一个常见的坑是 NULL 值。canal 的 Column 对象中，如果某个字段是 NULL，value 字段会是空字符串还是 null？不同版本行为不完全一样。我在代码里处理的时候，会先判断 column.getIsNull()，如果是 true，直接认定该字段为 null，不依赖 value 字段。

4.4 全量 + 增量同步的衔接思路

canal 本身做的是增量同步，也就是启动之后发生的变更。但很多时候，目标端原本是空的，需要先做一次全量数据初始化。

全量加增量的衔接，我常用的方式有两种。

第一种是“先全量，后增量”。先在目标端导入全量数据，然后启动 canal 消费增量。这种方式的难点在于全量导入期间产生的增量数据可能会漏掉，所以一般需要先建立 canal 位点，再开始全量，全量完成后从记录的位点继续消费增量。

第二种是“先增量，后补偿”。先启动 canal 消费增量，写入一个临时缓冲表或者消息队列，然后执行全量导入，全量完成后再把增量和全量做一次幂等合并。这种方式适合目标端有主键可以幂等覆盖的场景。

两种方式没有绝对的好坏，取决于目标存储的特性。如果目标端是 Elasticsearch，我通常采用第二种，因为 ES 按文档 ID 写入，天然幂等，增量消费可以先落一份，全量再覆盖一遍，最后再消费积压的增量。

这里要注意：canal 只提供增量能力，全量导入需要结合 DataX 这类工具完成。canal 解决“持续变更”，DataX 解决“初始同步”，两者配合才能形成完整方案。

5. 常见问题与排查技巧实录

5.1 高频问题与排查速查表

我在实际排查中遇到的典型问题，整理为一张速查表，方便你对照处理：

这张表是我实际运维中提炼出来的，每一条都真实遇到过。尤其是第一条，看起来简单，但排查起来经常浪费很多时间。

有一个容易忽略的点：如果 MySQL 的主从复制链路里已经存在多个从库，主库会为每个从库分配资源。canal 伪装成从库时，它也需要一个唯一的 server-id。如果你在 canal 的 instance.properties 里没配置 server-id，canal 会随机生成一个，但随机生成可能和其他从库冲突，最好手动指定一个固定值。

5.2 生产环境监控关注点

canal 服务端跑起来之后就进入长期运行阶段，此时需要关注几个监控指标。

第一个是 dump 延迟。canal 通过拉取 binlog 的位点和 MySQL 当前最新位点之间的差距，判断同步是否延迟。如果延迟持续拉大，说明下游消费速度跟不上，或者 canal 服务本身性能不够。建议对延迟指标设置告警阈值。

第二个是内存队列积压。EventStore 使用环形队列缓存数据事件，默认容量是有限的。如果下游消费慢，队列会持续处于高位，此时需要关注消费端性能。

第三个是 JVM 内存和 GC。canal 本身是 Java 应用，长时间运行后堆内存使用率和 GC 频率会直接影响同步稳定性。生产环境建议给 canal 分配足够的堆内存，并开启 GC 日志，便于排查。

第四个是位点保存。canal 默认使用本地文件保存位点信息，如果服务器宕机，可能有一定程度的位置丢失或重复消费。重要的业务场景建议将位点存储配置调整为数据库模式，增强可靠性。

5.3 踩坑心得与后续扩展

最后说几个只有实际跑过才会明白的细节。

第一个教训是关于表结构变更的。canal 对 DDL 事件的处理需要额外注意。如果你在 MySQL 里执行了 ALTER TABLE 增加字段，canal 默认会把 DDL 事件也推给客户端，但客户端的解析代码如果没处理 DDL，可能会直接报错。处理方式是在消费端判断 entry 类型，如果是 DDL 事件，根据业务需要选择忽略或者更新下游 schema。

第二个教训是批量操作。如果 MySQL 一次更新了很多行，canal 解析出来的事件里会包含同一个事务下的大量行数据。处理的时候要注意批量语义，尽量不要在每一条数据上做高延迟操作，可以把一批数据攒起来批量写入下游，能显著提升吞吐。

第三个教训是监控告警要及时。canal 服务异常时，很多时候不会直接导致业务可见故障，但会造成数据不一致。如果下游是缓存，缓存和数据库不一致会在告警延迟很久之后才暴露出来。所以 canal 本身的健康状态一定要纳入监控体系。

后续如果想扩展，可以考虑把 canal 接入数据同步平台，结合 DataX 做全量增量一体化的数据管道。也可以将 canal 的事件投递到消息队列后，通过流式计算框架做实时指标统计。这套链路本质上就是 CDC（Change Data Capture）架构的一个具体落地，而 canal 在其中扮演了关键的数据捕获角色。

我个人在实际运维中的体会是：canal 部署并不难，真正难的是消费端的稳定性和数据一致性保障。配置好 binlog、写好消费代码，只是第一步；后续的监控、位点管理、异常重试链路，才决定这套同步方案能不能在生产环境长期跑下去。希望这篇实践总结能帮你少走一些弯路。