1. 为什么你需要Canal一个数据同步的真实故事几年前我接手了一个电商项目当时我们遇到了一个非常头疼的问题。每当用户在商品详情页点击“加入购物车”时这个操作会写入主数据库但负责生成用户行为分析报表的另一个系统却总是要等上几分钟甚至更久才能看到这条新数据。这就导致运营同学看到的“实时热销商品”榜单其实是几分钟前的“旧闻”。为了解决这个延迟我们试过各种土办法比如定时去数据库里扫表结果不仅把数据库CPU拖得很高还经常漏掉数据或者重复处理。直到后来我们发现了Canal。你可以把它想象成数据库的“窃听器”或者“复印机”。它不打扰MySQL的正常工作只是安静地蹲在一边把数据库里发生的每一笔数据变动比如新增一条订单、修改用户信息、删除一条评论都原原本本地复制下来然后实时地“递送”给其他需要的系统。用了Canal之后我们的报表系统几乎能做到秒级同步用户体验和运营效率一下子就上来了。简单来说Canal就是一个用Java写的、专门抓取MySQL数据变更日志的工具。它的工作原理很巧妙把自己伪装成MySQL的一个从库Slave然后向主库说“嗨我是你的小跟班有什么数据更新都发我一份呗。”主库就会把记录数据变化的二进制日志binlog同步给它。Canal拿到这些二进制的日志后进行解析转换成我们能看懂的结构化数据再通过各种方式比如直接TCP连接、或者放到Kafka/RocketMQ这样的消息队列里发送出去。所以如果你也在为下面这些事烦恼那Canal可能就是你的菜数据实时同步像我们刚才说的需要把数据立刻同步到另一个数据库比如ES做搜索、Hive做分析或者缓存比如Redis。异地多活两个机房的数据需要保持同步。审计与监控需要记录下谁在什么时候改了哪些数据。缓存更新数据库数据一变对应的缓存能自动失效或更新。接下来我就手把手带你从零开始搭一个能用的Canal服务并写一个简单的客户端来消费这些数据变化。整个过程我会把我在实战中踩过的坑和总结的经验都告诉你保证你跟着做就能跑通。2. 动手之前准备好你的“舞台”搭建任何服务环境准备都是第一步也是最容易出问题的一步。这里我们假设你有一台干净的CentOS 7服务器用虚拟机或者云服务器都行咱们从零开始准备。2.1 基础环境Java与MySQLCanal是Java写的所以第一件事就是安装JDK。我习惯用yum安装省心。# 检查是否已安装Java java -version # 如果没安装安装OpenJDK 8Canal 1.1.5兼容性很好 yum install -y java-1.8.0-openjdk-devel # 安装后再次检查 java -version看到类似openjdk version 1.8.0_392的输出就对了。接下来是重头戏配置MySQL。Canal工作的前提是MySQL必须开启binlog并且格式必须设置为ROW。这是很多新手会栽跟头的地方。编辑MySQL配置文件vim /etc/my.cnf在[mysqld]这个段落里确保有以下几行。如果你的文件里已经有这些项就修改它们如果没有就加上。[mysqld] # 启用binlog并设置基础文件名 log-binmysql-bin # 设置binlog格式为ROW这是Canal能解析的唯一格式 binlog-formatROW # 给MySQL实例一个唯一的ID这里设为1。记住这个数字后面会用到。 server_id1 # 可选为binlog保留7天防止磁盘被撑爆 expire_logs_days7注意server_id非常重要在MySQL主从架构中用于唯一标识一个实例。即使你不是主从也要设一个而且不能和后面Canal配置里的slaveId重复。重启MySQL服务让配置生效systemctl restart mysqld验证配置是否成功 登录MySQL命令行mysql -u root -p执行以下命令检查-- 查看binlog是否开启Value应该是ON SHOW VARIABLES LIKE log_bin; -- 查看当前的binlog格式应该是ROW SHOW VARIABLES LIKE binlog_format; -- 查看当前正在写入的binlog文件状态记下File和Position后面可能有用 SHOW MASTER STATUS;你会看到类似这样的结果------------------------------------------------------------------------------- | File | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set | ------------------------------------------------------------------------------- | mysql-bin.000001 | 154 | | | | -------------------------------------------------------------------------------2.2 为Canal创建专属账号出于安全考虑我们不应该直接用root账号。需要创建一个专门给Canal使用的数据库账号并授予它读取binlog的权限。继续在MySQL命令行里执行-- 创建一个用户用户名是canal密码是Canal123456请改成你自己的强密码 -- 注意host是%表示允许从任何IP连接。生产环境建议指定Canal服务器的IP。 CREATE USER canal% IDENTIFIED BY Canal123456; -- 授予这个账号复制从库所需的权限 GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; -- 刷新权限使其立即生效 FLUSH PRIVILEGES;这里的关键权限是REPLICATION SLAVE和REPLICATION CLIENT有了它们Canal才能像从库一样拉取binlog。2.3 准备测试用的数据库和表为了待会儿演示效果我们创建一个简单的数据库和表。-- 创建一个名为canal_test的数据库 CREATE DATABASE canal_test DEFAULT CHARACTER SET utf8mb4; -- 切换到新数据库 USE canal_test; -- 创建一张用户表 CREATE TABLE user ( id bigint(20) NOT NULL AUTO_INCREMENT COMMENT 主键ID, name varchar(50) DEFAULT NULL COMMENT 用户名, age int(11) DEFAULT NULL COMMENT 年龄, email varchar(100) DEFAULT NULL COMMENT 邮箱, update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 更新时间, PRIMARY KEY (id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT用户测试表; -- 插入两条初始数据 INSERT INTO user (name, age, email) VALUES (张三, 25, zhangsanexample.com); INSERT INTO user (name, age, email) VALUES (李四, 30, lisiexample.com);好了数据库这边舞台已经搭好演员数据也已就位。接下来请主角Canal登场。3. 搭建Canal服务端让管道开始工作Canal的服务端部署其实挺简单的主要就是下载、解压、改配置、启动。但配置里的门道不少咱们一步步来。3.1 下载与解压我们去Canal的GitHub Release页面下载最新稳定版。写这篇文章时最新版是1.1.7比原始文章的1.1.5要新修复了一些问题。# 进入一个常用的安装目录比如/usr/local cd /usr/local # 下载deployer包这是服务端部署包 wget https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.deployer-1.1.7.tar.gz # 解压 tar -zxvf canal.deployer-1.1.7.tar.gz # 重命名一下目录方便识别 mv canal.deployer-1.1.7 canal-server cd canal-server解压后你会看到这么几个目录bin/启动、停止脚本就在这。conf/核心所有配置文件都在这里。lib/Canal运行需要的所有Jar包。logs/日志文件会输出到这里排查问题必看。plugin/一些插件比如对接Kafka、RocketMQ的适配器。3.2 核心配置文件详解避坑指南conf目录下文件不少我们重点关注两个canal.properties和instance实例配置。首先看全局配置canal.propertiesvim conf/canal.properties这个文件控制Canal服务本身。我们找到并修改几个关键地方################################################# ######### common argument ############# ################################################# # canal服务绑定的IP默认不配置就行表示所有网卡 # canal.ip # canal服务的端口默认11111客户端就连接这个端口 canal.port 11111 # 这里定义了这个Canal服务里要运行哪些“实例”(instance) # 一个实例可以理解为一个独立的同步任务监听一个MySQL源 # 默认是example我们可以改成自己的比如canal_test canal.destinations canal_test # 实例配置文件的根目录 canal.conf.dir ../conf # 运行模式默认tcp。表示客户端通过TCP直连来获取数据。 # 如果后期流量大可以改为kafka, rocketMQ等让消息先入队列。 canal.serverMode tcp对于初次搭建上面这几项改了就够。文件里还有很多关于性能、HA高可用的配置比如canal.instance.memory.buffer.size内存缓冲区大小可以先保持默认等真正有性能压力时再调优。然后配置实例文件canal.destinations canal_test这一行告诉Canal去../conf目录下找一个叫canal_test的文件夹加载里面的实例配置。默认有个example文件夹我们复制它作为模板。# 复制example文件夹创建我们的实例配置目录 cp -r conf/example conf/canal_test现在来编辑这个实例的核心配置vim conf/canal_test/instance.properties这个文件是告诉Canal你要去连哪个MySQL监听哪些表用什么账号################################################# ## mysql serverId 模拟MySQL从库的ID必须是一个唯一的数字不能和主库的server_id重复 canal.instance.mysql.slaveId 1234 # 连接的主库地址和端口 canal.instance.master.address 127.0.0.1:3306 # 连接主库的账号密码就是我们之前创建的canal账号 canal.instance.dbUsername canal canal.instance.dbPassword Canal123456 # 字符集 canal.instance.connectionCharset UTF-8 # 最重要的配置要监听哪些表 # .*\\..* 是默认值表示所有库的所有表。这是正则表达式。 canal.instance.filter.regex canal_test\\..*关于canal.instance.filter.regex这个过滤规则我多解释几句因为它太常用了canal_test\\..*表示监听canal_test数据库下的所有表。注意正则里点号.需要转义\\.而Java字符串里反斜杠本身也要转义所以成了\\.。如果你想监听多个库可以用逗号隔开canal_test\\..*,order_db\\..*如果只想监听特定表canal_test\\.user,canal_test\\.order排除某些系统表canal.instance.filter.black.regex mysql\\.slave_.*配置到这里一个最基本的Canal服务端就准备好了。它将会连接到本机3306端口的MySQL使用canal账号并监听canal_test库的所有数据变化。3.3 启动服务与查看日志启动命令很简单# 在canal-server目录下执行 bin/startup.sh启动后怎么知道它成功了呢看日志查看Canal服务本身的日志tail -f logs/canal/canal.log你期望看到的关键信息是... INFO c.a.otter.canal.instance.core.AbstractCanalInstance - start successful.... ... INFO c.a.o.canal.deployer.CanalStarter - ## the canal server is running now ......看到the canal server is running now说明服务进程启动OK。查看我们刚配置的canal_test实例的日志tail -f logs/canal_test/canal_test.log这里的信息更重要它反映了Canal和MySQL之间的连接状态。成功的标志是... INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-canal_test ... INFO c.a.o.canal.parse.inbound.mysql.dbsync.DirectLogFetcher - --- begin to find start position, it will be long time for reset or first position ... INFO c.a.o.canal.parse.inbound.mysql.dbsync.DirectLogFetcher - prepare to find start position just show master status ... INFO c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - --- find start position successfully, EntryPosition[includedfalse,journalNamemysql-bin.000001,position154,serverId1,gtid,timestamp0] ...这表示Canal已经成功连接上MySQL并定位到了binlog的起始位置就是我们之前SHOW MASTER STATUS看到的位置。常见启动失败问题连不上MySQL检查canal.instance.master.address、用户名密码、以及MySQL是否允许远程连接如果Canal和MySQL不在同一台机器。权限错误确认给canal账号授予了REPLICATION SLAVE, REPLICATION CLIENT权限。MySQL 8.0加密方式问题如果你用的是MySQL 8.0并且创建用户时没有指定默认会使用caching_sha2_password加密插件旧版本的Canal客户端可能不支持。解决方法就是像我们之前那样创建用户时指定使用mysql_native_password插件。CREATE USER canal% IDENTIFIED WITH mysql_native_password BY Canal123456;服务端跑起来后它就在11111端口上默默等待客户端来连接了。接下来我们写一个客户端程序去“接收”这些变化。4. 编写Canal客户端接收并处理数据变更服务端是“发报机”客户端就是“收音机”。这里我用最经典的Spring Boot项目来演示你可以轻松集成到自己的Java项目中。4.1 项目初始化与依赖引入首先创建一个Spring Boot项目用你喜欢的IDE或者Spring Initializr。在pom.xml中加入Canal客户端依赖dependencies !-- Canal Java客户端 -- dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.7/version !-- 版本尽量与服务端一致 -- /dependency !-- Spring Boot基础依赖 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter/artifactId /dependency !-- 方便打日志 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-logging/artifactId /dependency !-- 测试用 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies4.2 核心消费逻辑实现我们不搞复杂的架构就写一个最直接的、在控制台打印数据变化的客户端。新建一个CanalClientService类。import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.CanalEntry.*; import com.alibaba.otter.canal.protocol.Message; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.InitializingBean; import org.springframework.stereotype.Component; import java.net.InetSocketAddress; import java.util.List; Slf4j Component public class SimpleCanalClient implements InitializingBean { // Canal服务器地址和端口 private static final String CANAL_SERVER_IP 你的Canal服务器IP; // 如果是本机就填127.0.0.1 private static final int CANAL_SERVER_PORT 11111; // 要连接的destination必须和服务端conf/canal.properties里配置的其中一个名字一致 private static final String DESTINATION canal_test; // 每次拉取数据的批量大小 private static final int BATCH_SIZE 1000; Override public void afterPropertiesSet() throws Exception { // 项目启动后在一个单独的线程里运行Canal客户端避免阻塞主线程 new Thread(this::run).start(); } private void run() { // 1. 创建连接器 // 参数Canal服务器地址 Destination 用户名 密码如果canal.properties里没设这里就为空 CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(CANAL_SERVER_IP, CANAL_SERVER_PORT), DESTINATION, , // 用户名默认空 // 密码默认空 ); try { // 2. 建立连接 connector.connect(); // 3. 订阅过滤条件。这里我们订阅canal_test库的所有表。 // 注意这个过滤条件会覆盖instance.properties里的filter.regex通常建议在客户端做更细粒度的过滤。 connector.subscribe(canal_test\\..*); // 4. 回滚到未进行ack确认的位置。如果是第一次启动就从头开始消费。 connector.rollback(); log.info(Canal客户端启动成功开始监听数据变更...); // 5. 无限循环持续拉取消息 while (true) { // 尝试获取指定数量的数据不进行确认getWithoutAck Message message connector.getWithoutAck(BATCH_SIZE); long batchId message.getId(); int size message.getEntries().size(); if (batchId -1 || size 0) { // 没有数据休息一下再试避免空转消耗CPU Thread.sleep(1000); } else { // 有数据开始处理 processEntries(message.getEntries()); } // 6. 处理完一批数据后向服务器确认ack // 确认后服务器才会认为这批数据消费成功下次从新的位置开始发送。 // 如果处理失败可以调用connector.rollback(batchId)回滚下次还会收到这批数据。 connector.ack(batchId); } } catch (Exception e) { log.error(Canal客户端运行异常, e); } finally { // 7. 断开连接 connector.disconnect(); } } private void processEntries(ListEntry entries) { for (Entry entry : entries) { // 跳过事务开始和结束的Entry我们只关心真实的数据行变更 if (entry.getEntryType() EntryType.TRANSACTIONBEGIN || entry.getEntryType() EntryType.TRANSACTIONEND) { continue; } // 只处理ROWDATA类型的数据 if (entry.getEntryType() EntryType.ROWDATA) { try { // 解析出行变更数据 RowChange rowChange RowChange.parseFrom(entry.getStoreValue()); // 获取事件类型INSERT, UPDATE, DELETE EventType eventType rowChange.getEventType(); // 获取表头信息库名、表名 String schemaName entry.getHeader().getSchemaName(); String tableName entry.getHeader().getTableName(); String logFileName entry.getHeader().getLogfileName(); long logFileOffset entry.getHeader().getLogfileOffset(); log.info( 收到变更 [binlog:{}:{}], 库[{}], 表[{}], 事件类型[{}], logFileName, logFileOffset, schemaName, tableName, eventType); // 遍历变更的每一行数据 for (RowData rowData : rowChange.getRowDatasList()) { if (eventType EventType.DELETE) { // 删除操作变更前的数据才是被删除的数据 printColumns(rowData.getBeforeColumnsList(), DELETE); } else if (eventType EventType.INSERT) { // 新增操作变更后的数据才是新增的数据 printColumns(rowData.getAfterColumnsList(), INSERT); } else if (eventType EventType.UPDATE) { // 更新操作变更前是旧数据变更后是新数据 log.info(------- 更新前); printColumns(rowData.getBeforeColumnsList(), null); log.info(------- 更新后); printColumns(rowData.getAfterColumnsList(), UPDATE); } } } catch (Exception e) { log.error(解析binlog数据出错, e); } } } } private void printColumns(ListColumn columns, String eventType) { for (Column column : columns) { // 这里只是简单打印真实场景下你会在这里写业务逻辑 // 比如将数据同步到ES、刷新缓存、发送通知等等。 log.info({} - 列名: {}, 值: {}, 是否更新: {}, eventType, column.getName(), column.getValue(), column.getUpdated()); } } }这段代码虽然不长但包含了Canal客户端最核心的流程连接 - 订阅 - 拉取 - 解析 - 处理 - 确认。我把几个关键点再拎出来说一下connector.subscribe(canal_test\\..*)这里的订阅表达式可以和服务端配置的不一样。最佳实践是服务端配置监听较宽的范围比如整个业务库客户端按需订阅自己关心的表这样更灵活。connector.getWithoutAck(BATCH_SIZE)拉取消息但不确认。拉取后你有时间去处理这批数据处理成功后再手动ack。如果处理中途程序崩溃了下次重启会从上次没有ack的位置重新拉取保证数据至少被消费一次at-least-once。processEntries方法这里是你的业务逻辑入口。现在只是打印你可以在这里把数据转换成JSON发到Kafka或者更新Redis缓存或者写入另一个数据库。4.3 运行与测试启动你的Spring Boot应用。如果一切正常日志里会看到Canal客户端启动成功开始监听数据变更...。现在让我们回到MySQL对测试表做一些操作看看客户端会不会有反应。USE canal_test; -- 1. 插入一条新数据 INSERT INTO user (name, age, email) VALUES (王五, 28, wangwuexample.com); -- 2. 更新一条数据 UPDATE user SET age 31 WHERE name 李四; -- 3. 删除一条数据 DELETE FROM user WHERE name 张三;迅速切换到你的客户端应用日志控制台你应该会看到类似这样的输出 收到变更 [binlog:mysql-bin.000001:2193], 库[canal_test], 表[user], 事件类型[INSERT] INSERT - 列名: id, 值: 3, 是否更新: false INSERT - 列名: name, 值: 王五, 是否更新: false ... 收到变更 [binlog:mysql-bin.000001:2456], 库[canal_test], 表[user], 事件类型[UPDATE] ------- 更新前 UPDATE - 列名: id, 值: 2, 是否更新: false UPDATE - 列名: name, 值: 李四, 是否更新: false UPDATE - 列名: age, 值: 30, 是否更新: true !-- 注意这里isUpdated为true表示这个字段被更改了 ... ------- 更新后 ... 收到变更 [binlog:mysql-bin.000001:2789], 库[canal_test], 表[user], 事件类型[DELETE] DELETE - 列名: id, 值: 1, 是否更新: false ...看到这些恭喜你你的第一个Canal增量数据同步系统已经成功跑起来了。数据库的每一次“风吹草动”你的客户端都了如指掌。5. 进阶配置与生产环境考量上面我们完成了一个最简可用的Demo。但要真正用到生产环境还有一些重要的坑需要提前填上。5.1 高可用与故障恢复单机版的Canal服务端有个问题如果这台机器挂了数据同步就中断了。Canal官方支持基于ZooKeeper的HA模式。修改canal.properties# 启用zk canal.zkServers zk1:2181,zk2:2181,zk3:2181 # 将运行模式改为HA canal.serverMode tcp # 注意HA模式下instance的配置管理方式也会变化修改instance.properties# 启用HA后instance的配置可以托管到ZK实现动态修改和故障转移 canal.instance.global.mode spring canal.instance.global.lazy false canal.instance.global.manager.address ${canal.zkServers} canal.instance.global.spring.xml classpath:spring/file-instance.xml配置HA后可以启动多个Canal Server它们会竞争同一个instance的锁只有一个处于工作状态Active其他处于备用状态Standby。当Active节点故障时Standby节点会迅速接管。5.2 性能调优参数当数据变更非常频繁时默认配置可能会成为瓶颈。有几个关键参数可以调整canal.instance.memory.buffer.size内存环形缓冲区大小。默认16K单位是canal.instance.memory.buffer.memunit默认1KB。所以总内存是 16384 * 1KB 16MB。如果解析速度跟不上Dump速度可以适当调大比如设置为 6553664MB。必须是2的幂。canal.instance.parser.parallel是否开启并行解析对于多核机器和大量表变更的场景开启true能显著提升解析吞吐量。canal.instance.transaction.size事务合并的最大大小。如果单个事务涉及的行数超过这个值Canal会把它拆分成多个消息发送。可以根据业务事务的平均大小调整。5.3 接入消息队列MQ直接TCP连接在客户端不多、网络稳定的情况下没问题。但在大规模、多消费者、需要保证可靠性和削峰填谷的场景下强烈建议将Canal Server接入消息队列。在canal.properties中修改# 将serverMode从tcp改为kafka或rocketMQ canal.serverMode kafka # 配置Kafka相关参数 kafka.bootstrap.servers kafka1:9092,kafka2:9092 canal.mq.topic canal_topic # 按表名动态路由到不同的Kafka分区提升并发度 canal.mq.dynamicTopic .*\\..*这样Canal Server解析出的数据变更事件会直接发到Kafka客户端不再直连Canal Server而是去消费Kafka里的消息。架构解耦可靠性、扩展性都大大增强。5.4 监控与运维线上服务离不开监控。Canal提供了一些方式日志监控密切关注logs/canal_test/canal_test.log看是否有连接异常、解析错误。Metrics监控Canal集成了Dropwizard Metrics可以通过JMX或HTTP端口canal.metrics.pull.port默认11112暴露指标比如接收的binlog事件数、解析成功/失败数、发送到MQ的延迟等。可以接入PrometheusGrafana。meta.dat文件在实例目录下如conf/canal_test这个文件记录了消费的binlog位置。如果客户端消费落后可以对比MySQL的SHOW MASTER STATUS和这里的position判断延迟情况。6. 实战中踩过的坑与解决方案最后分享几个我实际项目中遇到的典型问题希望能帮你少走弯路。坑1数据重复消费或丢失现象客户端处理完数据后没有正确ack或者处理逻辑抛出异常导致ack失败。下次重启客户端又会拉取到同一批数据导致重复消费。极端情况下如果客户端处理完就崩溃且没有ack服务端也不知道该不该推进位置可能造成数据“假丢失”实际在binlog里但客户端没处理。解决确保你的消费逻辑是幂等的。即同一条数据变更消息即使收到多次执行的结果也是一样的。同时做好异常捕获只有业务逻辑真正成功执行后才调用connector.ack(batchId)。坑2同步延迟越来越高现象刚开始实时性很好运行几天后发现数据同步要慢好几分钟。解决检查客户端消费速度是不是你的业务处理逻辑太慢比如同步到ES的批量操作太频繁或者网络延迟高。可以考虑优化处理逻辑或者引入MQ缓冲。调整Canal Server参数如前面提到的增大canal.instance.memory.buffer.size开启canal.instance.parser.parallel。检查MySQL主库压力如果主库本身写压力就很大生成binlog的速度超过了Canal的拉取和解析速度延迟必然增大。需要从数据库层面优化。坑3DDL语句导致客户端报错现象在MySQL里执行了一个ALTER TABLE ADD COLUMN ...客户端解析时抛出异常比如Cannot find column: xxx。解决Canal默认会解析DDL语句。如果你的客户端逻辑强依赖表结构比如用ORM框架做映射DDL变更会导致新旧数据结构不一致。可以在instance.properties中关闭DDL同步canal.instance.filter.druid.ddl false。但更好的做法是在客户端代码里对DDL事件进行特殊处理或者触发一个全量同步流程。坑4大事务阻塞现象业务上执行了一个更新几十万行数据的大事务Canal客户端好像“卡住”了很久没收到新消息。原因Canal默认会等一个完整的事务提交后才把这一批行变更打包成一个Message发给客户端。这是为了保证事务的原子性。解决调整canal.instance.transaction.size参数。如果单个事务的行数超过这个值Canal会把它拆分成多个小Message发送避免一个超大消息阻塞管道。但要注意这破坏了事务的边界你的消费端逻辑需要能处理“部分事务”的情况。搭建和配置Canal的过程就像是在数据库和应用之间铺设一条精准的数据管道。一开始可能会觉得配置项繁多但一旦跑通你会发现它为系统架构带来的解耦和实时能力是非常值得的。从我自己的经验来看从最早的手写定时任务同步到引入Canal再到后来基于CanalMQ构建完整的数据流平台每一步都让系统的实时性和可维护性上了一个台阶。现在遇到需要数据实时流动的场景Cana几乎是我的首选方案。希望这篇从零开始的指南能帮你顺利搭起这条“管道”把数据流动起来。