先亮个底这篇文章要聊的是我实际搭过的一条数据同步链路——把 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_binMySQL 就会把INSERT、UPDATE、DELETE、CREATE TABLE、ALTER TABLE这类操作按顺序写进日志文件。从实验角度可以把 binlog 理解成数据库自己写的一份“操作录音带”。它记录了“某时刻执行了什么SQL、影响了哪些行、变更前后的数据长什么样”。只要录音带还在你就能顺着它回放整个数据库的变更历史。binlog 有三种格式选型直接影响后面能不能做实时同步具体区别如下格式记录内容适合场景对 CDC 的意义Statement记录原始 SQL传统主从复制无法可靠还原行级变更不考虑Row记录每一行的变更前后值主从复制、CDC数据精准、字段明确推荐Mixed根据 SQL 自动选择混合场景不确定性强不建议 CDC 采用所以做监听的时候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 订阅的开源方案有好几个不能只会用一个至少要知道他们之间的差异。CanalJava 开发专盯 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配合acksall才能尽量避免消息丢失。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 falsecanal.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 1234master.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 建好之后启动 Canalbin/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, 测试用户, testexample.com, 1);消费者端会立刻收到一条类似这样的消息{ data: [ { id: 10001, name: 测试用户, email: testexample.com, status: 1 } ], old: null, database: demo, table: t_user, type: INSERT, ts: 1710000000000, sql: , pkNames: [id] }执行 UPDATEUPDATE t_user SET status 2 WHERE id 10001;收到的消息里data是更新后的新值old是更新前的旧值type变成UPDATE{ data: [ { id: 10001, name: 测试用户, email: testexample.com, status: 2 } ], old: [ { status: 1 } ], database: demo, table: t_user, type: UPDATE, ts: 1710000001000, sql: , pkNames: [id] }执行 DELETEDELETE FROM t_user WHERE id 10001;此时data仍是被删行数据但type是DELETEold为null。下游可以根据type字段区分业务语义。如果你想少传点冗余字段可以把canal.mq.flatMessage设为true消息会从嵌套结构变成扁平结构比如data里的字段直接摆到最外层。这种格式体积小、容易解析但会丢一些元数据。我的经验是宁可消息大一点也要保留old字段和完整data因为后面审计、回滚、对比都靠它。4.3 消息字段说明和常见理解误区很多第一次接触这条链路的人会把data和old的含义搞混。记住一句话data是这一行当前的最新镜像old是变更前的旧值。UPDATE时old里只出现被修改的字段不是整行旧值。比如上面只改了statusold自然只有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 的acksall配合min.insync.replicas2能保证 broker 收到消息并持久化副本后才确认给生产者。如果只设acks0那 Canal 发出去就算成功丢消息是大概率事件。这个参数必须在生产集群上设置不能省。第三段是 Kafka 到下游消费者。消费者从 Kafka 拉取消息后如果先提交 offset 再处理业务处理失败时就会丢消息如果先处理再提交又会重复消费。这是所有消费端都绕不开的经典问题。我的处理方式是业务消费逻辑必须幂等。Kafka 不保证不重复只保证不丢在合理配置下。所谓幂等就是相同的消息消费两次业务结果一致。比如更新操作直接用消息里的id做主键 upsertINSERT用INSERT ... ON DUPLICATE KEY UPDATE。这样即便重复消费也只是多执行一次无副作用操作。5.5 DDL 变更对下游的冲击binlog 不只记录 DML还记录 DDL。Canal 默认会解析并发送 DDL 事件甚至你在filter.regex里订阅的表执行了ALTER TABLEKafka 里会收到一条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、幂等消费、动态字段解析这三件事做好再谈高并发和精确一次顺序不能反。