先亮个底：这篇文章要聊的是我实际搭过的一条数据同步链路——把 MySQL 的 binlog 当作数据源，持续监听表里的每一次增删改，然后把变更记录实时送进 Kafka。核心目的就一个，让下游系统不要再靠“定时扫表”这种笨办法拿增量数据，把数据延迟从分钟级直接压到秒级以内。

这套东西听起来像是个中间件项目，其实本质上就是“变更数据捕获”，英文简称 CDC。如果你维护过几个对数据实时性有要求的系统，比如缓存同步、数据仓库增量更新、搜索索引刷新、消息通知触发，大概率都遇到过一个尴尬：上游业务库改了数据，下游只能在凌晨跑批或者每 5 分钟轮询一次，怎么都绕不开延迟。

而监听 binlog 走 Kafka 这种方式，等于把数据库的每一次变化当作一条事件消息广播出去。谁需要这份数据，谁就去 Kafka 里按自己的节奏消费，业务方之间完全解耦。这篇文章会把这条链路的原理、选型、配置、验证、排坑完整讲一遍，适合正在折腾数据同步、又不想只停留在“听说过 canal”阶段的同学。

1. 为什么监听 binlog，而不是定时扫表

1.1 定时扫表到底输在哪里

先看大家最熟悉的增量同步方案：给每张表加一个update_time字段，然后写个任务每 N 秒扫一次where update_time > ?。看起来简单，但踩过坑的都懂。

第一是延迟没法低。哪怕把轮询周期压到 10 秒，依然有 10 秒的窗口拿不到最新数据。第二是给业务库增加额外压力，尤其大表扫索引、回表查数据，高频轮询对生产库不友好。第三最致命：delete操作你感知不到。你只能靠“数据少了”反推，根源数据早就没了。还有update到旧值也没有，下游需要审计或缓存旧值时无从下手。

这些问题不是通过“把轮询调快一点”能解决的。真正合适的方式是：让数据库告诉你“我变了”，而不是你去问“你变了吗”。

1.2 binlog 到底记录了什么

binlog 是 MySQL 的二进制日志，专门记录数据库的结构变更和数据变更。只要开启了log_bin，MySQL 就会把INSERT、UPDATE、DELETE、CREATE TABLE、ALTER TABLE这类操作按顺序写进日志文件。

从实验角度，可以把 binlog 理解成数据库自己写的一份“操作录音带”。它记录了“某时刻执行了什么SQL、影响了哪些行、变更前后的数据长什么样”。只要录音带还在，你就能顺着它回放整个数据库的变更历史。

binlog 有三种格式，选型直接影响后面能不能做实时同步，具体区别如下：

所以做监听的时候，MySQL 侧必须把binlog_format设为ROW。否则你会拿不到“这一行改成什么样了”的关键信息。

binlog_row_image这个参数也值得关注。默认FULL表示记录所有字段的旧值和新值；如果设成MINIMAL，只记录被修改的字段，虽然日志体积变小，但下游拿不到完整字段镜像，很多场景下不够用。生产环境做实时同步，建议保持FULL。

1.3 这条链路的一句话模型

整套架构可以简化成一句话：MySQL 把 binlog 里的变更记录“漏”给一个中间件，中间件再把它包装成消息推到 Kafka。

中间件做的事，其实和 MySQL 从库有点像。从库要同步主库数据，也是连上主库之后请求 binlog，然后在本地回放。我们常用的方案就是“假装成从库”去订阅 binlog，拿到之后再转发出去。

放到生活场景里理解：你订阅了一个报刊的发行记录，只要新一期印刷出来，发行站就会把信息推送给你；你再把它抄送到微信群里，订阅你群的人就都知道了。中间的那几层，各自只关心自己负责的那一段。

这个模型的好处是：业务库不需要做改动，Kafka 里的消息可以被多个下游同时消费，不同团队按自己的节奏处理，不再相互牵制。

2. 工具选型：主流的 binlog 订阅方案怎么选

2.1 常见方案各有各的脾气

社区里做 MySQL binlog 订阅的开源方案有好几个，不能只会用一个，至少要知道他们之间的差异。

• Canal：Java 开发，专盯 MySQL。它提供 TCP 客户端协议，也内置了 Kafka/RocketMQ 的适配器，配置上可以做到“binlog 直接进 MQ”。中文资料多，国内团队用得多。
• Maxwell：轻量级方案，以 JSON 格式输出变更事件，默认设计就是发往 Kafka 等消息队列。部署最简单，但定制性不如 Canal 灵活。
• Debezium：基于 Kafka Connect 生态，支持 MySQL、PostgreSQL 等多种数据库，适合本身已经搭了 Kafka Connect 的团队，扩展性最强，但组件更多、也需要接触 Connect 框架。

选型没有绝对最优，主要看团队技术栈和系统现状。只聊 MySQL、并且希望把 MQ 发送能力内置好的场景，Canal 确实更顺手。

2.2 我为什么用 canal 加 Kafka 这条链路

最直接的原因是：Canal 原生支持 MQ 模式，配置serverMode = kafka之后，它会把解析出来的 binlog 事件直接发到指定 topic，不需要我再写一层 Java 客户端消费再推送。

另外它的消息格式对下游很友好。可以输出带data、old、type、database、table等字段的结构化 JSON，下游拿到消息就能直接判断这是一条INSERT还是UPDATE，不用自己去解析 binlog 原始字节。对 Java 技术栈的团队来说，部署和排查都更容易。

还有一个隐藏优势：Canal 内部有位点管理机制。它消费 binlog 会记录 position，中间件重启后可以接着上次的位置继续消费，这比“每次全量重扫”要稳得多。

2.3 关键的设计取舍

很多第一次接触这个方案的人会问：下游系统为什么不直接连 Canal，非要中间塞一个 Kafka？

我在实际项目中强制要求走 Kafka，基于两个原因。第一是解耦。Canal 作为一个独立中间件，不适合被十几个业务方直接连上来，那样连接数、重启影响、延时要求都会被放大。第二是削峰。大数据量变更时常见瞬间写入高峰，Kafka 天然提供缓冲，下游按自己的吞吐能力消费，不会把订阅方全部打垮。

所以 Canal 负责“采集和解析”，Kafka 负责“传递和堆积”，下游负责“按需处理”。各司其职，链路才不至于被某一方的抖动拖死。

3. 环境准备：MySQL、Kafka、Canal 三端的核心配置

3.1 MySQL 侧：开 binlog 不等于就能用

MySQL 是整条链路的源头，源头没配好，后面全白搭。我一般在my.cnf的[mysqld]区域维护以下配置：

[mysqld] server_id = 100 log_bin = mysql-bin binlog_format = ROW binlog_row_image = FULL expire_logs_days = 7

解释一下几个点。server_id在集群内必须唯一，它标识每个节点。log_bin设置日志文件前缀。binlog_format = ROW是行级日志，这是 Canal 解析的关键。expire_logs_days = 7只保留 7 天日志，如果下游故障时间超过这个窗口，旧日志会被清理掉，导致位点丢失。

改完配置必须重启 MySQL 才生效。这里有个坑：MySQL 默认可能已经开了 binlog，但格式还是STATEMENT，你SHOW VARIABLES LIKE 'binlog_format';看一眼就知道实际生效值。

接着创建增量同步专用账号。Canal 需要模拟从库拉取日志，所以账号至少要两个复制权限：

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

用最小权限原则，实际日常数据读写权限不需要给这个账号。别图省事直接给 root，后面出了安全事故不好交代。

3.2 Kafka 端的 Topic 规划与配置建议

Kafka 这边看起来只是建个 topic，但参数设置会直接影响后续消息顺序和可靠性。

我习惯先做一次数据量估算：假设业务表高峰期每秒变更 1000 行，每行消息体按 1KB 算，高峰期就有 1MB/s 的写入流量。副本和分区规划要能撑住 2 到 3 倍的余量。

topic 的关键参数建议：

• 如果下游对消息顺序有强要求，一开始就建单分区。单分区最简单，天然保证顺序，吞吐不足时拆多个 topic 或考虑对顺序不敏感的场景。
• 对顺序不太敏感的场景，分区数可以按消费者并行度估算，通常 6 到 12 个分区够用。
• 副本数至少 2，配合acks=all才能尽量避免消息丢失。
• cleanup.policy建议delete，保留时间按下游重放需求设定，我常用 3 天。

不是必须一步到位，但 topic 分区数一旦确定，以后想扩容就比较麻烦，所以前期想清楚比后期补强更重要。

3.3 Canal 的部署和两个核心配置文件

Canal 安装不复杂，下载发行包后主要要改两个配置：conf/canal.properties和conf/example/instance.properties。

canal.properties是进程级的全局配置。我需要关注的是运行模式和 MQ 接入地址：

canal.serverMode = kafka canal.mq.servers = 127.0.0.1:9092 canal.mq.topic = binlog_topic canal.mq.flatMessage = false

canal.serverMode = kafka是整条链路的开关，没有它 Canal 只提供 TCP 模式，消息不会进 Kafka。canal.mq.servers填 Kafka 的 broker 地址，多个 broker 用逗号分隔。canal.mq.flatMessage = false表示输出完整 JSON 结构，后面会看到它长什么样。

instance.properties是数据源实例级配置。一个 Canal 进程可以配置多个 instance，每个 instance 对应一套 MySQL 连接配置和订阅规则。最核心的是这一段：

canal.instance.master.address = 127.0.0.1:3306 canal.instance.dbUsername = canal canal.instance.dbPassword = canal_pass canal.instance.connectionCharset = utf8mb4 canal.instance.filter.regex = demo\\.t_user,demo\\.t_order canal.instance.mysql.slaveId = 1234

master.address是 MySQL 的主库地址和端口。filter.regex是订阅规则，注意这里要用双反斜杠分隔库名和表名，多个表用逗号分隔。slaveId必须和 MySQL 的server_id不冲突，它相当于 Canal 没话找话当了一个假从库的编号。

配置里我看到很多人踩过一个坑：只改了canal.properties里的 topic，没改instance.properties里的canal.mq.topic。实际上实例优先读自己配置文件的 topic，两边不一致就会把数据发到两个不同位置。建议只在 instance 文件里维护 topic，全局文件保留默认值。

4. 从启动到验证：让一条 INSERT 5 秒内出现在 Kafka 里

4.1 启动顺序和启动方式

环境配置好之后，启动顺序有个基本原则：先把下游备好，再起上游采集。

我的顺序是先确认 Kafka 已经启动并且 topic 存在，再启动 Canal。Kafka 如果用的是传统模式，还得先启动 Zookeeper。现在新版已经支持 KRaft 模式，简化了很多。

启动 Kafka 之前先把 topic 建好：

bin/kafka-topics.sh --create --bootstrap-server 127.0.0.1:9092 \ --topic binlog_topic --partitions 1 --replication-factor 1

单机演示用副本 1 没问题，生产环境副本至少 2。topic 建好之后，启动 Canal：

bin/startup.sh

然后观察日志：

tail -f logs/example/example.log

如果看到类似start successful或者是已经连接到 MySQL 开始发送位点的信息，链路基本就通了。如果日志里出现connect failed，别急着往下走，先回到 MySQL 网络和权限排查。

4.2 用消费者命令实测三类 DML 消息

链路通了没通，一定要实测。我习惯先起一个控制台消费者挂在那里，然后去 MySQL 里做增删改，眼看到消息实时刷出来才算数。

控制台消费命令：

bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 \ --topic binlog_topic --from-beginning

接着在 MySQL 侧执行：

INSERT INTO t_user (id, name, email, status) VALUES (10001, '测试用户', 'test@example.com', 1);

消费者端会立刻收到一条类似这样的消息：

{ "data": [ { "id": 10001, "name": "测试用户", "email": "test@example.com", "status": 1 } ], "old": null, "database": "demo", "table": "t_user", "type": "INSERT", "ts": 1710000000000, "sql": "", "pkNames": ["id"] }

执行 UPDATE：

UPDATE t_user SET status = 2 WHERE id = 10001;

收到的消息里data是更新后的新值，old是更新前的旧值，type变成UPDATE：

{ "data": [ { "id": 10001, "name": "测试用户", "email": "test@example.com", "status": 2 } ], "old": [ { "status": 1 } ], "database": "demo", "table": "t_user", "type": "UPDATE", "ts": 1710000001000, "sql": "", "pkNames": ["id"] }

执行 DELETE：

DELETE FROM t_user WHERE id = 10001;

此时data仍是被删行数据，但type是DELETE，old为null。下游可以根据type字段区分业务语义。

如果你想少传点冗余字段，可以把canal.mq.flatMessage设为true，消息会从嵌套结构变成扁平结构，比如data里的字段直接摆到最外层。这种格式体积小、容易解析，但会丢一些元数据。我的经验是：宁可消息大一点，也要保留old字段和完整data，因为后面审计、回滚、对比都靠它。

4.3 消息字段说明和常见理解误区

很多第一次接触这条链路的人会把data和old的含义搞混。记住一句话：data是这一行当前的最新镜像，old是变更前的旧值。

UPDATE时old里只出现被修改的字段，不是整行旧值。比如上面只改了status，old自然只有status。如果你下游需要整行旧值做对比，得设置binlog_row_image = FULL，并确认 Canal 打的是完整镜像，但old依然默认只记录变化字段。

ts字段表示事件时间戳。这个时间更接近 Canal 处理日志的时间，而不是 MySQL 里发起事务的时间。做延迟统计时可以对比 MySQL 业务字段里的时间和ts的差值，但别把它当严格的事务发生时间。

还有个容易忽略的点：Canal 只发送已提交事务的 binlog 变更。如果 MySQL 里的事务执行了 INSERT 却迟迟不 COMMIT，消息不会出现在 Kafka 里。这点符合直观预期，但也意味着 Kafka 里的消息顺序是“提交顺序”，不是“执行顺序”。

5. 生产环境常见问题排查：照着案例排雷

5.1 启动时连不上 MySQL，十有八九是这三个原因

Canal 启动日志里出现connect failed是最常见的故障，我总结下来的原因主要在三个方向。

第一个是网络不通。Canal 部署机和 MySQL 不在同一个网段，或者中间防火墙拦了 3306 端口。用telnet 127.0.0.1 3306在部署机上手动测一下，秒级就能判断。

第二个是账号权限不足。账号至少要REPLICATION SLAVE和REPLICATION CLIENT。只给SELECT权限，Canal 能连上但拉取 binlog 时会报权限错误。我排查时直接执行SHOW GRANTS FOR 'canal'@'%';，一目了然。

第三个是slaveId冲突。Canal 的canal.instance.mysql.slaveId和 MySQL 的server_id必须不同。曾经有人在已有从库的集群里把 Canal 配成了和从库一样的 ID，导致从库复制中断，排查了半天才意识到。建议 Canal 的slaveId用一个独立网段内的特殊值，比如 123456，避开常见的 1 到 1000。

5.2 只同步了部分变更：binlog 格式没生效

场景非常典型：Kafka 里能收到消息，但收到的数据结构不对，或者某些表收了某些表没收。

先检查 MySQL 的 binlog 格式：

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

如果发现binlog_format是STATEMENT，即使 Canal 能跑，消息内容也识别不了行级变更。必须改成ROW并重启 MySQL。

如果是某些表收了某些表没收，重点看filter.regex正则。Canal 的过滤规则是按库名\\表名写的，冒号、逗号、反斜杠的写法容易出错。写完正则建议先测一下模式，别只在 Canal 日志里猜。过滤器只影响订阅范围，不影响那些已经订阅上的表的消息格式。

5.3 Kafka 里消息顺序错乱

有业务方反馈消费到的同一张表 ID 顺序不对，明明先 INSERT 变成 UPDATE，下游却先收到了 UPDATE。

这个问题的根源基本都在分区。Kafka 只有在单个分区内才有顺序保证。如果 topic 配了多个分区，Canal 默认按生产者策略分发数据，同一条记录的不同事件可能被分到不同分区，消费顺序自然无法保证。

解决办法就两个方向：要么对顺序敏感的表走单分区 topic；要么显式设计分区键，让同一行数据的消息都进同一分区。按主键哈希分区的方案能兼顾并行度和顺序，但要注意主键变更时，消息会落到不同分区，顺序依然无法完全保证。

我对强顺序场景的一贯建议是：不要相信分布式系统里的“几乎有序”。要做到强顺序，分区键必须设计得无懈可击，否则就把分区数设为 1，简单可靠，别为难自己。

5.4 消息会不会丢：可靠性模型

一条数据变更从 MySQL 到 Kafka，涉及三段可靠性边界。

第一段是 MySQL 到 Canal。Canal 通过模拟从库拉取 binlog，它会定期记录 position。拉取成功并且发送成功后，position 才会推进。如果 Canal 进程中途挂了，重启后会从上次最后成功的位点继续，所以这段基本不会丢，前提是 MySQL 的 binlog 保留时间足够长。

第二段是 Canal 到 Kafka。这里取决于 Kafka 端配置。Kafka 的acks=all配合min.insync.replicas=2，能保证 broker 收到消息并持久化副本后才确认给生产者。如果只设acks=0，那 Canal 发出去就算成功，丢消息是大概率事件。这个参数必须在生产集群上设置，不能省。

第三段是 Kafka 到下游消费者。消费者从 Kafka 拉取消息后，如果先提交 offset 再处理业务，处理失败时就会丢消息；如果先处理再提交，又会重复消费。这是所有消费端都绕不开的经典问题。

我的处理方式是：业务消费逻辑必须幂等。Kafka 不保证不重复，只保证不丢（在合理配置下）。所谓幂等，就是相同的消息消费两次，业务结果一致。比如更新操作直接用消息里的id做主键 upsert，INSERT用INSERT ... ON DUPLICATE KEY UPDATE。这样即便重复消费，也只是多执行一次无副作用操作。

5.5 DDL 变更对下游的冲击

binlog 不只记录 DML，还记录 DDL。Canal 默认会解析并发送 DDL 事件，甚至你在filter.regex里订阅的表执行了ALTER TABLE，Kafka 里会收到一条type为ALTER的消息。

这对下游是好事，也是坏事。好的一面是下游能感知表结构变化；坏的一面是，如果下游消费者的解析逻辑写死了字段列表，老消息还能消费，但新消息多了一列，消费端直接报错。

我的应对方式是：消费者解析 JSON 时不要写死“必须有哪些字段”，尽量动态取字段名；对未知字段跳过而非抛异常。同时在 DDL 发生时通过旁路触发一次表结构变更通知，让下游执行结构刷新。Canal 的tsdb功能也能帮助记录历史表结构，但对消费端来说，结构兼容设计更重要。

6. 落地过程中的几个扩展建议

6.1 只订阅关注的表，不要全库扫

刚开始做验证的时候，很多人图省事直接订阅全部表。等到生产环境数据量一大，全库 binlog 会把 Kafka topic 撑爆，而且下游还要过滤掉不关心的表，资源浪费很严重。

filter.regex应当按白名单配置。比如只关心订单相关的两张表：

canal.instance.filter.regex = shop\\.t_order,shop\\.t_order_item

如果库名或表名里本身包含正则特殊字符，记得转义。这个规则也支持模式匹配，比如shop\\..*表示订阅shop库下所有表。能用白名单解决的问题，别用黑名单。

6.2 多业务库和多实例部署

一套 MySQL 实例里有多个业务库，如果你希望不同业务线的数据隔离，可以在 Canal 里配置多个 instance。每个 instance 指向同一个 MySQL 地址，但filter.regex不同，输出 topic 也可以不同。

这样做的收益是：某个订阅方消费慢了，不会拖垮所有 instance。Canal 进程本身也可以按 instance 重启，互不干扰。

不过多个 instance 都是模拟从库，会各自占用一个复制连接。MySQL 的连接数和客户端资源要留出余量。instance 数量不多时没问题，上百个就有点过度设计了，这种规模通常应该考虑直接对接消息总线。

6.3 下游消费者的几条经验

如果你用 Java 写 Kafka 消费者，group.id一定要按业务场景区分。同一个 topic 可以有多个消费组，它们各拉各的互不影响。比如一套消费组刷缓存，另一套同步数仓，天然并行。

消费者线程数和分区数的关系要匹配。Kafka 一个分区在同一个消费组内只能被一个消费者线程持有，所以分区数决定并发上限。消费者线程数如果大于分区数，多余线程只能空转。想要提升并发，前期 topic 分区数就得有规划。

最后提醒一个容易被忽略的点：Canal 发出的消息没有显式的事务边界。一个业务事务里改了订单表和订单明细表，会生成多条独立消息，每条消息在 Kafka 里可能是分开的。如果下游要求事务级一致性，比如要么两张表都更新成功，要么都不更新，不能依赖单条消息完成，得在消费端做消息聚合或者接受最终一致。

我个人在项目里做了两年的 binlog 到 Kafka 的同步链路，最大的感受是：这个方案的难点根本不在“把消息发到 Kafka”，而在于下游如何理解这些消息、如何面对重复消费和乱序、如何跟上结构变化。Canal 和 Kafka 把你从“轮询数据”里解放出来，接下来要做的，是认真设计消息模型和消费逻辑。建议你先把单分区 topic、幂等消费、动态字段解析这三件事做好，再谈高并发和精确一次，顺序不能反。