1. 先搞清楚canal到底在解决什么问题如果说这两年数据同步领域哪个组件最值得花一个星期认真学一遍我的答案一定是canal。这名字你可能在简历上见过在招聘JD里见过甚至在同事的代码评审里被吐槽过但真要问你它到底是干嘛的很多人的回答其实是模糊的嗯就是同步MySQL数据的——差不多对但不够准确。我换个说法你就明白了canal做的是MySQL增量数据订阅也就是监听MySQL的binlog日志每当数据库里发生insert、update、delete操作canal就能像装了窃听器一样在毫秒级把这个变更事件推给你。注意关键词是增量——它不是全量复制不是每天导一次数据而是从你部署那一刻开始实时关注每一条新变化。那它解决的到底是什么问题我举个最常见的场景。你们系统里肯定有一张热点表比如商品信息、用户资料、配置数据业务方读得很频繁。为了扛住流量你引入了Redis做缓存查询走缓存写操作更新数据库。听起来没问题但更新数据库的同时同步缓存这一步写法稍微不严谨就会出现缓存和数据库不一致的脏数据。常见的做法有几种双写、定时任务全量刷新、消息队列异步通知。双写逻辑侵入业务代码定时任务有延迟且浪费资源消息队列异步通知倒是好但你需要手动把每一次数据变更封装成消息发出去代码量不小。canal的思路是我不碰你的业务代码我在数据库这一侧偷听binlog把变更事件统一抓出来再推给你的下游。无论你是想去同步缓存、同步到ES、同步到数据仓库还是同步到另一个MySQL都不需要改业务代码。这个解耦的价值用过的人都懂——只需要一次配置全公司的数据消费方都能受益。对小白来说canal最友好的一点是它自带一个客户端订阅模型你只要能写Java代码几行就能跑通一个Demo完全不需要一开始就理解它的存储原理。但如果你想用好它光是会跑肯定不够因为真正的坑全在细节里binlog格式没配对、位点丢失导致重复消费、和多个订阅端抢数据……这些问题我用后面几个章节慢慢展开。顺便说一句canal项目的名称来自canal这个英文单词中文意思是运河、水道寓意很直白数据从MySQL流向下游的通道。这个比喻特别贴切你越往后用越会觉得它真的就像一条管道把数据库的血液输送到各个需要的地方。2. 安装前的三个关键开关JDK、binlog和权限安装canal本身不难难的是准备工作。很多小白按照网上的教程一步步来结果客户端连不上、收不到消息回头一看全是配套环境没弄对。所以这一章我把安装前置条件拆成三个开关每个都要提前确认坏了哪一个都会让你后面抓狂。2.1 第一个开关JDK版本canal的server端是Java写的所以你机器上得有JDK。这里有个坑需要你避一下canal 1.1.x系列用JDK 8没问题但如果你用的是比较新的canal 1.1.7版本或者想跑源码调试建议直接用JDK 11。我遇到过用JDK 8跑新版本canal启动的时候报了一些奇怪的并发类加载错误看起来像是代码问题其实是编译目标版本太高导致的。装好之后确认一下版本java -version正常情况下你会看到类似openjdk version 1.8.0_***或openjdk version 11.0.***的输出。如果你的机器上存在多个JDK建议用export JAVA_HOME/path/to/jdk指定清楚避免canal的启动脚本拿到错误的版本。2.2 第二个开关MySQL必须开启binlogcanal的工作原理是伪装成MySQL的从节点向主库请求binlog日志。所以你的MySQL必须开启binlog这是整个链条的源头。修改MySQL配置文件Linux下一般是/etc/my.cnf或/etc/mysql/mysql.conf.d/mysqld.cnf[mysqld] log-binmysql-bin binlog-formatROW server-id1这三行缺一不可尤其第二行binlog-formatROW。我见过有人在测试环境用了默认的STATEMENT格式结果canal解析出来一堆SQL语句根本拿不到精确的行级变更数据。因为canal的核心输出是某行某个字段从旧值变成了新值而这种结构化信息只有ROW格式才完整保留。改完之后记得重启MySQLservice mysqld restart或者systemctl restart mysqld。然后进入MySQL命令行验证SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format;两个结果都符合预期这个开关才算真正打开。还有一个细节容易被忽略server-id不能和canal实例配置的server-id重复。我用一个真实的踩坑经历说明第一次部署的时候我随手给canal设置了server-id1和MySQL自己的server-id冲突结果canal连接主库时直接被拒日志里报ERROR: Got error 1236。这个问题排查了很久才定位到因为报错信息很不直观。所以提前规划好MySQL给一个server-idcanal给另一个比如101、102这种互不干扰。2.3 第三个开关MySQL授权账号canal需要用一个专门的MySQL账号连接主库这个账号要有REPLICATION SLAVE和REPLICATION CLIENT权限。很多人在这步直接用root账号倒也能跑通但从规范角度不推荐尤其到了生产环境最少权限原则是底线。创建账号的SQL如下CREATE USER canal% IDENTIFIED BY canal123; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;这里给%是允许任意主机连接实际生产建议限定为运行canal的那台机器IP安全第一。另外注意如果数据库里已有的表结构比较复杂canal在解析DDL事件时可能还需要访问表定义所以顺手把SELECT权限也加上。这三件事做完环境准备基本就绪。老实说这个过程十分钟都用不了但价值远超十分钟——它决定了你后面所有排查动作能不能有一个干净的基础。3. 一条数据从MySQL到客户端的完整路径很多教程上来就让你装canal、跑Demo等你跑通了还是一头雾水它到底是怎么拿到数据的这一章我从原理层面完整讲一遍大概需要你花五分钟耐心读读完之后你就拥有了排查问题的底层地图遇到任何诡异现象都能知道往哪个环节去查。3.1 先理解MySQL主从复制是怎么回事canal的核心机制其实是在模仿MySQL的从库。MySQL的主从复制过程很简单主库把每一次变更写入binlog从库连接到主库请求从某个binlog位点开始的数据拿到之后在本地重放。canal做的事情和从库一模一样只不过它拿到这些日志之后重放的不是SQL而是把日志解析成结构化的数据变更事件交给下游程序处理。这个过程涉及binlog里的几个概念我用大白话解释一下binlog文件MySQL落盘生成的日志文件比如mysql-bin.000001达到一定大小后自动切换下一个文件。binlog位点position一条日志在文件中的偏移量它是唯一的。客户端恢复消费时只要记住上次消费到的position就能从断点继续。GTID全局事务ID是比position更高级的定位方式不依赖具体文件名和偏移量。新版本canal推荐用GTID模式主从切换时能自动适应。3.2 canal内部的三段式流水线canal从拿到binlog到吐出数据内部经历了三个核心阶段第一阶段抓取Parser。canal伪装成从库向MySQL发送dump协议请求。MySQL就会从指定的binlog位点开始把日志一条条推送给canal。这个阶段最纯粹只负责拿。第二阶段解析Sink。拿到的binlog是事件流需要解析。比如一个UPDATE事件会解析出旧值before和新值after这个是canal最有价值的输出——下游可以直接拿到{id: 1, name: 旧值}变成{id: 1, name: 新值}这样的JSON数据而不需要自己拼SQL去反查。第三阶段存储Store。解析完成后事件会写入一个内存环形队列等待客户端来拉取。canal内置了一个基于内存的存储组件默认容量是16384个事件客户端消费速度跟不上时事件会在内存里排队排满了之后才会出现丢数据或者阻塞的情况。整个链条如下图所示文字版MySQL binlog → canal Server 抓取 → 解析 → 内存队列 → 客户端(Client) 拉取 → 你的业务代码/ES/Redis...3.3 你眼里的canal组件Server、Instance和Client实际部署时你会接触到几个概念这里一次性讲清楚避免后续被配置项绕晕canal Server一个独立运行的Java进程核心服务。canal InstanceServer内部的一个逻辑实例每个Instance对应一个MySQL数据源。一个Server可以配置多个Instance每个Instance有独立的配置、独立的订阅位点、独立的队列。canal Client你写的消费端程序通过TCP协议从Server拉取数据。打个比方Server是餐厅Instance是厨房里的一个灶台——每个灶台对应一个供应商MySQLClient是来吃饭的顾客。一个餐厅有多个灶台很正常每个灶台负责炒不同的菜互不干扰。从部署角度小白刚开始只需要一个Server、一个Instance、一个Client后面需要订阅多个库或者对接多套下游时再考虑多Instance。4. 跑通第一个Demo用Java代码订阅一条insert前面讲了那么多理论和准备这一章我们真正动手。我用一台Linux机器做演示canal部署包从官方GitHub Release页面下载最新稳定版即可解压后目录结构大致如下/canal ├── bin ├── conf │ ├── canal.properties │ ├── example │ │ └── instance.propertiesconf/canal.properties是全局配置conf/example是默认实例的配置目录instance.properties是实例级配置。4.1 修改实例配置先改instance.properties这是核心# MySQL主库地址 canal.instance.master.address127.0.0.1:3306 # 连接MySQL的账号密码 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal123 # 订阅的数据库和表*表示所有 canal.instance.filter.regextest\\..* # 位点定位方式 canal.instance.gtidfalse重点说明一下filter.regex这个配置。它的格式是数据库名\\..*比如我只想订阅test库的所有表就写test\\..*想订阅test库下的user表就写test\\.user想订阅多个库用|分隔。这个正则很容易写错因为Java字符串本身也要转义所以\\.写成\\.而不是\.。我见过有人照抄配置把转义符弄丢了结果canal直接不输出任何数据查了半天才发现是正则匹配问题。4.2 修改全局配置再看canal.properties重点确认以下几项canal.port11111 canal.destinationsexample canal.instance.global.modetcpcanal.port是客户端连接的TCP端口默认11111不用改。canal.destinations指定启动哪些实例默认example对应conf目录下的example文件夹。canal.instance.global.modetcp表示消费端通过TCP直连方式拉取数据——这是最简单的方式后面聊MQ模式时再说。4.3 启动Servercd /canal/bin ./startup.sh启动后看日志确认没有异常tail -f /canal/logs/canal/canal.log如果看到canal server is running now之类的字样并且没有报错说明Server起来了。再检查实例状态tail -f /canal/logs/example/example.log正常情况下能看到start successful。这里有个新手最容易遇到的情况日志里显示canal is starting然后卡住不动多半是MySQL连接没通去查example.log里有没有connect timeout或者Access denied字样。4.4 写一个最小客户端canal官方提供了一个客户端库canal.client用Maven引入dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.7/version /dependency然后写一个最简单的Java类核心逻辑是建立连接 → 拉取消息 → 解析事件 → 打印变更内容。import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; import java.net.InetSocketAddress; import java.util.List; public class CanalDemo { public static void main(String[] args) { // 连接canal server CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, // instance名称 , // 无认证时填空 ); connector.connect(); // 订阅所有表这里要和instance的filter.regex对应 connector.subscribe(test\\..*); while (true) { // 一次拉取100条不设置超时 Message message connector.getWithoutAck(100); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { // 没有数据睡1秒再试 try { Thread.sleep(1000); } catch (InterruptedException e) {} continue; } for (CanalEntry.Entry entry : message.getEntries()) { if (entry.getEntryType() ! 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) { System.out.println(INSERT: rowData.getAfterColumnsList()); } else if (eventType CanalEntry.EventType.UPDATE) { System.out.println(UPDATE before: rowData.getBeforeColumnsList()); System.out.println(UPDATE after: rowData.getAfterColumnsList()); } else if (eventType CanalEntry.EventType.DELETE) { System.out.println(DELETE: rowData.getBeforeColumnsList()); } } } // 确认消费成功 connector.ack(batchId); } } }这段代码看着有点长但核心就三件事连接、拉取、确认。所有canal客户端的骨架都是这个模式你以后无论用Spring Boot封装还是写其他语言的客户端都逃不开这个套路。4.5 实测验证运行上面的Java类然后在MySQL里执行INSERT INTO test.user (name, age) VALUES (张三, 25);立刻就能在控制台看到类似下面的输出INSERT: [name: 张三, age: 25]看到这一行你的canal入门第一个Demo就算真正跑通了。从这一刻起任何数据库变更都会实时出现在你的消费端面前剩下的就是你怎么利用这些数据了。5. 小白最容易踩的五个坑含完整排查思路说实话canal的部署和Demo真的很简单但越是这种看起来简单的组件越容易在细节上翻车。这一章我把自己和周围人踩过的坑整理出来每一项都附上排查链路而不是直接给结论——因为下次你遇到的可能只是个相似但不完全一样的问题思路比答案更重要。5.1 坑一binlog格式不是ROW解析结果全是SQL现象客户端能连上能收到消息但解析出来的RowData是空的只有SQL语句。排查链路第一步回到MySQL执行SHOW VARIABLES LIKE binlog_format;如果结果是STATEMENT或MIXED基本就是问题所在。第二步修改配置为ROW重启MySQL。第三步重新执行一条INSERT看canal是否输出结构化的列值。根因只有ROW格式才会记录每一行变更的具体列值STATEMENT格式记录的是原始SQL文本canal无法从中还原出行级别数据。5.2 坑二server-id冲突导致连接被拒现象canal日志里出现ERROR: Got error 1236或者直接连接不上MySQL。排查链路第一步检查MySQL的server-idSELECT server_id;第二步检查canal的canal.instance.properties里有没有canal.instance.mysql.slaveId配置项这个值默认是123456。如果和MySQL的server-id一样就改成一个不同的值比如654321。第三步重启canal实例观察日志。根因MySQL主从复制机制要求每个从节点的server-id必须全局唯一。canal伪装成一个从节点如果它的server-id和MySQL主库相同主库会认为这是我自己直接拒绝复制请求。这里插一句生产中很多人会忽略这个参数因为它的默认值看起来没存在感。建议部署清单里专门加一行确认slaveId与所有MySQL节点均不重复。5.3 坑三位点丢失导致重复消费或漏消费现象canal重启后客户端收到了之前已经处理过的旧数据或者直接跳过了部分数据。排查链路第一步确认canal的meta信息存储在哪里。默认情况下位点存在内存中Server重启就会丢失从当前最新位置开始消费。如果你需要持久化要在canal.properties里配置canal.instance.global.spring.xmlclasspath:spring/memory-instance.xml改为file-instance.xml可以把meta写到本地文件。第二步检查客户端代码是否正确ack。只get不ack重启后会从上次ack的位置重新拉取产生重复消费ack太快也可能出问题如果先ack再处理业务逻辑处理失败时数据就再也拉不回来了这是至少一次语义下最常见的取舍。第三步设计消费幂等。无论位点多可靠分布式环境下重复消费几乎不可避免所以消费端的幂等才是最终兜底。根因canal的位点机制是经典的有且至少一次消费语义重复消费是常态漏消费是因为位点推进到了业务成功之前。5.4 坑四多个客户端实例同时消费同一个instance现象多个服务实例都连了同一个canal destination结果每条数据只被其中一个实例收到而不是每个实例都收到。排查链路第一步确认你的业务诉求是广播还是负载均衡。canal不保证多个client同时消费时每个client都能拿到全量数据默认是竞争消费——一条数据只会给一个client。第二步如果确实需要多个下游分别消费同一份数据可以有三种方案一是每个下游对应一个独立的instance二是通过MQ模式发布订阅三是使用canal的CanalMQStarter配多个不同的group。根因canal的tcp模式的定位是一个destination对应一个消费组多个client是在消费组内竞争拉取不是复制分发。5.5 坑五DDL事件和事务边界导致的数据异常现象执行ALTER TABLE之后消费端收到的数据突然解析失败或者一个事务内的多变更变得不完整。排查链路第一步检查消费端是否处理了EntryType.ROWDATA之外的事件。DDL事件属于ENTRY_TYPE的其他类型如果你在消费端直接跳过所有非ROWDATA事件DDL对下游可能造成结构不一致。第二步开启canal的并行消费时要注意事务一致性。如果下游是数据库或缓存建议按事务边界整体提交避免同一个事务的变更被拆到不同地方执行。根因canal按事务为单位读取binlog但消费端如果并行处理事务边界会被人为拆散。这五个坑基本覆盖了我见过的90%小白问题。说实话每个坑背后都对应一个理解盲区不是性能问题不是bug而是对canal是怎么工作的理解不到位。所以每次出问题先回到第3章的原理图里定位是在哪个环节排查速度会快很多。6. 进阶思路从单机Demo走向生产环境Demo跑通、坑也避开之后你可能会问那真实项目里一般怎么用这一章我讲三个最常用的落地方向以及小白从单机走向生产时需要补的功课。6.1 实战方向一数据库变更同步到Redis这是canal最常见的用法。假设你的商品表product里有价格字段价格变了缓存也得变。传统做法是改业务代码在更新数据库后删除缓存用canal之后消费端收到UPDATE事件直接从JSON里提取最新值执行SET key value或者删除缓存。业务代码零侵入。要注意两点第一canal发来的是列名值的结构不是SQL所以你需要自己把变更映射成业务对象这个映射逻辑一般写在消费端一个独立模块里第二如果变更频繁要注意给缓存加防抖比如1秒内同一行的多次变更合并成一次刷缓存避免抖动。6.2 实战方向二异构数据同步到Elasticsearch把MySQL的数据同步到ES做全文检索也是高频场景。流程大致是canal监听MySQL变更 → 消费端把变更拼成ES文档 → 调用ES的index/update接口。这里有个常见的设计问题ES里可能需要宽表比如订单文档里要包含用户名、商品名而这些字段来自其他表。如果只订阅订单表拿不到关联信息。解决思路有两个一是订阅多张表在消费端做内存关联二是让业务方写出包含所有所需字段的预聚合表canal只订阅这一张。第二个方案更简单可靠推荐小白先这么做。6.3 从单机到集群三个生产级配置生产环境不可能只跑一个单机canal但我也要提醒你一开始别急着上集群先把下面三件事做好比盲目加机器更有价值。第一个是位点持久化。前面说过默认位点在内存在Server重启就丢。生产环境务必配置成file模式或zookeeper模式。我用的是file模式最简单但要注意磁盘空间和文件权限。第二个是订阅粒度收敛。开发环境订阅所有库所有表没问题生产环境一定要用filter.regex收窄到具体业务库和表不然会白白消耗大量性能还可能把其他团队的数据变更也带进你的消费逻辑。第三个是监控告警。canal是实时链路一旦它停了下游全部数据滞后。至少要做到进程存活监控、消费延迟监控canal有delayed指标、客户端积压消息数监控。这些指标都可以通过JMX或者集成到现有监控系统里别等业务方来问为什么数据还没同步过来你才发现问题。至于集群模式canal支持多Server ZooKeeper的HA方案多个instance由ZK协调一台挂了自动切换到另一台。这个配置本身不复杂但我建议在你的业务真正需要高可用之前先把单机的监控和运维做扎实因为集群也解决不了消费端逻辑的bug。7. 最后的实用建议这一章不聊原理、不聊配置就说一些零散但真实的经验。第一开发环境一定要把binlog打开。我见过不少团队本地MySQL默认没开binlog结果canal跑不起来排查半天发现是本地环境的锅。你可以在MySQL配置里直接默认开启对开发环境性能影响微乎其微。第二消费端一定要做幂等。不管你用哪种位点配置重复消费都是可能发生的所以消费逻辑必须能反复执行而结果一致。比如同步到ES用文档ID做upsert天然幂等同步到Redis覆盖写本来就幂等同步到另一个MySQL用主键做INSERT ... ON DUPLICATE KEY UPDATE就能保证不炸。第三binlog里的时间字段要小心。canal解析出的时间字段默认是字符串格式时区可能有偏移。如果你发现同步的数据在时间上差了8个小时去检查canal的时区配置。第四测试的时候别只测INSERT。我见过有人Demo只验证了insert就以为一切正常结果上线后update和delete全没处理数据越差越多。测试至少覆盖增删改三种事件最好再测一个事务里多条变更的顺序性。最后说一句掏心窝的话canal入门真的不难难的是你要理解它和业务系统的边界在哪里。它能帮你解决数据同步但它不会帮你把下游的消费逻辑写好也不会帮你兜住消费端抛出的异常。把它当成一个可靠的数据搬运工真正干活的内容还是要靠你精心设计消费端来处理。我用canal做了好几个项目的增量同步从最开始照猫画虎跑通Demo到现在能快速定位各种奇怪问题最大的感受就是工具本身只是管道关键还是对数据变化链路有一颗敬畏之心。希望你也能顺着这篇文章跑通自己的第一个Demo然后在这个基础上一步步把数据流动做得更稳、更优雅。 Canal实战：基于MySQL binlog的增量数据同步全解析
1. 先搞清楚canal到底在解决什么问题如果说这两年数据同步领域哪个组件最值得花一个星期认真学一遍我的答案一定是canal。这名字你可能在简历上见过在招聘JD里见过甚至在同事的代码评审里被吐槽过但真要问你它到底是干嘛的很多人的回答其实是模糊的嗯就是同步MySQL数据的——差不多对但不够准确。我换个说法你就明白了canal做的是MySQL增量数据订阅也就是监听MySQL的binlog日志每当数据库里发生insert、update、delete操作canal就能像装了窃听器一样在毫秒级把这个变更事件推给你。注意关键词是增量——它不是全量复制不是每天导一次数据而是从你部署那一刻开始实时关注每一条新变化。那它解决的到底是什么问题我举个最常见的场景。你们系统里肯定有一张热点表比如商品信息、用户资料、配置数据业务方读得很频繁。为了扛住流量你引入了Redis做缓存查询走缓存写操作更新数据库。听起来没问题但更新数据库的同时同步缓存这一步写法稍微不严谨就会出现缓存和数据库不一致的脏数据。常见的做法有几种双写、定时任务全量刷新、消息队列异步通知。双写逻辑侵入业务代码定时任务有延迟且浪费资源消息队列异步通知倒是好但你需要手动把每一次数据变更封装成消息发出去代码量不小。canal的思路是我不碰你的业务代码我在数据库这一侧偷听binlog把变更事件统一抓出来再推给你的下游。无论你是想去同步缓存、同步到ES、同步到数据仓库还是同步到另一个MySQL都不需要改业务代码。这个解耦的价值用过的人都懂——只需要一次配置全公司的数据消费方都能受益。对小白来说canal最友好的一点是它自带一个客户端订阅模型你只要能写Java代码几行就能跑通一个Demo完全不需要一开始就理解它的存储原理。但如果你想用好它光是会跑肯定不够因为真正的坑全在细节里binlog格式没配对、位点丢失导致重复消费、和多个订阅端抢数据……这些问题我用后面几个章节慢慢展开。顺便说一句canal项目的名称来自canal这个英文单词中文意思是运河、水道寓意很直白数据从MySQL流向下游的通道。这个比喻特别贴切你越往后用越会觉得它真的就像一条管道把数据库的血液输送到各个需要的地方。2. 安装前的三个关键开关JDK、binlog和权限安装canal本身不难难的是准备工作。很多小白按照网上的教程一步步来结果客户端连不上、收不到消息回头一看全是配套环境没弄对。所以这一章我把安装前置条件拆成三个开关每个都要提前确认坏了哪一个都会让你后面抓狂。2.1 第一个开关JDK版本canal的server端是Java写的所以你机器上得有JDK。这里有个坑需要你避一下canal 1.1.x系列用JDK 8没问题但如果你用的是比较新的canal 1.1.7版本或者想跑源码调试建议直接用JDK 11。我遇到过用JDK 8跑新版本canal启动的时候报了一些奇怪的并发类加载错误看起来像是代码问题其实是编译目标版本太高导致的。装好之后确认一下版本java -version正常情况下你会看到类似openjdk version 1.8.0_***或openjdk version 11.0.***的输出。如果你的机器上存在多个JDK建议用export JAVA_HOME/path/to/jdk指定清楚避免canal的启动脚本拿到错误的版本。2.2 第二个开关MySQL必须开启binlogcanal的工作原理是伪装成MySQL的从节点向主库请求binlog日志。所以你的MySQL必须开启binlog这是整个链条的源头。修改MySQL配置文件Linux下一般是/etc/my.cnf或/etc/mysql/mysql.conf.d/mysqld.cnf[mysqld] log-binmysql-bin binlog-formatROW server-id1这三行缺一不可尤其第二行binlog-formatROW。我见过有人在测试环境用了默认的STATEMENT格式结果canal解析出来一堆SQL语句根本拿不到精确的行级变更数据。因为canal的核心输出是某行某个字段从旧值变成了新值而这种结构化信息只有ROW格式才完整保留。改完之后记得重启MySQLservice mysqld restart或者systemctl restart mysqld。然后进入MySQL命令行验证SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format;两个结果都符合预期这个开关才算真正打开。还有一个细节容易被忽略server-id不能和canal实例配置的server-id重复。我用一个真实的踩坑经历说明第一次部署的时候我随手给canal设置了server-id1和MySQL自己的server-id冲突结果canal连接主库时直接被拒日志里报ERROR: Got error 1236。这个问题排查了很久才定位到因为报错信息很不直观。所以提前规划好MySQL给一个server-idcanal给另一个比如101、102这种互不干扰。2.3 第三个开关MySQL授权账号canal需要用一个专门的MySQL账号连接主库这个账号要有REPLICATION SLAVE和REPLICATION CLIENT权限。很多人在这步直接用root账号倒也能跑通但从规范角度不推荐尤其到了生产环境最少权限原则是底线。创建账号的SQL如下CREATE USER canal% IDENTIFIED BY canal123; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;这里给%是允许任意主机连接实际生产建议限定为运行canal的那台机器IP安全第一。另外注意如果数据库里已有的表结构比较复杂canal在解析DDL事件时可能还需要访问表定义所以顺手把SELECT权限也加上。这三件事做完环境准备基本就绪。老实说这个过程十分钟都用不了但价值远超十分钟——它决定了你后面所有排查动作能不能有一个干净的基础。3. 一条数据从MySQL到客户端的完整路径很多教程上来就让你装canal、跑Demo等你跑通了还是一头雾水它到底是怎么拿到数据的这一章我从原理层面完整讲一遍大概需要你花五分钟耐心读读完之后你就拥有了排查问题的底层地图遇到任何诡异现象都能知道往哪个环节去查。3.1 先理解MySQL主从复制是怎么回事canal的核心机制其实是在模仿MySQL的从库。MySQL的主从复制过程很简单主库把每一次变更写入binlog从库连接到主库请求从某个binlog位点开始的数据拿到之后在本地重放。canal做的事情和从库一模一样只不过它拿到这些日志之后重放的不是SQL而是把日志解析成结构化的数据变更事件交给下游程序处理。这个过程涉及binlog里的几个概念我用大白话解释一下binlog文件MySQL落盘生成的日志文件比如mysql-bin.000001达到一定大小后自动切换下一个文件。binlog位点position一条日志在文件中的偏移量它是唯一的。客户端恢复消费时只要记住上次消费到的position就能从断点继续。GTID全局事务ID是比position更高级的定位方式不依赖具体文件名和偏移量。新版本canal推荐用GTID模式主从切换时能自动适应。3.2 canal内部的三段式流水线canal从拿到binlog到吐出数据内部经历了三个核心阶段第一阶段抓取Parser。canal伪装成从库向MySQL发送dump协议请求。MySQL就会从指定的binlog位点开始把日志一条条推送给canal。这个阶段最纯粹只负责拿。第二阶段解析Sink。拿到的binlog是事件流需要解析。比如一个UPDATE事件会解析出旧值before和新值after这个是canal最有价值的输出——下游可以直接拿到{id: 1, name: 旧值}变成{id: 1, name: 新值}这样的JSON数据而不需要自己拼SQL去反查。第三阶段存储Store。解析完成后事件会写入一个内存环形队列等待客户端来拉取。canal内置了一个基于内存的存储组件默认容量是16384个事件客户端消费速度跟不上时事件会在内存里排队排满了之后才会出现丢数据或者阻塞的情况。整个链条如下图所示文字版MySQL binlog → canal Server 抓取 → 解析 → 内存队列 → 客户端(Client) 拉取 → 你的业务代码/ES/Redis...3.3 你眼里的canal组件Server、Instance和Client实际部署时你会接触到几个概念这里一次性讲清楚避免后续被配置项绕晕canal Server一个独立运行的Java进程核心服务。canal InstanceServer内部的一个逻辑实例每个Instance对应一个MySQL数据源。一个Server可以配置多个Instance每个Instance有独立的配置、独立的订阅位点、独立的队列。canal Client你写的消费端程序通过TCP协议从Server拉取数据。打个比方Server是餐厅Instance是厨房里的一个灶台——每个灶台对应一个供应商MySQLClient是来吃饭的顾客。一个餐厅有多个灶台很正常每个灶台负责炒不同的菜互不干扰。从部署角度小白刚开始只需要一个Server、一个Instance、一个Client后面需要订阅多个库或者对接多套下游时再考虑多Instance。4. 跑通第一个Demo用Java代码订阅一条insert前面讲了那么多理论和准备这一章我们真正动手。我用一台Linux机器做演示canal部署包从官方GitHub Release页面下载最新稳定版即可解压后目录结构大致如下/canal ├── bin ├── conf │ ├── canal.properties │ ├── example │ │ └── instance.propertiesconf/canal.properties是全局配置conf/example是默认实例的配置目录instance.properties是实例级配置。4.1 修改实例配置先改instance.properties这是核心# MySQL主库地址 canal.instance.master.address127.0.0.1:3306 # 连接MySQL的账号密码 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal123 # 订阅的数据库和表*表示所有 canal.instance.filter.regextest\\..* # 位点定位方式 canal.instance.gtidfalse重点说明一下filter.regex这个配置。它的格式是数据库名\\..*比如我只想订阅test库的所有表就写test\\..*想订阅test库下的user表就写test\\.user想订阅多个库用|分隔。这个正则很容易写错因为Java字符串本身也要转义所以\\.写成\\.而不是\.。我见过有人照抄配置把转义符弄丢了结果canal直接不输出任何数据查了半天才发现是正则匹配问题。4.2 修改全局配置再看canal.properties重点确认以下几项canal.port11111 canal.destinationsexample canal.instance.global.modetcpcanal.port是客户端连接的TCP端口默认11111不用改。canal.destinations指定启动哪些实例默认example对应conf目录下的example文件夹。canal.instance.global.modetcp表示消费端通过TCP直连方式拉取数据——这是最简单的方式后面聊MQ模式时再说。4.3 启动Servercd /canal/bin ./startup.sh启动后看日志确认没有异常tail -f /canal/logs/canal/canal.log如果看到canal server is running now之类的字样并且没有报错说明Server起来了。再检查实例状态tail -f /canal/logs/example/example.log正常情况下能看到start successful。这里有个新手最容易遇到的情况日志里显示canal is starting然后卡住不动多半是MySQL连接没通去查example.log里有没有connect timeout或者Access denied字样。4.4 写一个最小客户端canal官方提供了一个客户端库canal.client用Maven引入dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.7/version /dependency然后写一个最简单的Java类核心逻辑是建立连接 → 拉取消息 → 解析事件 → 打印变更内容。import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; import java.net.InetSocketAddress; import java.util.List; public class CanalDemo { public static void main(String[] args) { // 连接canal server CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, // instance名称 , // 无认证时填空 ); connector.connect(); // 订阅所有表这里要和instance的filter.regex对应 connector.subscribe(test\\..*); while (true) { // 一次拉取100条不设置超时 Message message connector.getWithoutAck(100); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { // 没有数据睡1秒再试 try { Thread.sleep(1000); } catch (InterruptedException e) {} continue; } for (CanalEntry.Entry entry : message.getEntries()) { if (entry.getEntryType() ! 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) { System.out.println(INSERT: rowData.getAfterColumnsList()); } else if (eventType CanalEntry.EventType.UPDATE) { System.out.println(UPDATE before: rowData.getBeforeColumnsList()); System.out.println(UPDATE after: rowData.getAfterColumnsList()); } else if (eventType CanalEntry.EventType.DELETE) { System.out.println(DELETE: rowData.getBeforeColumnsList()); } } } // 确认消费成功 connector.ack(batchId); } } }这段代码看着有点长但核心就三件事连接、拉取、确认。所有canal客户端的骨架都是这个模式你以后无论用Spring Boot封装还是写其他语言的客户端都逃不开这个套路。4.5 实测验证运行上面的Java类然后在MySQL里执行INSERT INTO test.user (name, age) VALUES (张三, 25);立刻就能在控制台看到类似下面的输出INSERT: [name: 张三, age: 25]看到这一行你的canal入门第一个Demo就算真正跑通了。从这一刻起任何数据库变更都会实时出现在你的消费端面前剩下的就是你怎么利用这些数据了。5. 小白最容易踩的五个坑含完整排查思路说实话canal的部署和Demo真的很简单但越是这种看起来简单的组件越容易在细节上翻车。这一章我把自己和周围人踩过的坑整理出来每一项都附上排查链路而不是直接给结论——因为下次你遇到的可能只是个相似但不完全一样的问题思路比答案更重要。5.1 坑一binlog格式不是ROW解析结果全是SQL现象客户端能连上能收到消息但解析出来的RowData是空的只有SQL语句。排查链路第一步回到MySQL执行SHOW VARIABLES LIKE binlog_format;如果结果是STATEMENT或MIXED基本就是问题所在。第二步修改配置为ROW重启MySQL。第三步重新执行一条INSERT看canal是否输出结构化的列值。根因只有ROW格式才会记录每一行变更的具体列值STATEMENT格式记录的是原始SQL文本canal无法从中还原出行级别数据。5.2 坑二server-id冲突导致连接被拒现象canal日志里出现ERROR: Got error 1236或者直接连接不上MySQL。排查链路第一步检查MySQL的server-idSELECT server_id;第二步检查canal的canal.instance.properties里有没有canal.instance.mysql.slaveId配置项这个值默认是123456。如果和MySQL的server-id一样就改成一个不同的值比如654321。第三步重启canal实例观察日志。根因MySQL主从复制机制要求每个从节点的server-id必须全局唯一。canal伪装成一个从节点如果它的server-id和MySQL主库相同主库会认为这是我自己直接拒绝复制请求。这里插一句生产中很多人会忽略这个参数因为它的默认值看起来没存在感。建议部署清单里专门加一行确认slaveId与所有MySQL节点均不重复。5.3 坑三位点丢失导致重复消费或漏消费现象canal重启后客户端收到了之前已经处理过的旧数据或者直接跳过了部分数据。排查链路第一步确认canal的meta信息存储在哪里。默认情况下位点存在内存中Server重启就会丢失从当前最新位置开始消费。如果你需要持久化要在canal.properties里配置canal.instance.global.spring.xmlclasspath:spring/memory-instance.xml改为file-instance.xml可以把meta写到本地文件。第二步检查客户端代码是否正确ack。只get不ack重启后会从上次ack的位置重新拉取产生重复消费ack太快也可能出问题如果先ack再处理业务逻辑处理失败时数据就再也拉不回来了这是至少一次语义下最常见的取舍。第三步设计消费幂等。无论位点多可靠分布式环境下重复消费几乎不可避免所以消费端的幂等才是最终兜底。根因canal的位点机制是经典的有且至少一次消费语义重复消费是常态漏消费是因为位点推进到了业务成功之前。5.4 坑四多个客户端实例同时消费同一个instance现象多个服务实例都连了同一个canal destination结果每条数据只被其中一个实例收到而不是每个实例都收到。排查链路第一步确认你的业务诉求是广播还是负载均衡。canal不保证多个client同时消费时每个client都能拿到全量数据默认是竞争消费——一条数据只会给一个client。第二步如果确实需要多个下游分别消费同一份数据可以有三种方案一是每个下游对应一个独立的instance二是通过MQ模式发布订阅三是使用canal的CanalMQStarter配多个不同的group。根因canal的tcp模式的定位是一个destination对应一个消费组多个client是在消费组内竞争拉取不是复制分发。5.5 坑五DDL事件和事务边界导致的数据异常现象执行ALTER TABLE之后消费端收到的数据突然解析失败或者一个事务内的多变更变得不完整。排查链路第一步检查消费端是否处理了EntryType.ROWDATA之外的事件。DDL事件属于ENTRY_TYPE的其他类型如果你在消费端直接跳过所有非ROWDATA事件DDL对下游可能造成结构不一致。第二步开启canal的并行消费时要注意事务一致性。如果下游是数据库或缓存建议按事务边界整体提交避免同一个事务的变更被拆到不同地方执行。根因canal按事务为单位读取binlog但消费端如果并行处理事务边界会被人为拆散。这五个坑基本覆盖了我见过的90%小白问题。说实话每个坑背后都对应一个理解盲区不是性能问题不是bug而是对canal是怎么工作的理解不到位。所以每次出问题先回到第3章的原理图里定位是在哪个环节排查速度会快很多。6. 进阶思路从单机Demo走向生产环境Demo跑通、坑也避开之后你可能会问那真实项目里一般怎么用这一章我讲三个最常用的落地方向以及小白从单机走向生产时需要补的功课。6.1 实战方向一数据库变更同步到Redis这是canal最常见的用法。假设你的商品表product里有价格字段价格变了缓存也得变。传统做法是改业务代码在更新数据库后删除缓存用canal之后消费端收到UPDATE事件直接从JSON里提取最新值执行SET key value或者删除缓存。业务代码零侵入。要注意两点第一canal发来的是列名值的结构不是SQL所以你需要自己把变更映射成业务对象这个映射逻辑一般写在消费端一个独立模块里第二如果变更频繁要注意给缓存加防抖比如1秒内同一行的多次变更合并成一次刷缓存避免抖动。6.2 实战方向二异构数据同步到Elasticsearch把MySQL的数据同步到ES做全文检索也是高频场景。流程大致是canal监听MySQL变更 → 消费端把变更拼成ES文档 → 调用ES的index/update接口。这里有个常见的设计问题ES里可能需要宽表比如订单文档里要包含用户名、商品名而这些字段来自其他表。如果只订阅订单表拿不到关联信息。解决思路有两个一是订阅多张表在消费端做内存关联二是让业务方写出包含所有所需字段的预聚合表canal只订阅这一张。第二个方案更简单可靠推荐小白先这么做。6.3 从单机到集群三个生产级配置生产环境不可能只跑一个单机canal但我也要提醒你一开始别急着上集群先把下面三件事做好比盲目加机器更有价值。第一个是位点持久化。前面说过默认位点在内存在Server重启就丢。生产环境务必配置成file模式或zookeeper模式。我用的是file模式最简单但要注意磁盘空间和文件权限。第二个是订阅粒度收敛。开发环境订阅所有库所有表没问题生产环境一定要用filter.regex收窄到具体业务库和表不然会白白消耗大量性能还可能把其他团队的数据变更也带进你的消费逻辑。第三个是监控告警。canal是实时链路一旦它停了下游全部数据滞后。至少要做到进程存活监控、消费延迟监控canal有delayed指标、客户端积压消息数监控。这些指标都可以通过JMX或者集成到现有监控系统里别等业务方来问为什么数据还没同步过来你才发现问题。至于集群模式canal支持多Server ZooKeeper的HA方案多个instance由ZK协调一台挂了自动切换到另一台。这个配置本身不复杂但我建议在你的业务真正需要高可用之前先把单机的监控和运维做扎实因为集群也解决不了消费端逻辑的bug。7. 最后的实用建议这一章不聊原理、不聊配置就说一些零散但真实的经验。第一开发环境一定要把binlog打开。我见过不少团队本地MySQL默认没开binlog结果canal跑不起来排查半天发现是本地环境的锅。你可以在MySQL配置里直接默认开启对开发环境性能影响微乎其微。第二消费端一定要做幂等。不管你用哪种位点配置重复消费都是可能发生的所以消费逻辑必须能反复执行而结果一致。比如同步到ES用文档ID做upsert天然幂等同步到Redis覆盖写本来就幂等同步到另一个MySQL用主键做INSERT ... ON DUPLICATE KEY UPDATE就能保证不炸。第三binlog里的时间字段要小心。canal解析出的时间字段默认是字符串格式时区可能有偏移。如果你发现同步的数据在时间上差了8个小时去检查canal的时区配置。第四测试的时候别只测INSERT。我见过有人Demo只验证了insert就以为一切正常结果上线后update和delete全没处理数据越差越多。测试至少覆盖增删改三种事件最好再测一个事务里多条变更的顺序性。最后说一句掏心窝的话canal入门真的不难难的是你要理解它和业务系统的边界在哪里。它能帮你解决数据同步但它不会帮你把下游的消费逻辑写好也不会帮你兜住消费端抛出的异常。把它当成一个可靠的数据搬运工真正干活的内容还是要靠你精心设计消费端来处理。我用canal做了好几个项目的增量同步从最开始照猫画虎跑通Demo到现在能快速定位各种奇怪问题最大的感受就是工具本身只是管道关键还是对数据变化链路有一颗敬畏之心。希望你也能顺着这篇文章跑通自己的第一个Demo然后在这个基础上一步步把数据流动做得更稳、更优雅。