canal是一个很经典的MySQL增量订阅组件这几年做数据同步、缓存更新、搜索索引构建的场景里它几乎是绕不开的一个工具。我最早接触它是因为业务上要实时感知数据库的变更把一张核心业务表的数据同步到另一个报表系统里。当时第一反应是在业务代码里到处加通知逻辑但想一想就放弃了耦合太重而且历史数据和脚本改库根本覆盖不到。后来换成canal思路一下子清晰了很多。这篇是第一篇实操笔记目标很纯粹把canal部署起来让它监控MySQL的数据变更然后写一个Java客户端把增删改的记录实时打印到控制台。整个链路不复杂非常适合第一次接触canal的人照着跑一遍跑通了后面再玩数据同步、缓存失效、消息分发这些就都有基础了。1. 先搞清楚canal在整条链路里的角色1.1 为什么是canal而不是业务代码自己通知如果数据库每次变更都要通知下游最常见的做法是在业务代码里写死了。比如用户改完资料代码里顺手调一下Redis删除缓存的逻辑订单状态流转代码里发一条消息给MQ。这个方案在小系统里没问题但一旦系统变复杂痛点就很明显。首先是耦合问题。下游系统越多业务代码里的通知代码就越臃肿改一个下游就得改业务方谁有关系谁就得跟着动。其次是漏数据问题。接口正常调用还能感知到但如果有人直接在数据库客户端执行了一条UPDATE或者凌晨跑了个定时批量脚本甚至DBA做了一次数据订正业务代码完全不知情这些变更就是漏网之鱼。第三是事务一致性问题。业务代码里先改库再通知如果通知失败了数据库其实已经变了下游永远收不到那一次变更。canal的办法是跳出来站在MySQL的视角解决。MySQL本身有binlog主从复制就是靠它工作的。canal模拟成一个MySQL从库跟主库建立复制关系主库的binlog会源源不断地推给它然后canal把binlog里的字节流解析成结构化的行数据变更事件。这样不管数据库的变更来自哪里只要是正经落到了binlog里canal都能感知到。感知数据变更这件事从业务系统里剥离出来了下游系统各自订阅各自需要的部分彼此不打扰。1.2 server和client是怎么配合的canal整体分两层。canal server负责跟MySQL打交道的脏活累活拉取binlog、解析binlog、按订阅规则过滤数据然后对外暴露一个TCP接口。canal client是业务侧的消费者负责建立连接、批量拉取解析好的数据事件再去做实际业务处理。这么拆分的好处是server只需要在机房部署一套它可以对接多个下游client。client之间也可以按照自己的需求订阅不同的库表。我做数据同步的消费端和做埋点统计的消费端订阅规则不同互不影响哪怕订阅的是同一张表两个client也各自有独立的消费位点不会互相干扰。好懂一点的比喻是MySQL是出版社binlog是印刷好的新书canal server是物流中转站它按图书分类把书打包好client就是书店。中转站不关心书店是把书卖出去还是摆在橱窗当装饰它只负责运货和分类。这个比喻虽然粗糙但能帮新手快速记住职责边界。1.3 这个demo要跑通的完整链路从大的层面看这次实操涉及三个节点链路是这样的第一段MySQL要配置好binlog并且设置成ROW模式。这条是canal能不能解析出字段级数据的前提。第二段canal server启动之后伪装成MySQL从库向MySQL发起binlog dump请求然后持续接收并解析binlog。第三段Java客户端连接到canal server获取解析好的数据事件打印到控制台。整个过程里不需要改任何业务代码也不用给表加字段纯外部视角完成数据库变更的感知。只要你往MySQL里插入一条、修改一条、删除一条控制台就能在几秒内看到对应的变更记录。2. 动手之前MySQL这边必须准备好2.1 binlog配置和格式验证说句实在话canal入门过程中大部分“服务端明明起来了但收不到数据”的问题根源都在MySQL配置这一环。先把binlog开启改配置文件my.cnfWindows是my.ini在[mysqld]段下加三行[mysqld] log-binmysql-bin binlog-formatROW server-id1log-bin开启binlog后面跟的是日志文件前缀binlog-formatROW这是canal工作的关键前提。如果设成STATEMENTbinlog里记录的是SQL语句本身canal解析不出某个字段从什么值变成了什么值只有ROW模式binlog里才是每一行前后的具体数据server-idMySQL复制环境里的标识这个值不能和已有的从库或canal的slaveId冲突否则复制关系会乱改完配置文件记得重启MySQL。启动之后进MySQL执行这两条验证命令SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format;log_bin的值是ONbinlog_format的值是ROW这个前置条件就算过了。如果binlog_format显示的是STATEMENT大概率是配置文件没被正确读取用一条命令确认一下MySQL实际读取的配置文件路径SHOW VARIABLES LIKE basedir;然后检查是不是把配置写到了错误的文件里。另外如果你的MySQL版本在5.6以上可以顺手确认一个参数binlog_row_image它决定binlog里是否记录完整行镜像。默认值是FULL也就是变更前后所有字段都记录。如果被改成MINIMALcanal解析UPDATE时经常拿不到未修改字段的旧值数据同步场景会非常头痛。保证这一项也是FULLSHOW VARIABLES LIKE binlog_row_image;注意binlog是主从复制的基础设施很多团队的MySQL虽然没配从库但binlog一直是开着的。如果你的MySQL已经处于其他主从复制环境里改配置前务必先确认server-id不要和别人冲突别为了跑一个demo把现有复制架构搞出问题。2.2 创建账号和一张测试表canal要连接MySQL需要给它的账号授权。权限主要三类SELECT权限用于读取表结构REPLICATION SLAVE权限用于模拟从库拉取binlogREPLICATION CLIENT权限用于查询主库状态。创建账号的命令可以这样写CREATE USER canal% IDENTIFIED BY canal; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;权限范围这里直接给了所有库。实际生产环境里如果只希望canal访问特定的库可以把*.换成具体的库名加.按最小权限原则收窄范围。我这个demo图省事先用全库权限。接着建库建表。测试表结构尽量简单但要能体现字段变化的过程我习惯建这样一张CREATE DATABASE IF NOT EXISTS test DEFAULT CHARSET utf8mb4; USE test; CREATE TABLE IF NOT EXISTS user ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(50), age INT, email VARCHAR(100), create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP );这里用utf8mb4而不是utf8是一个我踩过的小坑。utf8在MySQL里并不是严格意义上的四字节UTF-8存emoji或者生僻字会有问题canal解析出来可能出现乱码。测试环境直接上utf8mb4省心。2.3 MySQL 8.0的认证插件问题如果你用的是MySQL 8.0创建账号后连接时偶尔会报认证失败。原因是8.0默认的认证插件换成了caching_sha2_password而老版本的canal客户端在连接时对这种认证方式兼容得不好。稳妥的解决办法是创建账号时显式指定mysql_native_passwordCREATE USER canal% IDENTIFIED WITH mysql_native_password BY canal; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;这个问题不解决canal server日志里报的错往往是Communications link failure、Access denied之类的排查半天可能都想不到是认证插件不兼容。3. canal服务端部署、配置与启动3.1 下载解压和目录认知canal是Java写的部署包以tar.gz或zip形式发布。到官方Release页面选一个版本我这次用的是1.1.5。下载完解压到本地一个目录结构如下canal.deployer-1.1.5/ ├── bin/ │ ├── startup.sh │ ├── startup.bat │ └── stop.sh ├── conf/ │ ├── canal.properties │ └── example/ │ └── instance.properties └── lib/bin目录是启停脚本。conf目录是配置所在canal.properties是server的全局配置example/instance.properties则是一个名为example的实例的配置。一个实例对应一个MySQL数据源后续想监听多个MySQL就要在conf下创建多个实例目录并在canal.properties里用canal.destinations声明。很多新手会犯一个错只改了conf/canal.properties然后发现怎么配置都不生效。原因往往是漏掉了conf/example/instance.properties。这两个文件的职责不一样必须先分清楚。3.2 canal.properties全局配置要点打开conf/canal.properties最核心的配置项并不多我先挑和本次demo直接相关的说明canal.port11111 canal.destinationsexamplecanal.portcanal server对外提供服务的TCP端口客户端连接用的就是这个端口canal.destinations声明要加载哪些实例多个实例用逗号分隔。比如canal.destinationsexample1,example2就必须在conf下建example1和example2两个目录文件里还有很多关于内存模式、MQ模式、持久化模式的配置如果是第一次做入门实操不建议动它们。保持默认就好先把链路跑通再去研究那些高级参数会顺畅得多。3.3 instance实例配置要点打开conf/example/instance.properties逐个调整数据库相关的配置canal.instance.mysql.slaveId54321 canal.instance.master.address127.0.0.1:3306 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal canal.instance.connectionCharsetUTF-8 canal.instance.filter.regextest\\..* canal.instance.filter.blackRegex逐项说明canal.instance.mysql.slaveIdcanal伪装成MySQL从库时使用的slaveId。如果MySQL已经有其他从库这个值不能和它们一样。默认54321基本够用canal.instance.master.addressMySQL的地址和端口canal.instance.dbUsername和dbPassword就是前面创建的那个canal账号canal.instance.connectionCharset连接MySQL时用的字符集写UTF-8canal.instance.filter.regex订阅规则重要。这个正则表达式决定了canal解析哪些库表的数据。测试环境我设成test\..表示订阅test库下所有表。这里有个细节正则里的点号是特殊字符需要转义所以中间有两个反斜杠。如果写成test.它表示的是test后面跟任意字符testxxx这种库也会被匹配进去虽然demo影响不大但容易误导。订阅所有库所有表则写.\..canal.instance.filter.blackRegex黑名单排除某些库表。比如我要排除test库里的log表可以写test\.logdemo阶段先留空filter.regex是正宗的Java正则多个不同的订阅范围可以用逗号并列比如test\..*,biz\..*表示同时订阅test库和biz库。掌握了这一层你就能更灵活地控制接收范围。3.4 启动和验证配置改完后启动canal server。Linux环境执行bin/startup.shWindows环境执行bin/startup.bat。启动后不要急着自己写客户端先看两点。第一看进程日志。日志在logs目录下重点观察两个文件logs/canal.log是server总日志logs/example/example.log是实例日志。如果配置正确日志里能确认instance已经启动并且正在从MySQL同步binlog位置。第二确认端口在监听。Linux下执行netstat -tlnp | grep 11111只要看到11111端口有进程在监听说明canal server对外服务已经就绪。如果端口起不来去日志里看报错。常见原因是MySQL账号密码错误、网络不通、或者example目录不存在。这些问题第5部分我会集中梳理。启动成功的标志是日志里看不到ERROR级别的异常同时端口有监听。不要只看进程还在就跑有时候canal会进入反复重试连接MySQL的状态进程活着但实际服务不可用。4. 写Java客户端消费数据并打印4.1 依赖引入canal官方客户端是Java库新建一个普通的Maven工程JDK 8以上即可。在pom.xml里加依赖dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.5/version /dependencycanal.client依赖会把协议包、protobuf相关依赖带过来常规情况下不用额外加东西。唯一要注意的是如果工程里已经使用了其他版本的protobuf出现方法签名对不上或者NoSuchMethodError之类的错误去看一下依赖树把重复依赖排掉。4.2 连接、订阅与循环消费先写连接和订阅部分。canal支持多种连接方式demo用最简单的单机直连CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, , ); connector.connect(); connector.subscribe(test\\..*); connector.rollback();Connector建立后connect建立连接subscribe声明要订阅的库表参数同样是正则表达式这里写的是test库所有表。后两个字符串参数是连接canal server用的账号密码默认server没有开启客户端认证传空字符串即可。rollback的作用是让canal从最新的位点开始推送避免残留的未确认数据干扰理解。接着写消费循环。canal的数据拉取模型可以概括为“批量获取、手动确认”。客户端每次调getWithoutAck从server端拉一批数据业务处理完再调ack告诉server这一批消费完了。如果中途处理失败调rollback让server重新推送这一批数据。典型循环如下while (true) { Message message connector.getWithoutAck(1000, TimeUnit.SECONDS.toMillis(10)); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { continue; } for (Entry entry : message.getEntries()) { if (entry.getEntryType() ! EntryType.ROWDATA) { continue; } RowChange rowChange RowChange.parseFrom(entry.getStoreValue()); EventType eventType rowChange.getEventType(); for (RowData rowData : rowChange.getRowDatasList()) { // 这里按eventType区分INSERT/UPDATE/DELETE } } connector.ack(batchId); }getWithoutAck的第一个参数表示最多拉多少条第二个参数是阻塞等待时间单位毫秒。如果10秒内没有新数据返回一个batchId为-1的空消息continue继续循环。这里你可以把它理解成类似消费者拉取消息拉到了就消费拉不到就继续等。Entry是canal解析后的最小数据单位。一个事务会产生三类Entry包括事务开始、真正的行数据、事务结束。代码里只用ROWDATA类型过滤。RowChange里封装了事件类型和行数据列表。INSERT事件看afterColumnsDELETE事件看beforeColumnsUPDATE事件两者都有一一打印出来方便对比。4.3 完整Demo和验证效果把上一节拼起来就是一套能直接运行的代码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.Column; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; import com.alibaba.otter.canal.protocol.CanalEntry.EventType; import com.alibaba.otter.canal.protocol.CanalEntry.RowChange; import com.alibaba.otter.canal.protocol.CanalEntry.RowData; import java.net.InetSocketAddress; import java.util.List; import java.util.concurrent.TimeUnit; public class CanalClientDemo { public static void main(String[] args) { CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, , ); connector.connect(); connector.subscribe(test\\..*); connector.rollback(); System.out.println(canal client started, waiting for mysql changes...); while (true) { try { Message message connector.getWithoutAck(1000, TimeUnit.SECONDS.toMillis(10)); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { continue; } for (Entry entry : message.getEntries()) { if (entry.getEntryType() ! EntryType.ROWDATA) { continue; } RowChange rowChange RowChange.parseFrom(entry.getStoreValue()); EventType eventType rowChange.getEventType(); System.out.println( binlog[ entry.getHeader().getLogfileName() : entry.getHeader().getLogfileOffset() ] , name[ entry.getHeader().getSchemaName() , entry.getHeader().getTableName() ] , eventType : eventType); for (RowData rowData : rowChange.getRowDatasList()) { if (eventType EventType.INSERT) { printColumns(rowData.getAfterColumnsList(), after); } else if (eventType EventType.UPDATE) { printColumns(rowData.getBeforeColumnsList(), before); printColumns(rowData.getAfterColumnsList(), after); } else if (eventType EventType.DELETE) { printColumns(rowData.getBeforeColumnsList(), before); } } } connector.ack(batchId); } catch (Exception e) { e.printStackTrace(); connector.rollback(); } } } private static void printColumns(ListColumn columns, String flag) { for (Column column : columns) { System.out.println(( flag ) column.getName() : column.getValue()); } } }运行这个主类看到“canal client started”后去MySQL执行一条插入INSERT INTO test.user (name, age, email) VALUES (张三, 25, zhangsanexample.com);控制台几秒内会打印出类似下面的内容 binlog[mysql-bin.000001:1234] , name[test,user] , eventType : INSERT (after) id : 1 (after) name : 张三 (after) age : 25 (after) email : zhangsanexample.com (after) create_time : 2025-01-01 12:00:00接下来再执行一条UPDATE和DELETE观察输出差异。UPDATE会同时打印before和afterDELETE只有before。通过这种逐列打印能很直观地理解canal的列值模型。4.4 对输出结果的解读拿到数据之后要能看懂每个字段的含义。第一行的binlog[mysql-bin.000001:1234]前半部分是binlog文件名后半部分是日志偏移量它代表这条变更在binlog里的确切位置。之后是库名、表名、事件类型。后面的每一行是列名和值。列数据里其实还有两个demo中没有细化的属性isNull表示该字段是否为NULLupdated表示该字段在本次变更中是否真的修改过。正式业务里这两个属性经常决定下游要不要处理。比如缓存更新场景如果某列没变可以跳过缓存刷新如果是UPDATE事件里所有字段updated都为false那这次变更可能可以直接忽略。5. 常见问题、排查思路与避坑速查5.1 服务端连接失败该怎么查连接不上canal server是高频问题。表现是客户端抛ConnectException: Connection refused。排查按这个顺序来先确认canal进程真的启动了。可以执行netstat -tlnp | grep 11111没有输出多半是server进程异常退出了直接看logs/canal.log和logs/example/example.log里的ERROR日志。再确认防火墙本地开发通常没问题跨机器连接时先放行11111端口。最后确认客户端连的是不是canal的11111端口我遇到过有人把canal服务端地址误写成了MySQL的3306端口那自然连不上。另一种典型报错是客户端连接时提示destination无法找到。原因基本是客户端连接字符串里的实例名和canal.destinations配置不一致。客户端写example配置里也必须是example两边大小写都不能差。还有一类账号认证失败表现是canal server日志里出现Access denied for user。这种问题先检查instance.properties里的dbUsername和dbPassword。如果账号密码确认无误再复查是不是MySQL 8.0的认证插件问题按前面说的方法用mysql_native_password重建账号通常能解决。5.2 拿不到数据的问题出在哪如果canal server正常启动客户端也连上了但控制台就是等不到数据这属于最让人头大的情况。从经验看按下面几个点逐个排查。第一binlog格式不对。确认binlog_format是ROWbinlog_row_image是FULL。曾经遇到一个环境binlog格式是STATEMENTcanal能启动能消费但拿到的数据永远是空的或者解析不出字段内容。先把MySQL配置修好。第二订阅过滤器写错。canal服务端有filter.regex客户端有subscribe的正则两边都要指向目标库表。很多人只改了服务端客户端忘了subscribe或者subscribe写成了别的库。这里建议把服务端和客户端的正则都统一成test\..*减少认知成本。第三操作了错误的库表。插数据时确认执行的INSERT语句确实落到了被订阅范围的库里。看binlog文件名有没有变化也是个好办法执行一条插入后用SHOW MASTER STATUS查看当前binlog的position是不是变了如果变了但canal没反应问题基本在canal侧的过滤规则或解析环节。第四字符集问题。如果配置里connectionCharset是UTF-8但表实际用的字符集是gbk或者latin1解析出来的中文可能直接乱码或者被截断。统一用utf8mb4建库能避免一大半这类问题。第五时间字段时区偏差。如果你看到的create_time比实际入库时间少了8小时通常是连接时区没配对。可以在canal实例配置里加上时区参数或者在MySQL连接层面统一对齐时间不然数据同步到其他系统后对账会很痛苦。5.3 消费确认与重复消费的坑canal消费模型里最容易忽略的是ack。新手写完循环数据处理完直接进下一次循环不调ack。这样canal会认为消息始终未确认重启后这些数据又会重新推给你。最直接的影响是看起来像在重复消费。但不要因为怕重复就赶紧ack。正确的姿势是业务处理成功后再ack。处理半路上异常崩溃应该调rollback让canal重新推送。ack和rollback是配套的没有ack意味着系统始终认为是未消费状态。即便ack了也并不能完全避免重复。应用消费到消息之后、ack之前宕机重启后同样会重复。所以下游业务逻辑最好设计成幂等的比如拿主键做去重或者消费时先查一下目标表再决定插入还是更新。这个建议放生产环境之前就想清楚等真的出现重复数据再回头改代价就大了。另外如果消费逻辑本身很慢getWithoutAck拉取的消息一直堆积在本地可以适当调大单次批量大小或者改用多线程消费一批数据。但多线程消费时要保证同一批消息的处理顺序和ack时机可控否则容易把业务状态搞乱。5.4 几个容易被忽视的细节binlog默认保留时间有限。如果canal停了很久再启动可能发现要从binlog开头开始拉或者binlog已经过期被清理canal会报找不到指定binlog位置的错误。生产环境应根据数据恢复需求设置expire_logs_days或者binlog过期时间canal也会定期记录消费位点避免重启后从远古位置补数据补到崩溃。canal和业务代码不要混在一个服务里。初期方便但线上要保证canal消费服务的独立性和重启可控性最好拆成单独的一个消费端服务。instance的日志很有用。遇到问题时example.log里的错误信息比猜靠谱得多。打印一下看看最关键的信息基本都在里面。如果只是验证效果控制台打印完全够用。但正式项目里建议把解析后的变更数据落到消息队列或下游存储同时做好监控告警比如消费延迟、连接状态这些指标都得有。6. 一些实操中的体会这个demo虽然简单但它是理解canal的基础。我第一次跑这个链路时花时间最多的不是写代码而是排查为什么客户端连上了却取不到数据最后定位到是binlog_format还是STATEMENT模式。从那以后我养成了一个习惯任何canal相关的环境先把MySQL的binlog配置验证一遍再谈其他。另外有两点想多说一句。第一canal拉到的数据是数据库层面的事实它不会告诉你业务上的含义是“哪个用户在什么操作下改了这条数据”这些信息binlog里并没有。所以下游消费时通常还得结合自己的业务上下文去翻译。第二订阅正则这种看起来不起眼的配置往往是线上事故的源头服务端和客户端两边都要写得明确且收敛。宁可订阅范围偏窄也不要一开始就全库全表一把梭等链路跑通再慢慢扩大范围。这个系列我还会接着往下写方向大概率是canal把变更数据实时同步到Redis、Kafka这些更贴近业务的场景。先把这篇的基础链路摸熟后面的扩展就是水到渠成的事。