接手这个需求挺偶然的系统里有一张核心业务表每次更新之后下游的统计、检索、缓存都需要尽快知道。最开始用的是定时任务每30秒扫一次结果业务量一上来延迟和数据库压力都让人头大。后来决定换成“监听数据库binlog日志变化将变动实时发送到Kafka”这套链路从MySQL的binlog里读变更通过Canal解析后推给Kafka下游按需消费。整个方案落地下来效果很明显也踩了不少意料之外的坑。这篇就把我完整的实操过程拆开讲一遍从环境配置到消息消费再到线上问题排查适合刚接触实时数据同步、想直接参考复现的同学。1. 方案选型与底层原理为什么是“伪从库”而不是轮询1.1 先搞清楚binlog到底能提供什么binlog是MySQL的二进制日志记录了所有能够引起数据变化的操作包括INSERT、UPDATE、DELETE还有DDL语句。MySQL主从复制就是靠它工作的主库生成binlog从库拉取并回放最终达到数据一致。既然主从复制能做到实时那一个外部程序模拟成从库去订阅binlog自然也就能做到对数据变更的实时感知。这里要重点提醒一点binlog有三种格式Statement、Row和Mixed。Statement记录的是SQL语句本身比如“UPDATE student SET age 18 WHERE id 1”这种格式同步到从库时如果语句里有NOW()这类非确定性函数两边结果可能不一致。Row格式记录的是每一行数据实际被改成什么样包含了变更前后的完整字段值。Canal这类工具要求使用Row格式否则拿不到具体的字段变化解析出来的消息会缺少关键信息。所以第一步就是在MySQL上把binlog格式设置成ROW而且建议把binlog_row_image设置成FULL。如果不设置成FULL默认在UPDATE时可能只记录被修改的字段和主键其他字段不会出现在binlog里。对下游做数据同步来说缺字段就意味着可能拿不到完整的行数据后患无穷。有同学可能会问为什么非要用binlog我自己写个轮询去查“更新时间大于上次查询时间”不行吗能行但有几个现实问题一是高频轮询对数据库有额外压力二是这种方式只能发现被修改的行删掉的行很难补齐三是无法感知表结构变化。binlog是数据库天生就生成好的日志读取它对业务无侵入对主库性能影响也极小这才是做实时变更捕获的正道。1.2 市面上的方案不少我为什么选了Canal当时我在技术选型上列了几个候选自研解析binlog的程序、开源的Maxwell、Debezium、以及Canal。自研解析这个想法看着很酷但我很快就放弃了。binlog的格式复杂跨版本还有差异加上位点管理、断点续传、表结构缓存全部自己实现一遍至少是几周的工作量而且很容易在细节上翻车。Maxwell轻量部署简单也能把变更输出到Kafka、Kinesis这类消息队列。但它的定位是简单同步场景对分库分表、多实例管理、位点运维界面这些支持得比较弱。Debezium是Red Hat开源的项目功能很强基于Kafka Connect生态丰富但运维依赖组件较多配置起来更重适合已经有Kafka Connect集群的团队。最终我用了Canal核心原因有三点一是它专门为MySQL binlog解析而生在阿里巴巴内部经过大规模生产验证对MySQL各种版本和特殊场景的处理很成熟二是它对位点的管理做得很好会把消费位点持久化到本地文件重启之后能续传三是它和Kafka的对接几乎是开箱即用的不需要自己额外写太多胶水代码。对于“监听binlog然后发到Kafka”这个需求来说Canal就是那条最短的路径。1.3 整条链路的实际工作流程数据链路大致是这样的MySQL产生binlog变更Canal把自己伪装成MySQL的一个从库向MySQL发送dump协议请求MySQL认为它是一个合法的slave于是源源不断推送binlog事件。Canal接收并解析这些事件处理成结构化的JSON消息然后写入Kafka的指定Topic。下游的消费者只需要订阅Kafka拿到消息后更新自己的索引、统计或缓存。Canal伪装从库这个点我需要强调一下理解了它就理解了大半原理。MySQL的slave连接的时候要提供server-idCanal就给自己分配一个独立的server-id向主库注册成从库。主库会把binlog流推给它这样Canal不需要去主动轮询文件而是被动的接收推送实时性自然就高。同时Canal内部维护了消费到的Binlog文件名和偏移量即使进程崩溃或重启也能从上次的位置继续读这是可靠性最关键的一环。2. MySQL端准备把binlog打开把账号权限给对2.1 binlog相关参数配置必须到位在动手部署Canal之前先把MySQL这边的配置处理好。我用的是MySQL 8.0大版本差异不大但有些细节需要额外注意。打开MySQL的配置文件一般路径是/etc/my.cnf或/etc/mysql/my.cnf设置以下几项[mysqld] server_id 112 log_bin /data/mysql/binlog/mysql-bin binlog_format ROW binlog_row_image FULL expire_logs_days 7 max_binlog_size 512M每一项都解释一下。server_id必须是和集群内其他节点都不重复的数字它用于标识一个MySQL实例。Canal连接过来模拟从库时如果它自带的slaveId和MySQL的server_id重复会被MySQL判为冲突直接断开。log_bin指定binlog存放路径和文件前缀务必保证这个目录存在并且MySQL进程有写权限。binlog_format ROW前面说过了是硬性要求。binlog_row_image FULL保证行数据完整。expire_logs_days控制binlog保留天数这个值很关键后面讲坑的时候会再提到建议线上保留至少7天防止Canal短暂的停机恢复时位点对应的binlog文件已经被清理。注意修改完参数后要重启MySQL。确认binlog是否开启执行SHOW VARIABLES LIKE log_bin;如果值是ON就说明已经生效。还可以执行一条SQL快速查看当前binlog状态SHOW MASTER STATUS;会返回一个类似mysql-bin.000014的文件名和Position偏移量。这个值在初始化Canal实例时有时候会用到比如你想让Canal从某个固定位置开始解析。2.2 创建采集账号权限要最小化Canal需要连接MySQL但是不能用root业务账号我单独创建了一个专用的采集账号。需要的权限只有三个SELECT、REPLICATION SLAVE、REPLICATION CLIENT。SELECT权限用于Canal拉取表结构元数据REPLICATION SLAVE和REPLICATION CLIENT则是它伪装成从库、读取binlog所需的核心权限。CREATE USER canal% IDENTIFIED BY canal_123456; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;这里有一个MySQL 8.0专属的坑默认的认证插件是caching_sha2_password早期版本的Canal客户端驱动可能不支持连接的时候会报认证失败。遇到这种情况有两个解决办法。一是在Canal连接串里指定SSL相关配置或支持该插件的驱动二是在MySQL端把这个账号的认证插件退化到mysql_native_passwordALTER USER canal% IDENTIFIED WITH mysql_native_password BY canal_123456;我个人倾向于用第二种方案简单直接兼容性最好。当然如果是在云数据库上控制台都有系统账号选项选择相应的授权方式即可操作思路是一致的。3. Canal部署从下载到Kafka收到第一条消息3.1 安装Canal并理解目录结构Canal的部署方式有单机版和集群版。我这次用的是单机版一台4C8G的云主机就够跑日常流量系统的峰值QPS在几百到一千左右Kafka和MySQL不在同一台机器上。直接去Canal的GitHub发布页下载对应版本的压缩包解压后目录结构大概是这样的canal.deployer-1.1.5/ ├── bin/ │ ├── startup.sh │ └── stop.sh ├── conf/ │ ├── canal.properties │ ├── example/ │ │ └── instance.properties │ └── logback.xml └── logs/ └── example/ └── example.logcanal.properties是Canal服务端的全局配置对应的是整个Canal进程包括集群信息、Kafka地址等。conf/example/instance.properties是具体的采集实例配置example是默认实例名表示监听一个MySQL实例的数据变更。如果你有多个MySQL实例要监听可以复制出多个同名目录各自改里面的配置互不干扰。日志目录在logs/下排障第一眼就是看这里的日志。提示解压后建议先确认Java环境Canal依赖Java 8以上。安装前执行java -version别让环境问题耽误后面的排查时间。3.2 全局配置和实例配置怎么改先改全局配置canal.properties主要关注两块一个是Canal服务端自身另一个是Kafka对接。我这里截取最核心的一段# 服务端端口默认11111用于Canal Adapter等客户端连接 canal.port 11111 # 指定Canal输出的消息队列类型 canal.serverMode kafka canal.mq.servers 192.168.10.20:9092 # 扁平化消息开关推荐打开 canal.mq.flatMessage true # 投递Kafka的批次大小 canal.mq.batchSize 2048这里canal.serverMode从默认的tcp改成kafka后Canal会把解析后的binlog事件直接往Kafka写不再需要额外的客户端去连接它拿数据。canal.mq.servers就是Kafka的broker地址多个地址用逗号分开。canal.mq.flatMessage这个参数我要重点说它决定消息是“扁平结构”还是“嵌套结构”。扁平结构里每一条数据就是一行字段直接平铺嵌套结构里data是一个数组。打开扁平消息后下游解析更简单我后面会给出具体消息样例。再改实例配置conf/example/instance.properties# 模拟从库的slaveId不能与MySQL server_id或其他slave冲突 canal.instance.mysql.slaveId 1234 # MySQL主库地址和端口 canal.instance.master.address 192.168.10.10:3306 # 连接MySQL的账号 canal.instance.dbUsername canal canal.instance.dbPassword canal_123456 # 连接MySQL时使用的字符集 canal.instance.connectionCharset UTF-8 # 监听哪些库和表这里写的是正则 canal.instance.filter.regex school\\..* # 输出到Kafka的Topic canal.mq.topic school-binlog # 分区数量 canal.mq.partitionsNum 3 # 分区哈希策略按主键哈希保证同一行数据有序 canal.mq.partitionHash school:student:pk正则表达式这里容易写错在配置文件里点号需要双斜杠转义。school\\..*表示监听school库下的所有表。如果需要监听多个库、多张表用逗号分隔比如db1\\.table1,db2\\..*。没有匹配到的库表Canal会直接忽略。关于Topic命名我建议在Topic上加上业务前缀比如school-binlog而不要用默认的实例名。这样Kafka管理页面上一眼能看出这个Topic的用途下游消费方交接起来也清晰。3.3 启动Canal并验证Kafka收到消息配置完成之后先确认Kafka的Topic已经创建好了。如果Kafka允许自动创建TopicCanal投递时也会自动创建但生产环境我会手动创建因为可以提前设置好分区数、副本数。用Kafka自带的命令创建kafka-topics.sh --create \ --topic school-binlog \ --partitions 3 \ --replication-factor 2 \ --bootstrap-server 192.168.10.20:9092然后启动Canalsh bin/startup.sh启动后立刻去看日志注意观察是否出现以下关键信息tail -f logs/example/example.log日志里重点找start mysql event parse - success、canal is start...以及类似find start position: BinlogPosition[mysql-bin.000014:123456]的记录。如果有这些说明Canal已经成功连上MySQL并且找到了开始解析的位点。接着去Kafka消费端验证消息。用控制台消费者直接看kafka-console-consumer.sh \ --topic school-binlog \ --from-beginning \ --bootstrap-server 192.168.10.20:9092然后在MySQL表里随便执行一条INSERT或UPDATE消费者的终端里应该立刻出现一条JSON消息。看到消息的那一刻那条“监听binlog到Kafka”的核心链路就算正式通了。如果终端里一直没反应按我的经验大概率不是Canal没跑起来而是没有新变更触发因为Canal默认从启动时的最新位点开始解析历史binlog不会主动补发。4. Kafka消费端怎么接消息格式、分区规则与幂等处理4.1 flatMessage打开之后消息到底长什么样Kafka消息格式这块很多同学会踩坑因为你如果不看控制台直接拿默认的canal.mq.flatMessage false去解析拿到的结构会比想象中复杂。我强烈建议在生产环境打开flatMessage true这样一条变更直接按一行记录平铺输出。我们看一个实际例子school库下有一张student表执行了一条INSERT语句INSERT INTO student (id, name, age) VALUES (1, 张三, 17);flatMessage开启后Kafka里收到的消息是{ id: 203, database: school, table: student, pkNames: [id], type: INSERT, es: 1710000000000, ts: 1710000001000, sql: , data: { id: 1, name: 张三, age: 17 } }如果是一条UPDATE它还会多一个old字段存放更新前的旧值。比如{ id: 204, database: school, table: student, pkNames: [id], type: UPDATE, es: 1710000002000, ts: 1710000003000, sql: , data: { id: 1, name: 李四, age: 18 }, old: { name: 张三, age: 17 } }es字段是binlog里记录的事件时间ts是Canal处理完并投递消息的时间这两个字段不能混淆。下游如果要做指标统计应该以es为准因为它更接近真实业务发生时间ts受解析和网络延迟影响只能作为参考。sql字段在Row格式的DML语句下通常是空的只有DDL事件里才有内容。如果不开flatMessagedata和old就变成了数组一个事务里变更了多行就会对应多条数据。下游如果只按单条消息处理就会有点别扭。所以除了有特殊的多行解析需求我建议无脑打开flatMessage能省掉很多不必要的复杂度。4.2 消费端必须处理的三个问题消息接到了不等于就完事了。Kafka消费端有几个问题必须在设计之初就想清楚不然上线之后一定会出乱子。第一个问题是重复消费。Canal在投递Kafka和持久化位点之间不是原子操作如果Canal在消息已经写入Kafka、但位点还没持久化时崩溃重启后它会从上个位点重新读取binlog于是同一批消息被再次投递到Kafka。Kafka自身的消费语义默认也是至少一次所以下游收到重复消息几乎是必然的。解决办法是下游做幂等比如写入缓存时用主键作为Redis Key用值覆盖写入数据库时用唯一键做防重或者记录一个消息ID去重表。第二个问题是顺序性。Canal的partitionHash配置很关键它决定了哪些数据落到同一分区。同一个分区内消息是有序的跨分区则不保证。所以对于同一张表需要按主键或者业务唯一键来做哈希保证同一条记录的多次变更始终进入同一分区这样才能保证下游处理时“先UPDATE后DELETE”的顺序正确。如果你不关心顺序那随便哈希也没有问题但大多数同步场景都是关心顺序的这里别偷懒。第三个问题是DDL事件。当下游执行了ALTER TABLE这类DDLCanal也会产生一条type为ALTER/DROP/CREATE的消息这种消息里data字段是空的sql字段会被填充。如果你的下游解析逻辑直接按data里的字段去映射遇到DDL事件会报空指针。我处理的办法是在消费端先判断事件类型如果是DDL先把消息记到日志里或者触发一次全量刷新不进入常规的字段处理逻辑。4.3 位点管理与数据不丢失的兜底思路Canal的位点信息默认存在conf/example/meta.dat文件里内容是纯文本记录当前的binlog文件名和Position。它保证了Canal在正常情况下重启之后可以从上次消费到的位置继续解析。这里有一个很大的隐藏风险如果meta.dat丢失或者被删除Canal重启后会从MySQL当前最新的binlog位点开始拉取等于自动放弃了丢失位点到当前之间的所有变更下游会静默地丢数据。所以运维上一定要保证meta.dat所在目录的持久化不要放在临时目录里并且做好定期备份。如果真出现丢位点的情况可以手工指定起始位点在instance.properties里配置canal.instance.master.journal.name mysql-bin.000014 canal.instance.master.position 123456然后重启Canal。这个操作的原理就是告诉Canal“从哪个文件哪个偏移量开始解析”相当于手工恢复断点。但前提是你必须知道准确的位点否则跳过或重复几条数据都正常。5. 上线之后最常踩的坑问题清单与排查思路5.1 常见错误速查表我把这次线上遇到的问题连同一个同事那边踩过的典型问题整理了一张速查表按“症状、原因、处理方式”排列排查时对照着看会快很多。问题现象常见原因处理方式启动报Communications link failureMySQL 8.0默认认证插件不兼容账号改用mysql_native_password插件或升级Canal连接驱动日志提示slaveId is already usedCanal的slaveId与集群中其他节点冲突给每个实例分配唯一的canal.instance.mysql.slaveId一直收不到Kafka消息没有新binlog事件位点配置错误在MySQL执行一条INSERT触发变更检查filter.regex是否匹配日志报Could not find first log file name in binary log indexbinlog被清理位点对应的文件不存在调大expire_logs_days必要时重置位点从最新开始收到消息后发现字段缺失binlog_row_image未设置成FULL修改MySQL配置为FULL并重启消费端解析报NPE遇到DDL事件data字段为空代码里先判空记录DDL事件日志同一条记录的消息乱序到达Kafka分区哈希策略配置不合理设置canal.mq.partitionHash按主键哈希这张表基本覆盖了从环境搭建到线上运行的常见故障。如果你的问题是这几类之外的我还有一个通用排查思路先看logs/example/example.log有没有异常堆栈再看Kafka消费组有没有积压再把Canal的位点信息和MySQL的SHOW MASTER STATUS做对比三步基本能把问题范围锁定。5.2 三个高发问题的排查手记第一个高发问题是“Canal连上了MySQL但Kafka没有任何消息”。我当时排查了很久最后发现问题出在filter.regex上。实例配置里我写的是school\\.student这种精确匹配但当时实际开发者是先建库后建表表名带了一个环境前缀导致正则没匹配上。Canal不会因为正则匹配不到任何表而报错它只是静默地过滤掉所有事件。排查办法就是写一个宽一点的正则比如school\\..*看能不能收到消息能收到说明是过滤规则的问题再逐步精确。另外一个常见原因是位点定位到了很久之前的binlog而那个binlog恰好已经被清理了这时候日志里会有明显报错按上面表格里的处理方法重置位点即可。第二个高发问题是“消息延迟越来越大”。Canal刚刚上线时延迟基本在毫秒级但高峰期发现延迟突破了秒级。我看了下系统监控MySQL的写入压力并不大Kafka这边也没有积压后来定位到是Canal解析端的batchSize配置太小一次拿到的binlog事件太少在高频写入场景下处理不过来。把canal.instance.batchSize从默认值调大同时调整canal.mq.batchSize到2048之后延迟立刻降了下来。如果调整后发现Kafka侧出现少量消息变大导致的网络问题还需要关注Kafka的max.message.bytes配置确保单条消息大小没有超过broker限制。第三个高发问题是数据库主从切换之后同步断了。主库发生切换后原来的binlog文件序列和server_id都可能变化Canal按照旧位点去拉数据会报错。这种情况不能指望Canal自动感知运维上要做两点一是在切换流程里把Canal停掉等新主库稳定后再重启让Canal重新定位二是将Canal的位点初始化逻辑改成从新主库当前位点开始宁可丢一小段数据也不能让进程反复重启卡死。这里要结合实际业务容忍度来决定如果容忍度低需要做更复杂的补偿机制但至少Canal进程要保证可用。5.3 几件小事不处理就是大麻烦最后再聊几个容易被忽视的小细节。binlog保留时长是其中最重要的一个。很多团队默认配置是保持一两天但Canal一旦停机超过这个时间位点失效恢复之后只能从最新开始中间这段时间的变更全部丢失。上线前务必跟DBA确认binlog保留时间至少是Canal允许的最长停机时间加上一定的安全余量。Kafka的Topic副本数也不能忽略。单副本Topic在broker宕机时会丢消息严重时候整个Topic不可用。用kafka-topics.sh --describe检查一下副本状态至少保证ReplicationFactor大于1。这个跟“监听binlog发Kafka”主链路的目的直接相关我们做实时同步不就是为了不下游丢数据所以Kafka自身的可靠性也得配到位。另外Canal所在机器的磁盘空间也要盯住。如果binlog量很大Canal的解析速度和Kafka的消费速度不匹配Canal本地会产生一些临时文件或者额外的日志输出虽然不会像binlog那样无限增长但日志磁盘满了会让进程直接崩溃。我给Canal挂了一台独立的监控磁盘使用率超过80%就告警目前线上一直很平稳。这些看起来都是小事但每一条都是实际把链路跑稳的关键。对于”监听数据库binlog日志变化将变动实时发送到Kafka”这个需求来说真正的难点不在打通链路而在于打通之后的可靠性细节把这个细节打磨完这套方案才算真正落地。