★ FEATURED ARTICLE 前阵子在做数据同步的时候我又撞上了那个经典问题业务库里的数据一变化下游的缓存、索引、报表都得跟着动。最粗放的做法是定时全量扫表但数据量大一点就吃不消想在应用里挨个写变更通知又容易漏。后来我干脆把目光放到 MySQL 的 binlog 上用 Canal 把增量变更实时接出来先落一个最简单的形态——监听 MySQL 数据然后把每一笔变更打印到控制台。这篇就把完整过程记录下来从 MySQL 侧准备、Canal 服务端部署到 Java 客户端怎么写、踩过哪些坑都尽量讲明白。1. 先搞清楚Canal在替我们做什么很多人第一次听到 Canal第一反应是“又一个消息中间件”其实它不是。Canal 是一个专门用来解析 MySQL binlog 的开源组件它的核心价值是把 MySQL 主库上的增量变更转换成结构化的事件再推给下游消费者。你可以把它理解成一个“数据库变更摄像头”MySQL 里每发生一次 insert、update、delete它都能感知到并且告诉你具体是哪张表、哪一行、变更成了什么。1.1 为什么我一开始没有选轮询和触发器在没有 Canal 这类工具之前常见的增量感知方案无非两种应用层主动轮询或者数据库触发器。轮询的问题在于“延迟和成本不可兼得”。如果你每秒查一次业务表能拿到秒级延迟但业务表大一点、查询条件复杂一点数据库压力立刻上去如果你把轮询频率降到每分钟一次延迟又变得不可接受。而且轮询本质上还是在“猜”哪些数据变了通常会带上一个 update_time 字段一旦某张表没这个字段或者历史数据被批量修复过轮询就会漏掉。触发器则是把变更逻辑塞进数据库里。每次 insert/update/delete 都去写一张变更记录表看起来直接但会把业务库和同步逻辑耦合在一起。触发器对主库写入性能有影响而且一旦同步逻辑出问题会反过来拖慢业务事务。更麻烦的是如果你有多套下游要消费触发器里的逻辑就很难复用。用 binlog 方案就不一样。binlog 是 MySQL 主从复制的基础本身就完整记录了所有数据变更且格式可以包含变更前后的整行数据。Canal 只是把自己伪装成一个 MySQL 从库然后去向主库要 binlog不侵入业务代码也不影响业务事务所有解析逻辑都在 Canal 和你的客户端里。1.2 Canal是怎么把自己伪装成从库的理解 Canal 之前得先理解 MySQL 主从复制的基本链路。主库开了 binlog 之后从库会建立一个专门的复制账号用这个账号向主库发起 dump 请求。主库收到请求后会把新的 binlog 事件源源不断地推给从库从库拿到这些事件后再在本地回放从而实现数据同步。Canal 干的事情就是把这套“从库”的逻辑在 Java 进程里重新实现了一遍。它不需要真的去存储数据而是把收到的 binlog 事件解析成 Java 对象然后通过 TCP 端口、MQ 或者 Kafka 等方式转发给客户端。所以对 MySQL 来说Canal 看起来就是一个普普通通的从库节点对业务应用来说Canal 变成了一条干干净净的增量数据管道。这也是为什么 Canal 能拿到非常精确的变更数据因为它读到的就是 MySQL 自己用于复制的那些日志。2. 动手前的MySQL准备binlog和账号一个都不能少在启动 Canal 之前MySQL 侧的配置是第一步。这一步如果没做好后面基本上全是白忙。我见过不少同事上来就下载 Canal改了配置却连不上 MySQL最后发现是 binlog 根本没开或者账号权限不对。这里我把需要检查的点全部列出来。2.1 开启binlog并让格式变成ROW首先确认 MySQL 是否开启了 binlog。如果你用的是 Linux 服务器通常在/etc/my.cnf或/etc/mysql/my.cnf里配置如果是本地测试用 Windows则在my.ini里配置。需要确保[mysqld]段下面有以下几行server_id100 log_binmysql-bin binlog_formatROW binlog_row_imageFULL expire_logs_days7这里每一项都有讲究。server_id是给这个 MySQL 实例的复制编号取值范围 1 到 2^32-1只要不和其他复制节点冲突就行。我习惯从 100 开始用方便后面加别的从库节点。log_bin开启 binlog 日志文件名前缀可以自己定义常见的是mysql-bin。binlog_formatROW是最关键的一项。MySQL 有三种 binlog 格式STATEMENT、MIXED、ROW。STATEMENT 记录的是 SQL 语句本身Canal 拿到的只是“执行了什么 SQL”很难得到精确的变更前后值MIXED 会根据情况切换也不稳定只有 ROW 格式下binlog 里记录的才是每一行数据实际被改成了什么样。Canal 官方也明确要求使用 ROW 格式所以这一步千万不能省。binlog_row_imageFULL同样重要。这个参数决定了 ROW 格式下 binlog 记录的是整行数据还是只记录被修改的列。如果设置成 MINIMAL更新操作可能只记录变更列和主键Canal 就没办法拿到完整的前后镜像后续做缓存同步时很吃亏。所以默认建议保持 FULL。expire_logs_days7控制 binlog 保留天数。Canal 如果是长期运行一般消费速度都能跟上这个参数不会太敏感但如果 Canal 宕机了很久而 binlog 已经被清理那中间的数据就追不回来了。做生产环境时可以根据业务容忍度调整保留时长。配置好之后记得重启 MySQL 让参数生效。2.2 创建一个只用于同步的Canal账号Canal 需要用一个 MySQL 账号去主库拉取 binlog。这个账号不需要超级权限但必须拥有 replication 相关的权限否则主库不会把 binlog 推给它。我一般会创建一个专门给 Canal 用的账号用户名就叫canal密码单独设置避免和业务账号混在一起。创建和授权的 SQL 如下CREATE USER canal% IDENTIFIED BY canal_pass; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;REPLICATION SLAVE是必须的Canal 靠它伪装成从库去请求 binlog。REPLICATION CLIENT用来查看主库状态比如SHOW MASTER STATUS这能帮助 Canal 在首次启动时定位到当前的 binlog 文件位置。SELECT权限是因为某些场景下Canal 需要查询表结构或字段信息提前给上更省心。需要注意的是如果你用的是 MySQL 8.0 及以上版本默认的密码认证方式是caching_sha2_passwordCanal 早期版本对这个认证方式支持得并不好连接时会直接报认证失败。解决办法是把这个账号改成mysql_native_password方式ALTER USER canal% IDENTIFIED WITH mysql_native_password BY canal_pass; FLUSH PRIVILEGES;这一点我踩过好几次MySQL 8.0 默认认证方式变化确实容易让人忽略。2.3 验证配置真的生效改完配置和账号后最好先手动验证一遍不要急着启动 Canal。用命令行连上 MySQL执行下面几条命令SHOW VARIABLES LIKE binlog_format; SHOW VARIABLES LIKE log_bin; SHOW MASTER STATUS;如果binlog_format返回ROWlog_bin返回ONSHOW MASTER STATUS能查到一个 binlog 文件名和 position说明主库侧已经准备好了。这个输出也可以顺手记下后面调 Canal 时可以用来对照。3. Canal服务端部署解压改配置启动MySQL 侧准备好了接下来就到 Canal 服务端。这里的“服务端”本质上是一个独立的 Java 进程负责连接 MySQL、拉取 binlog、解析事件然后提供给客户端。我使用的是 Canal 1.1.x 版本整个部署过程在 Linux 和 Windows 上差别不大我以 Linux 为例说明。3.1 下载、解压、认识目录从 Canal 官方发布渠道下载canal.deployer的压缩包解压到一个固定目录比如mkdir -p ~/canal tar -zxvf canal.deployer-1.1.4.tar.gz -C ~/canal解压后的目录结构大致是这样bin/ conf/ lib/ logs/bin下面存放启动和停止脚本conf下面放全局配置和实例配置lib是运行依赖的 jar 包logs是日志目录。最核心的两个配置文件在conf下一个是canal.properties对应整个 Canal Server 的全局配置另一个是conf/example/instance.properties对应一个独立的 Canal 实例。这里要理解“实例”的概念。一个 Canal Server 可以同时跑多个实例每个实例负责从某一个 MySQL 主库拉取 binlog实例之间互不影响。默认自带了一个名为example的示例实例我们直接拿它来改。3.2 canal.properties里真正需要调的点打开canal.properties里面有大量参数但第一次跑通其实需要动的很少。我建议先把注意力放在端口和实例列表上。canal.port11111 canal.destinationsexamplecanal.port是 Canal Server 对外提供 TCP 服务的端口客户端就是通过这个端口来订阅数据的。默认是 11111如果你本机端口被占用可以改成别的比如 11112但要记住后面客户端连接时要用同一个端口。canal.destinations表示当前 Server 要加载的实例名称列表多个实例用逗号分隔。默认就是example对应conf/example目录。如果你想新增一个实例一般建议复制一份 example 目录然后把文件夹名加到canal.destinations里。还有一些 MQ 相关的配置比如canal.serverMode如果是 TCP 模式默认就是 TCP不需要改。后面如果要用 Kafka 或者 RocketMQ才需要仔细调canal.mq.servers、canal.mq.topic这些参数。3.3 instance.properties是你最该盯住的文件真正决定 Canal 从哪里拉数据、怎么过滤表是conf/example/instance.properties这个文件。我每次部署时都会重点检查以下几项canal.instance.mysql.slaveId1234 canal.instance.master.address127.0.0.1:3306 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal_pass canal.instance.connectionCharsetUTF-8 canal.instance.filter.regex.*\\..*canal.instance.mysql.slaveId是 Canal 伪装成 MySQL 从库时使用的 slaveId。这个 id 不能和任何真实 MySQL 节点的 server_id 重复否则 MySQL 主库会认为有冲突。这里写成1234没问题。canal.instance.master.address就是要监听的 MySQL 主库地址和端口注意一定是主库的地址不是从库。如果你本机测试就写127.0.0.1:3306。dbUsername和dbPassword就是第二步创建的那个账号密码。canal.instance.filter.regex是表过滤规则这里的正则语法使用的是 Java 正则默认.*\\..*表示匹配所有库的所有表。如果你只想监听某个库比如test库的所有表可以写成canal.instance.filter.regextest\\..*如果想监听test库的user表和order表canal.instance.filter.regextest\\.user,test\\.order注意配置里要保留双反斜杠因为这是一个 Java 字符串形式的正则单反斜杠会被转义掉。这个规则非常容易写错我后面会专门说一个翻车案例。3.4 启动并看懂日志配置改完之后就可以启动 Canal 了。Linux 下执行cd ~/canal bin/startup.sh启动后不要急着看客户端先看日志。Canal 的日志集中在logs目录下有两个文件很关键logs/canal.logCanal Server 的全局日志能看到启动过程、端口绑定等。logs/example/example.log具体实例的运行日志能看到它是否成功连上 MySQL是否开始拉取 binlog。如果一切正常example.log里会出现类似start successful的日志。如果 MySQL 连不上这里会有异常堆栈报错原因一般都能直接搜到。如果是在 Windows 上测试用bin/startup.bat启动也是一样的。4. Java客户端把binlog里的RowData变成一行行控制台文本Canal Server 跑起来之后还缺一个消费者。官方提供了独立的客户端 jar 包支持 Java 直接连接 Canal 的 TCP 端口订阅实例然后循环拉取消息。下面我会写一个最小可运行的客户端目标就是把 MySQL 的增删改操作实时打印到控制台。4.1 连接Canal的核心参数新建一个 Java 工程引入 Canal 客户端依赖。不同版本的 Canal 客户端 API 基本一致我用的是 1.1.x 版本。如果你的项目是 Maven 管理可以直接在中央仓库搜索canal.client添加依赖也可以从 Canal 部署包解压后的 lib 目录里找到 canal client 相关的 jar 包手动引入。连接 Canal Server 的核心代码很简单CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, , );newSingleConnector表示连接一个 Canal Server。第一个参数是地址和端口对应服务端的canal.port第二个参数是实例名对应canal.destinations和conf/example里的example后两个参数用户名密码如果是 TCP 模式官方通常默认不校验可以留空。连接之后还需要两个关键操作subscribe和rollback。subscribe是告诉 Canal 你要订阅哪些库表这里可以传入正则比如.*\\..*表示订阅全部也可以传入test\\..*来订阅某个库。rollback的作用是把未确认的批量消息回滚到未发送状态保证从最新的位置开始消费初次启动时执行一下可以避免客户端启动后立刻收到之前堆积的旧消息。完整骨架代码如下connector.connect(); connector.subscribe(.*\\..*); connector.rollback();4.2 订阅、拉取、确认的骨架循环Canal 客户端是典型的“拉模式”客户端在一个 while 循环里不断向 Canal Server 获取消息拿到一批后解析处理处理完再发送 ack 确认。为什么要有 ack因为 Canal 需要知道客户端已经把消息处理完了如果客户端挂了没 ack之后重启还能重新消费这一批数据保证不丢。核心循环如下while (true) { Message message connector.getWithoutAck(100); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { connector.ack(batchId); continue; } for (CanalEntry.Entry entry : message.getEntries()) { // 解析entry } connector.ack(batchId); }getWithoutAck(100)表示尝试获取一批消息100 是超时时间毫秒。如果 Canal 在 100 毫秒内没有新数据会返回一条 batchId 为 -1 的空消息这时候什么都不用做ack 掉继续等下一条即可。message.getEntries()里才是真正包含 binlog 事件的数据结构。每条 entry 可能有不同类型比如事务开始、事务结束、行数据变更。我们需要过滤掉非行数据的条目只处理ROWDATA类型。4.3 解析Insert/Update/Delete并格式化输出现在到最核心的解析部分。CanalEntry.Entry里有很多字段常用的是entry.getHeader().getSchemaName()库名entry.getHeader().getTableName()表名entry.getStoreValue()序列化后的变更数据需要用CanalEntry.RowChange.parseFrom()反序列化rowChange.getEventType()事件类型比如 INSERT、UPDATE、DELETErowChange.getRowDatasList()变更的行数据列表一条 binlog 事件可能包含多行变更我写了一个formatColumn方法用来把一行数据里的列名和值拼成一个易读的字符串方便直接打印。完整代码大致如下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 java.net.InetSocketAddress; import java.util.List; public class CanalConsolePrinter { public static void main(String[] args) throws Exception { CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, , ); connector.connect(); connector.subscribe(.*\\..*); connector.rollback(); System.out.println(Canal client started, waiting for binlog events...); while (true) { Message message connector.getWithoutAck(100); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { connector.ack(batchId); continue; } for (CanalEntry.Entry entry : message.getEntries()) { if (entry.getEntryType() ! CanalEntry.EntryType.ROWDATA) { continue; } String schema entry.getHeader().getSchemaName(); String table entry.getHeader().getTableName(); CanalEntry.RowChange rowChange; try { rowChange CanalEntry.RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException(Parse event error, data: entry.toString(), e); } CanalEntry.EventType eventType rowChange.getEventType(); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { System.out.println( schema . table eventType ); if (eventType CanalEntry.EventType.INSERT) { System.out.println(after : formatColumns(rowData.getAfterColumnsList())); } else if (eventType CanalEntry.EventType.UPDATE) { System.out.println(before: formatColumns(rowData.getBeforeColumnsList())); System.out.println(after : formatColumns(rowData.getAfterColumnsList())); } else if (eventType CanalEntry.EventType.DELETE) { System.out.println(before: formatColumns(rowData.getBeforeColumnsList())); } } } connector.ack(batchId); } } private static String formatColumns(ListCanalEntry.Column columns) { StringBuilder sb new StringBuilder(); for (CanalEntry.Column column : columns) { if (sb.length() 0) { sb.append(, ); } sb.append(column.getName()).append().append(column.getValue()); } return sb.toString(); } }这段代码里我特别保留了CanalEntry.Column的getValue()它拿到的是字段的字符串形式。如果要拿到真正的 Java 类型可以根据column.getSqlType()再转换不过打印到控制台时字符串就够用了。UPDATE事件比较特殊它会同时携带beforeColumnsList和afterColumnsList分别对应更新前和更新后的数据。这样你就可以在客户端里做“旧值对比新值”的逻辑比如判断某个字段是否真的变了。而INSERT只有 after 数据DELETE只有 before 数据这是 RowData 结构上的规律。4.4 空消息、批量、ack这几个细节别忽略第一次写 Canal 客户端时最容易忽略的是ack的处理。如果你处理完业务逻辑后忘记 ackCanal 会认为这批消息还没消费成功会在你下次getWithoutAck时重新发送于是形成重复消费。虽然重复消费在打印场景下无所谓但在真正的同步逻辑里会引发脏数据。批量问题也需要注意。一个事务可能包含多条 binlog 事件但 Canal 会把它们打包在同一个Message里。所以循环里要遍历entryList不能只取第一条。如果同一个事务里先 update 了 10 行你会在一批消息里看到 10 个RowData循环处理时不要在中间过早ack。另外connector.rollback()的语义是回滚当前未确认的消息。很多人不理解为什么启动时要 rollback 一下其实是为了让 Canal 从上次 ack 成功的位点继续而不是从内存里的最新位点开始。如果你希望客户端启动后不丢任何消息rollback 是很有用的。5. 实测效果与翻车现场理论讲完了必须实际跑一遍。我本机的 MySQL 里新建了一个test库和一张user表然后启动 Canal Server 和 Java 客户端接下来在 MySQL 里做增删改看控制台输出。5.1 建表、插入、更新、删除看输出先建表CREATE TABLE test.user ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(50), age INT );客户端已经跑起来然后执行一次插入INSERT INTO test.user (name, age) VALUES (小明, 20);控制台会打印类似这样的内容 test.user INSERT after : id1, name小明, age20再执行一次更新UPDATE test.user SET age 21 WHERE id 1;输出 test.user UPDATE before: id1, name小明, age20 after : id1, name小明, age21再执行删除DELETE FROM test.user WHERE id 1;输出 test.user DELETE before: id1, name小明, age21到这里最基础的“监控 MySQL 数据打印到控制台”已经跑通了。可以看到Canal 给的信息比普通日志要完整得多尤其是 UPDATE 的前后对比对做数据同步、审计、缓存失效判断都非常有用。5.2 启动失败的高频根因与排查路径实际部署中最常见的启动失败集中在三个点。第一是 MySQL 连接不上。报错信息里通常会直接出现connect timed out或者Access denied。前者先检查canal.instance.master.address是否写对了以及 Canal 所在机器能否访问到 MySQL 的 3306 端口可以用telnet 127.0.0.1 3306验证后者多半是账号密码或者权限问题检查 MySQL 用户是否存在、REPLICATION SLAVE权限是否给到位。第二是 MySQL 8.0 的认证方式问题。如果你用的是 MySQL 8.0创建用户后直接让 Canal 去连很可能会报Authentication plugin caching_sha2_password cannot be loaded。解决办法就是我前面讲的把canal账号改成mysql_native_password。这一点如果不提前处理会卡住很久。第三是 slaveId 冲突。报错信息里会有类似slaveId already exists或者conflict的字样。解决办法是修改instance.properties里的canal.instance.mysql.slaveId确保和 MySQL 的server_id以及任何真实从库的 server_id 都不一样。排查路径我总结成一句话先看canal.log有没有启动异常再看example.log有没有连接或解析异常最后用 MySQL 命令行验证账号权限和 binlog 参数基本都能定位到问题。5.3 filter正则写错会把所有消息过滤掉这是一个非常隐蔽的坑。我在配置里写过canal.instance.filter.regextest\.user看上去没问题但实际上在 properties 文件里反斜杠会被 Java 加载时再做一次转义。所以正确写法必须是双反斜杠canal.instance.filter.regextest\\.user如果你只写了一个反斜杠最终得到的正则就变成了test.user。在正则语法里点号表示任意字符所以它会匹配testXuser、test1user这种表名但不会匹配真正的test.user。结果就是Canal 启动成功了客户端也连上了但 MySQL 里无论怎么增删改控制台都收不到一条消息。遇到这种“空转”现象优先检查 filter 正则尤其是转义字符。还有一个办法你可以在 MySQL 里把 Canal 账号的SELECT权限用好通过 Canal 日志观察它过滤后的订阅列表或者改成.*\\..*先验证全量订阅是否能收到消息再逐步缩小范围。6. 我的一些实践体会和下一步打算从 MySQL 配置到 Canal Server 部署再到 Java 客户端打印整个过程走下来我的最大感受是Canal 的上手门槛其实不高真正费时间的反而是各种环境细节。binlog 格式、账号权限、正则转义、ack 机制任何一个地方踩坑都会让你绕远路。建议第一次做的时候先在一个独立的 MySQL 实例上跑通不要直接在生产库上测试等流程完全顺了再评估灰度方案。另一个体会是控制台打印只是“第一步”它的价值在于帮你建立对 binlog 数据结构的直觉。你会慢慢理解Entry里有哪些字段RowData里的 before/after 到底长什么样以及事务边界在消息里如何体现。这些理解对后面做数据同步、监听重要表变更、对接消息队列都非常有帮助。下一篇我打算接着写 Canal 的进阶玩法比如如何同时监听多个 MySQL 实例、如何把变更数据投递到 MQ、以及在消费端如何做幂等和过滤。如果你在跑这个 demo 的过程中也遇到了奇怪的问题欢迎对照着我前面的排查思路再捋一遍很多时候问题就藏在那些“看起来没问题”的配置里。
阅读完成 · 觉得有帮助？
