1. 从数据同步的痛点说起为什么我们需要Canal在任何一个稍具规模的互联网业务里数据同步都是一个绕不开的“老大难”问题。想象一下这样的场景你的核心交易数据都存放在MySQL数据库中但你的运营后台需要实时报表你的推荐系统需要最新的用户行为你的搜索服务需要商品信息的即时更新。最直接的做法是什么让各个下游系统都去轮询查询主库这会给主库带来巨大的、不必要的读压力而且延迟高实时性差。另一种常见的做法是业务代码在写数据库的同时再写一份消息到消息队列如Kafka。这看似解决了实时性问题但引入了新的复杂度——业务逻辑变得臃肿需要维护两套写入逻辑并且无法保证事务一致性万一消息发送失败数据就不同步了。有没有一种方案能像数据库的“监听器”一样在数据发生变更增删改的第一时间无侵入地、准确地捕获到这个变更并通知给所有关心它的下游系统这就是Canal诞生的背景。Canal这个名字取自“管道”其核心目标就是扮演MySQL数据库的“变更数据捕获”Change Data Capture CDC管道。它通过模拟MySQL Slave的交互协议将自己伪装成一个数据库的从库然后向主库请求增量日志binlog解析这些日志最终将结构化的变更数据比如某张表的主键ID123的记录被更新了某个字段推送出去。整个过程对业务代码完全透明你只需要像往常一样操作数据库Canal就会在背后默默帮你把变更事件同步到任何你需要的地方。我第一次接触Canal是在一个用户画像实时更新的项目里。当时我们尝试过多种方案最终Canal以其稳定性和对MySQL原生协议的支持脱颖而出成为了我们数据流架构的基石。今天我就结合自己多年的踩坑和实战经验为你拆解Canal从部署、配置到高级应用的每一个细节。2. Canal的工作原理深度拆解它如何“伪装”成一个从库要真正用好Canal避免在配置和排错时一头雾水必须理解其底层的工作原理。这不仅仅是知道“它能同步数据”更要明白它是“如何同步”的。2.1 MySQL主从复制Canal的灵感来源Canal的整个设计思想都源于MySQL自身的主从复制Replication机制。简单回顾一下这个过程主库Master将所有数据变更DML、DDL以事件event的形式记录到二进制日志Binary Log 即binlog中。从库Slave启动一个I/O线程向主库发起连接并发送一个COM_BINLOG_DUMP命令告诉主库“请从某个binlog文件的某个位置开始把之后的日志发给我”。主库响应主库的Binlog Dump线程接收到请求后就开始从这个指定位置读取binlog事件并通过网络发送给从库的I/O线程。从库写入与重放从库的I/O线程将收到的事件写入本地的中继日志Relay Log然后由另一个SQL线程读取中继日志中的事件并在从库上重新执行Replay一遍从而实现数据同步。Canal的核心就是把自己完美地“伪装”成这个从库。它实现了MySQL Slave的通信协议能够像真正的从库一样向主库发起COM_BINLOG_DUMP请求从而源源不断地获取到原始的binlog流。2.2 Canal的三层架构Server、Instance与Client理解了伪装原理我们再来看Canal自身的架构它清晰地分为三层职责分明Canal Server可以理解为一个服务容器。它负责启动、管理和承载一个或多个Canal Instance。在生产环境中我们通常部署的就是Canal Server。Canal Instance这是实际工作的核心单元。一个Instance对应一个数据源一个MySQL数据库实例负责完成与MySQL的链接、binlog订阅、解析和存储等全套动作。每个Instance有自己独立的配置文件。Canal Client客户端。用户需要自己编写或使用现有的Client程序连接到Canal Server订阅指定的Instance然后消费解析后的数据变更事件。Canal官方提供了Java客户端社区也有其他语言的适配。这种架构设计带来了很好的灵活性。比如你可以在一台Canal Server上为多个不同的MySQL数据库实例配置多个Instance也可以让多个不同的Client如一个用于同步到ES一个用于同步到HBase同时订阅同一个Instance互不干扰。2.3 数据流转的全链路从Binlog到你的业务代码让我们跟随一条UPDATE user SET name‘张三’ WHERE id1;的SQL语句看看它在Canal中是如何流转的业务应用在MySQL主库上执行了这条更新语句。MySQL主库将此次更新事件包含表名user、操作类型UPDATE、修改前数据[id1, name‘李四’]、修改后数据[id1, name‘张三’]等信息写入binlog文件。binlog有多种格式最常用的是ROW格式它会记录每一行数据修改的细节这也是Canal推荐和依赖的格式。Canal Instance伪装连接Instance启动时以其配置的用户名密码连接MySQL并执行SHOW MASTER STATUS获取当前binlog位置然后发送COM_BINLOG_DUMP命令开始拉取。协议解析接收主库发来的二进制数据流按照MySQL网络协议进行拆包和解码得到一个个原始的binlog事件对象。事件过滤与解析根据Instance配置文件中的过滤规则如只监听db1.user表对事件进行初步过滤。然后调用Canal Parser模块默认是使用开源的dbsync项目解析库对binlog事件进行深度解析。解析器需要知道表结构Schema才能将二进制数据还原成有意义的字段值。Canal通过两种方式获取表结构运行时查询在解析到某个表的事件时立即去MySQL查询一次SHOW CREATE TABLE。这种方式简单但每次解析都有查询开销和网络延迟。内存管理Canal会在内存中维护一份表结构的快照Memory TableMeta。当监听到ALTER TABLE等DDL事件时会主动更新这个快照。这种方式效率高是推荐的生产模式。事件存储解析后的结构化数据Canal内部称为Entry会被存储起来。Canal支持多种存储方式最常用的是Memory和File。Memory模式性能极高但宕机或重启会丢失内存中未消费的事件File模式将事件持久化到本地磁盘如${canal.dir}/memory提供了数据可靠性。投递就绪存储后的事件等待Client来拉取。Canal Client连接与订阅Client连接到Canal Server指定要订阅的Instance和起始位置可以指定binlog文件名位点也可以指定时间戳或者直接订阅最新的。批量拉取Client调用get或getWithoutAck方法从Instance的存储中拉取一批Entry。业务处理Client遍历这批Entry每个Entry对应一个RowChange对象里面包含了完整的变更数据。你的业务代码就在这里可以将数据转换成JSON写入Kafka或者更新到Elasticsearch的索引中。确认消费处理成功后Client调用ack()方法告知Canal Server这批数据已经成功消费Server可以清理这部分存储对于Memory模式就是释放内存。如果处理失败可以调用rollback()让Server下次重新投递这批数据。注意这里有一个非常重要的设计——Canal Server本身不保证At-Least-Once或Exactly-Once语义。它采用类似消息队列的ack机制消息被Client拉取后在收到ack前会处于“未确认”状态。如果Client崩溃未ack的消息会被重新投递。这意味着你的Client业务逻辑必须实现幂等性即同一条变更消息即使被重复消费也不会导致最终数据错误。这是使用Canal构建可靠数据管道的第一原则。3. 从零开始搭建与配置让Canal跑起来理论讲得再多不如亲手搭一遍。下面我将以最常用的1.1.7版本为例带你完成一次标准的单机部署和基础配置。假设我们的目标是从一个名为192.168.1.100:3306的MySQL同步test库的user表到Kafka。3.1 环境准备MySQL与Canal Server第一步MySQL端配置Canal要伪装成从库MySQL主库必须开启binlog并且授权一个专门的账号给Canal。开启binlog编辑MySQL配置文件my.cnf通常在/etc/mysql/或/etc/my.cnf确保有以下配置[mysqld] # 启用binlog并设置文件名前缀 log-binmysql-bin # 设置server-id在一个复制拓扑中必须唯一这里设为1 server-id1 # 强烈推荐使用ROW格式这是Canal解析数据变更细节的基础 binlog-formatROW # 可选指定binlog过期时间避免磁盘占满 expire_logs_days7修改后重启MySQL服务。创建Canal专用账号登录MySQL执行以下SQL。这个账号需要REPLICATION SLAVE和REPLICATION CLIENT权限这是从库拉取binlog所必需的。同时为了获取表结构还需要对应数据库的SELECT权限。CREATE USER canal% IDENTIFIED BY canal_password; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; -- 如果Canal Server和MySQL不在同一机器‘%’允许任何主机连接。生产环境建议限制为Canal Server的IP。 FLUSH PRIVILEGES;验证配置执行SHOW MASTER STATUS;应该能看到File和Position信息证明binlog已开启。第二步下载与部署Canal Server从Canal的GitHub Release页面下载deployer版本如canal.deployer-1.1.7.tar.gz。解压到安装目录例如/opt/canal。tar -zxvf canal.deployer-1.1.7.tar.gz -C /opt/canal目录结构如下/opt/canal ├── bin/ # 启动停止脚本 ├── conf/ # 配置文件 │ ├── canal.properties # Server全局配置 │ └── example/ # 一个Instance配置示例目录 │ └── instance.properties ├── lib/ # 依赖库 └── logs/ # 日志目录3.2 核心配置文件详解避开第一个大坑Canal的配置是新手最容易出错的地方。我们逐一拆解两个核心文件。conf/canal.properties- Server全局配置这个文件控制Canal Server的整体行为我们重点关注以下几项# Canal Server的ID单机部署可不用改集群部署时必须不同 canal.id 1 # Canal Server监听的IP和端口客户端Canal Client将连接这个地址 canal.ip 192.168.1.200 canal.port 11111 # Instance列表多个用逗号分隔。这里定义了一个名为example的Instance。 # 它会自动加载conf/example/目录下的instance.properties作为配置。 canal.destinations example # Instance配置的自动扫描间隔毫秒 canal.conf.scan.interval 5 # Instance存储模式。memory-内存模式file-文件模式。生产环境追求可靠性建议用file。 canal.instance.global.mode memory # 如果上面是file这里定义文件存储的路径 canal.instance.global.spring.xml classpath:spring/file-instance.xml # 全局的解析器类型默认即可 canal.instance.parser.parallel true对于初次使用通常只需要确认canal.destinations和网络配置即可。conf/example/instance.properties- Instance实例配置这是重头戏它定义了Canal如何连接你的MySQL以及同步哪些数据。################################################# ## mysql serverId 链接信息 ################################################# # 配置一个在MySQL主从拓扑中唯一的slaveId不能与任何其他从库包括真实的和其他Canal Instance冲突。 canal.instance.mysql.slaveId 1234 # 数据库地址、端口、用户名、密码刚才创建的 canal.instance.master.address 192.168.1.100:3306 canal.instance.dbUsername canal canal.instance.dbPassword canal_password # 字符集建议与数据库保持一致 canal.instance.connectionCharset UTF-8 ################################################# ## 订阅的库表过滤规则 ################################################# # 1. 所有表.*\\..* 或 .* (不推荐数据量巨大) # 2. 同步test库的所有表test\\..* # 3. 同步test库的user表test.user # 4. 同步多个规则用逗号分隔test.user, order.t_order_detail canal.instance.filter.regex test\\..* # 黑名单过滤优先级高于白名单。一般不用。 # canal.instance.filter.black.regex ################################################# ## 表结构元数据存储与获取方式 ################################################# # 关闭运行时表结构查询启用内存表结构管理。这是性能关键 canal.instance.tsdb.enable true # 内存表结构快照的持久化文件路径 canal.instance.tsdb.dir ${canal.file.data.dir:../conf}/${canal.instance.destination:} ################################################# ## 存储与消费位点控制 ################################################# # 存储模式继承自全局配置也可在此覆盖。memory/file canal.instance.memory.batch.mode MEMORY # 消费位点信息binlog文件名位置的持久化文件。 # 这个文件非常重要它记录了Canal消费到了哪个位置重启后从此处继续。 canal.instance.detecting.enable false canal.instance.detecting.sql select 1这里最关键的几个配置是slaveId必须唯一、数据库连接信息、过滤规则filter.regex以及tsdb.enable true。很多人在测试时发现Canal延迟高很可能就是因为没开启tsdb每次解析都要去查一次表结构。3.3 启动、验证与基础客户端消费启动Canal Servercd /opt/canal sh bin/startup.sh查看日志确认无报错且看到Canal Server started successfully字样tail -f logs/canal/canal.log查看Instance日志tail -f logs/example/example.log你应该能看到类似以下信息表示Canal已成功连接到MySQL并开始拉取binlog... start successful.... ... subscribe filter change to .*\\..* ... prepare to find start position ... ... find start position successfully ...编写一个最简单的Java客户端进行测试 引入Canal Client的Maven依赖dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.7/version /dependency编写消费代码public class SimpleCanalClient { public static void main(String[] args) { // 1. 创建连接 CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(192.168.1.200, 11111), example, // destination对应instance名称 , ); connector.connect(); // 2. 订阅过滤规则可以覆盖instance.properties中的配置 connector.subscribe(test\\..*); // 3. 回滚到未ack的位置如果是新的消费则从server保存的位点开始 connector.rollback(); while (true) { // 4. 获取指定数量的数据100条或等待5秒 Message message connector.getWithoutAck(100, 5000L, TimeUnit.MILLISECONDS); long batchId message.getId(); if (batchId ! -1 !message.getEntries().isEmpty()) { for (CanalEntry.Entry entry : message.getEntries()) { // 只处理ROWDATA类型的数据变更 if (entry.getEntryType() CanalEntry.EntryType.ROWDATA) { CanalEntry.RowChange rowChange; try { rowChange CanalEntry.RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException(解析数据错误, data: entry.toString(), e); } CanalEntry.EventType eventType rowChange.getEventType(); System.out.println(String.format( binlog[%s:%s], name[%s,%s], eventType: %s, entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(), entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType)); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { // 判断是INSERT/UPDATE/DELETE打印变更前后的数据 if (eventType CanalEntry.EventType.DELETE) { printColumn(rowData.getBeforeColumnsList()); } else if (eventType CanalEntry.EventType.INSERT) { printColumn(rowData.getAfterColumnsList()); } else { // UPDATE System.out.println(------- 更新前); printColumn(rowData.getBeforeColumnsList()); System.out.println(------- 更新后); printColumn(rowData.getAfterColumnsList()); } } } } // 5. 确认消费成功 connector.ack(batchId); } else { // 没拉到数据稍作休息 try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } } private static void printColumn(ListCanalEntry.Column columns) { for (CanalEntry.Column column : columns) { System.out.println(column.getName() : column.getValue() update column.getUpdated()); } } }运行这个客户端然后在MySQL的test.user表里插入或更新一条数据你将在控制台看到详细的变更信息输出。至此一个最基本的Canal同步链路就打通了。4. 生产级部署与高可用架构设计单机版的Canal只能用于测试和学习。一旦上线我们必须考虑可靠性、性能和可维护性。下面分享几种常见的生产级架构。4.1 单机多Instance部署这是最简单的一种生产模式。在一台配置较高的服务器上部署一个Canal Server但为其配置多个Instance每个Instance负责同步一个独立的MySQL实例或数据库。只需要在canal.properties中配置canal.destinations instance1,instance2,instance3并在conf/目录下创建对应的instance1/,instance2/等文件夹各自放置自己的instance.properties配置文件即可。优点部署简单资源复用率高。缺点存在单点故障。如果这台Canal Server宕机所有数据同步都会中断。适用场景对数据同步实时性要求不是极高且有较短时间恢复能力的业务。4.2 基于ZooKeeper的HA部署这是Canal官方推荐的高可用方案。核心思想是利用ZooKeeperZK进行集群管理和消费位点的集中式存储。架构角色Canal Server集群部署多台Canal Server它们启动后会向ZK注册临时节点争抢同一个Instance的running锁。最终只有一台Server能抢到锁成为该Instance的“工作节点”负责真正的数据拉取和解析。其他Server作为“备用节点”待命。ZooKeeper集群作为协调者负责维护Server的存活状态、Instance的工作权选举以及存储所有Client的消费位点。Canal Client启动时从ZK获取当前活跃的Server地址进行连接。当主Server宕机ZK上的running锁释放备用Server会立即抢占并开始工作。Client会感知到连接断开并重新从ZK获取新的Server地址进行重连从而实现故障转移。配置关键点修改所有Canal Server的canal.properties启用ZK模式# 启用集群模式 canal.zkServers zk1:2181,zk2:2181,zk3:2181 # 可选在ZK上的根路径 canal.zookeeper.flush.period 1000每个Instance的instance.properties中位点存储模式要改为ZKcanal.instance.global.spring.xml classpath:spring/default-instance.xml # 注意在HA模式下memory模式不再安全应使用file或mixed canal.instance.memory.batch.mode MemsizeClient端连接Client需要使用CanalConnectors.newClusterConnector来连接ZK集群而不是直接连某台Server。CanalConnector connector CanalConnectors.newClusterConnector( Lists.newArrayList(zk1:2181, zk2:2181, zk3:2181), example, , );优点实现了Server层的自动故障转移提高了服务可用性。注意点ZK本身需要维护增加了架构复杂度。故障切换时可能会有秒级的同步中断和数据重复因为新的Server需要从ZK记录的位点开始拉取这个位点可能略旧于原Server内存中已解析但未持久化的数据。4.3 基于RocketMQ/Kafka的解耦架构在上述HA架构中Canal Client直接连接Canal Server。这要求Client必须保持高可用并且消费速度要能跟上Server的解析速度否则容易造成Server端内存堆积。更优雅的做法是引入一个消息队列MQ进行解耦。架构流程Canal ServerAdapter或Client解析出数据后直接将其投递到RocketMQ或Kafka的Topic中。下游的各种消费服务同步到ES、刷新缓存、计算实时指标等再订阅这个Topic进行消费。Canal的MQ模式从1.1.x版本开始Canal官方提供了canal-server直接投递到MQ的能力通过canal.properties中的canal.serverMode kafka或rocketMQ进行配置。更常见的做法是使用独立的canal-adapter或canal-client项目它们作为Canal Server的客户端拉取数据并转换成指定格式如JSON后写入MQ。优势解耦Canal Server只负责捕获和解析不关心下游是谁、消费速度如何。下游服务可以随时扩容、下线或新增。缓冲MQ天然具备削峰填谷的能力可以应对源库的突发流量和下游消费的不均衡。多订阅一份数据发布到MQ可以被多个不同的消费者组重复消费轻松实现“一源多供”。数据回溯MQ通常具备多日的数据保留能力方便下游服务出错时重新消费。在实际项目中我最为推荐的架构是Canal Server (HA集群) Kafka 多个下游消费者。这几乎成为了现代实时数据栈的标准配置之一。5. 高级特性与实战避坑指南掌握了基础部署和架构我们来看看Canal的一些高级特性和那些“只有踩过才知道”的坑。5.1 DDL语句的同步处理Canal默认会解析并传递DDL语句如CREATE TABLE,ALTER TABLE,DROP TABLE。这对于需要实时维护下游数据表结构的场景非常有用。在解析到的Entry中EntryType会是DDL其storeValue中存储的就是原始的SQL字符串。坑点一表结构元数据不一致如果下游系统如ES、HBase的表结构/映射需要随MySQL变更而自动变更处理DDL会非常复杂。一个常见的做法是在Client端过滤掉DDL事件然后通过人工或自动化脚本在两边执行变更。更高级的做法是解析DDL SQL转换成下游系统的DSL如Elasticsearch的Put Mapping API再执行但这实现成本很高。坑点二DDL导致位点跳跃在MySQL中一个DDL语句尤其是涉及表锁的可能会产生一个非常大的binlog事件或者在某些场景下如使用ROW格式binlog_row_imageFULL时ALTER TABLE后紧接着的DML事件其行数据格式可能已经改变。Canal的解析器需要能正确处理这种变化。务必确保canal.instance.tsdb.enable true让Canal在内存中及时更新表结构否则可能在DDL后解析出错。5.2 消费位点管理重置与回溯消费位点是Canal可靠性的核心。它保存在conf/example/meta.dat文件单机模式或ZooKeeperHA模式中。手动重置位点当你想从头消费binlog或者消费到某个指定时间点的数据时就需要重置位点。停止Canal Instance。删除位点文件meta.dat或清除ZK上对应的节点。修改instance.properties指定新的起始位点# 指定具体的binlog文件和位置 canal.instance.master.journal.name mysql-bin.000001 canal.instance.master.position 100 # 或者指定时间戳Canal会查找该时间点之后第一个binlog事件的位置 canal.instance.master.timestamp 1621234567000启动Instance。注意时间戳定位是一个近似值Canal会找到该时间点之后的第一个事件开始消费可能丢失这个时间点附近的部分数据。Client端控制位点在创建CanalConnector时可以传入一个CanalEntry.Position对象来指定起始位置这比修改配置文件更灵活常用于按需回溯数据的场景。5.3 性能调优与监控当同步的表数量多、数据变更频繁时性能问题就会浮现。提升解析性能canal.instance.parser.parallel true开启多线程并行解析充分利用多核CPU。这是最重要的性能开关。canal.instance.parser.parallelThreads设置并行解析线程数建议设置为CPU核心数。合理设置canal.instance.filter.regex只订阅必要的表避免解析无关的binlog事件这是最有效的优化手段。提升传输性能canal.instance.memory.batch.mode和canal.instance.memory.batch.size调整获取和发送的批次大小。增大批次可以减少网络交互次数但会增加内存占用和延迟。需要在吞吐量和实时性之间权衡。使用MQ模式将数据推到Kafka等高性能MQ彻底解决Server和Client之间的传输瓶颈。监控Canal通过metrics模块暴露了丰富的JMX指标如canal_instance_parse_time解析耗时、canal_instance_get_time拉取耗时、canal_instance_put_time存储耗时以及最重要的canal_instance_binlog_latencybinlog延迟即最新消费到的binlog时间与当前时间的差值。可以将这些指标接入PrometheusGrafana建立监控大盘实时掌握同步链路的健康状况。5.4 常见问题排查心法Canal启动后无数据同步检查连接查看example.log确认是否成功连接MySQL。常见问题是网络不通、防火墙、账号权限不足缺少REPLICATION SLAVE权限。检查位点确认meta.dat中的位点是否远落后于当前位点SHOW MASTER STATUS。如果位点差太大Canal需要追很久的binlog初期可能看不到数据。检查过滤规则确认filter.regex是否正确匹配了你的库表。可以用.*\\..*全库同步测试一下。检查binlog格式必须是ROW格式。执行SHOW VARIABLES LIKE ‘binlog_format’;确认。同步延迟高binlog_latency持续增长下游消费慢这是最常见原因。检查Canal Client或下游MQ消费者的处理速度。可能是业务逻辑复杂、数据库/ES写入慢、网络延迟等。Canal Server性能瓶颈观察服务器CPU、内存、IO。开启并行解析检查是否订阅了过多无关表。网络问题MySQL与Canal Server之间或Canal Server与下游之间网络带宽不足或延迟高。解析错误日志中出现parse row data failed等异常表结构不一致极有可能是tsdb.enable未开启且MySQL表结构发生了变更如增加了一个有默认值的列。Canal用旧的表结构去解析新格式的binlog导致失败。开启tsdb并重启Instance通常能解决。不支持的字段类型Canal的解析器可能不支持某些非常用或新版本的MySQL字段类型。检查日志中的具体错误信息。数据重复消费根本原因是Client处理了消息但未能成功ack。可能是Client处理逻辑抛出异常、进程崩溃、或网络问题导致ack请求未到达Server。解决方案如前所述下游业务逻辑必须设计为幂等。例如基于数据库主键进行upsert操作或者使用消息的唯一ID如binlog位点在消费端做去重。在我经历的一个电商大促项目中就曾因为下游ES集群写入性能达到瓶颈导致Canal同步延迟从几百毫秒飙升到几分钟。当时的应急方案是临时增加ES索引的分片数并扩容ES节点。长远来看我们引入了Kafka作为缓冲层并设置了监控告警当延迟超过阈值时自动扩容下游消费者这才从根本上解决了问题。这个经历让我深刻体会到数据同步从来不是“配置好就能一劳永逸”的事情它是一套需要持续监控和优化的活系统。
