企业官网建设流程全解析

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.cnf（Windows是my.ini），在[mysqld]段下加三行：

[mysqld] log-bin=mysql-bin binlog-format=ROW server-id=1

• log-bin：开启binlog，后面跟的是日志文件前缀
• binlog-format=ROW：这是canal工作的关键前提。如果设成STATEMENT，binlog里记录的是SQL语句本身，canal解析不出某个字段从什么值变成了什么值；只有ROW模式，binlog里才是每一行前后的具体数据
• server-id：MySQL复制环境里的标识，这个值不能和已有的从库或canal的slaveId冲突，否则复制关系会乱

改完配置文件记得重启MySQL。启动之后，进MySQL执行这两条验证命令：

SHOW VARIABLES LIKE 'log_bin'; SHOW VARIABLES LIKE 'binlog_format';

log_bin的值是ON，binlog_format的值是ROW，这个前置条件就算过了。如果binlog_format显示的是STATEMENT，大概率是配置文件没被正确读取，用一条命令确认一下MySQL实际读取的配置文件路径：

SHOW VARIABLES LIKE 'basedir';

然后检查是不是把配置写到了错误的文件里。

另外，如果你的MySQL版本在5.6以上，可以顺手确认一个参数binlog_row_image，它决定binlog里是否记录完整行镜像。默认值是FULL，也就是变更前后所有字段都记录。如果被改成MINIMAL，canal解析UPDATE时经常拿不到未修改字段的旧值，数据同步场景会非常头痛。保证这一项也是FULL：

SHOW VARIABLES LIKE 'binlog_row_image';

注意：binlog是主从复制的基础设施，很多团队的MySQL虽然没配从库，但binlog一直是开着的。如果你的MySQL已经处于其他主从复制环境里，改配置前务必先确认server-id不要和别人冲突，别为了跑一个demo把现有复制架构搞出问题。

2.2 创建账号和一张测试表

canal要连接MySQL，需要给它的账号授权。权限主要三类：SELECT权限，用于读取表结构；REPLICATION SLAVE权限，用于模拟从库拉取binlog；REPLICATION 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_password：

CREATE 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.port=11111 canal.destinations=example

• canal.port：canal server对外提供服务的TCP端口，客户端连接用的就是这个端口
• canal.destinations：声明要加载哪些实例，多个实例用逗号分隔。比如canal.destinations=example1,example2，就必须在conf下建example1和example2两个目录

文件里还有很多关于内存模式、MQ模式、持久化模式的配置，如果是第一次做入门实操，不建议动它们。保持默认就好，先把链路跑通，再去研究那些高级参数，会顺畅得多。

3.3 instance实例配置要点

打开conf/example/instance.properties，逐个调整数据库相关的配置：

canal.instance.mysql.slaveId=54321 canal.instance.master.address=127.0.0.1:3306 canal.instance.dbUsername=canal canal.instance.dbPassword=canal canal.instance.connectionCharset=UTF-8 canal.instance.filter.regex=test\\..* canal.instance.filter.blackRegex=

逐项说明：

• canal.instance.mysql.slaveId：canal伪装成MySQL从库时使用的slaveId。如果MySQL已经有其他从库，这个值不能和它们一样。默认54321基本够用
• canal.instance.master.address：MySQL的地址和端口
• canal.instance.dbUsername和dbPassword：就是前面创建的那个canal账号
• canal.instance.connectionCharset：连接MySQL时用的字符集，写UTF-8
• canal.instance.filter.regex：订阅规则，重要。这个正则表达式决定了canal解析哪些库表的数据。测试环境我设成test\..，表示订阅test库下所有表。这里有个细节，正则里的点号是特殊字符，需要转义，所以中间有两个反斜杠。如果写成test.，它表示的是test后面跟任意字符，testxxx这种库也会被匹配进去，虽然demo影响不大，但容易误导。订阅所有库所有表则写.\..
• canal.instance.filter.blackRegex：黑名单，排除某些库表。比如我要排除test库里的log表，可以写test\.log，demo阶段先留空

filter.regex是正宗的Java正则，多个不同的订阅范围可以用逗号并列，比如test\..*,biz\..*表示同时订阅test库和biz库。掌握了这一层，你就能更灵活地控制接收范围。

3.4 启动和验证

配置改完后，启动canal server。Linux环境执行：

bin/startup.sh

Windows环境执行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> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.5</version> </dependency>

canal.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事件看afterColumns，DELETE事件看beforeColumns，UPDATE事件两者都有，一一打印出来方便对比。

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(List<Column> 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, 'zhangsan@example.com');

控制台几秒内会打印出类似下面的内容：

================> binlog[mysql-bin.000001:1234] , name[test,user] , eventType : INSERT (after) id : 1 (after) name : 张三 (after) age : 25 (after) email : zhangsan@example.com (after) create_time : 2025-01-01 12:00:00

接下来再执行一条UPDATE和DELETE，观察输出差异。UPDATE会同时打印before和after，DELETE只有before。通过这种逐列打印，能很直观地理解canal的列值模型。

4.4 对输出结果的解读

拿到数据之后，要能看懂每个字段的含义。第一行的binlog[mysql-bin.000001:1234]，前半部分是binlog文件名，后半部分是日志偏移量，它代表这条变更在binlog里的确切位置。之后是库名、表名、事件类型。后面的每一行是列名和值。

列数据里其实还有两个demo中没有细化的属性：isNull表示该字段是否为NULL，updated表示该字段在本次变更中是否真的修改过。正式业务里，这两个属性经常决定下游要不要处理。比如缓存更新场景，如果某列没变，可以跳过缓存刷新；如果是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是ROW，binlog_row_image是FULL。曾经遇到一个环境，binlog格式是STATEMENT，canal能启动能消费，但拿到的数据永远是空的或者解析不出字段内容。先把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这些更贴近业务的场景。先把这篇的基础链路摸熟，后面的扩展就是水到渠成的事。