1. 项目概述为什么我们需要Canal在数据驱动的业务场景里我们经常遇到一个经典难题如何将业务数据库比如MySQL中的数据变更实时、可靠地同步到其他系统无论是更新搜索引擎的索引、刷新缓存、还是驱动实时数仓的ETL流程传统基于定时查询的“拉取”模式都存在延迟高、对源库压力大、可能漏数据等问题。这时候一个基于数据库增量日志解析的“订阅”模式工具就显得至关重要而Canal正是这个领域的佼佼者。简单来说Canal扮演了一个“数据库变更监听器”的角色。它伪装成MySQL的从库Slave向主库Master发起一个Binlog Dump请求。主库会将其产生的二进制日志Binlog推送给Canal。Canal拿到这些原始的、记录着所有数据变更增删改的日志后进行解析、过滤和转换最终将结构化的变更数据例如一条UPDATE语句影响了哪张表、哪行数据、变更前后的值是什么推送出去。下游的Kafka、RocketMQ、Elasticsearch等系统消费这些消息就能实现近乎实时的数据同步。我之所以花时间折腾Canal的部署是因为在微服务架构和数据分析项目中它几乎是解耦业务库与数据消费方的标准答案。相比于自己写代码轮询数据库用Canal更优雅、更高效也更能保证数据的一致性。接下来我会结合多次在生产环境和测试环境部署的经验从零开始拆解Canal的安装、配置、核心原理以及那些容易踩坑的细节。2. 环境准备与前置条件解析部署Canal之前必须把地基打牢。很多部署失败的问题根源都在于环境没配置对。这一部分我们详细梳理每个前置条件。2.1 MySQL数据库配置Canal的工作原理决定了它对MySQL的配置有强依赖。你的MySQL必须开启二进制日志Binlog并且格式必须是ROW模式。1. 检查并修改MySQL配置my.cnf或my.ini你需要确保以下配置项已设置。通常它们位于MySQL服务器的配置文件中。[mysqld] # 启用Binlog并指定基础名称。这是必须的。 log-binmysql-bin # 为Binlog文件设置一个前缀这里使用mysql-bin binlog-formatROW # 必须为ROW模式Canal才能解析出行级变更前后的完整数据 server_id1 # 设置一个唯一的服务器ID对于Canal伪装成从库至关重要注意binlog-formatROW是核心。STATEMENT或MIXED格式下Canal无法可靠地获取变更后的数据值。修改此配置后通常需要重启MySQL服务才能生效。2. 创建Canal专用数据库账号Canal需要连接MySQL来拉取Binlog这个账号需要特定的权限。绝对不要使用root账号遵循最小权限原则。登录MySQL后执行以下SQL语句-- 创建一个用户用户名canal密码canal。生产环境请使用强密码。 CREATE USER canal% IDENTIFIED BY canal; -- 授予必要的权限 GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; -- 刷新权限使设置生效 FLUSH PRIVILEGES;权限说明SELECT: 需要读取information_schema库中的元数据信息。REPLICATION SLAVE: 核心权限允许Canal以从库身份连接并接收Binlog事件。REPLICATION CLIENT: 允许使用SHOW MASTER STATUS等命令查看主库状态。3. 验证配置是否生效配置完成后通过MySQL客户端进行验证-- 查看Binlog格式和状态 SHOW VARIABLES LIKE binlog_format; -- 输出应为 ROW SHOW VARIABLES LIKE log_bin; -- 输出应为 ON SHOW MASTER STATUS; -- 查看当前正在写入的Binlog文件及位置确保有输出2.2 Java运行环境部署Canal服务端是用Java编写的所以需要JDK。推荐使用JDK 8或JDK 11这两个是经过广泛验证的稳定版本。1. 安装OpenJDK在Linux系统上通常使用包管理器安装# 对于CentOS/RHEL系统 sudo yum install -y java-1.8.0-openjdk-devel # 对于Ubuntu/Debian系统 sudo apt update sudo apt install -y openjdk-8-jdk安装后检查版本java -version # 应输出类似openjdk version 1.8.0_4022. 配置JAVA_HOME环境变量虽然运行简单命令可能不需要但为了一些脚本和工具能正常工作最好设置一下。# 查找Java安装路径 readlink -f $(which java) # 通常路径类似 /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.402.b07-2.el8.x86_64/jre/bin/java # 其上级目录的上级是JAVA_HOME例如 /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.402.b07-2.el8.x86_64 # 编辑profile文件例如 ~/.bashrc 或 /etc/profile echo export JAVA_HOME/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.402.b07-2.el8.x86_64 ~/.bashrc echo export PATH$JAVA_HOME/bin:$PATH ~/.bashrc # 使配置生效 source ~/.bashrc # 验证 echo $JAVA_HOME2.3 获取Canal部署包Canal的发布包可以从其GitHub的Release页面下载。这里我们选择当前广泛使用的稳定版本canal.deployer-1.1.7。# 进入一个工作目录例如 /opt cd /opt # 下载Canal部署包 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 # 解压后会得到一个 canal.deployer-1.1.7 目录可以重命名为 canal 方便使用 mv canal.deployer-1.1.7 canal cd canal解压后的目录结构如下你需要重点关注几个目录canal/ ├── bin/ # 启动停止脚本 ├── conf/ # 配置文件目录核心 │ ├── canal.properties # Canal Server全局配置 │ └── example/ # 一个实例Instance的配置样例 │ ├── instance.properties # 实例配置连接哪个MySQL等 │ └── ... ├── lib/ # 依赖的Jar包 ├── logs/ # 日志目录 └── plugin/ # 插件目录3. Canal服务端核心配置详解配置文件是Canal的灵魂理解每一个关键配置项是部署成功和稳定运行的基础。我们主要修改两个文件canal.properties和instance.properties。3.1 全局配置canal.properties这个文件配置Canal服务本身的运行参数。对于初次部署我们重点关注以下几项# conf/canal.properties # Canal Server的工作模式默认为tcp。我们先用tcp后续可以改为kafka/rocketmq等将解析结果直接投递到MQ。 canal.serverMode tcp # TCP服务的监听端口客户端如Canal Adapter或你自己写的客户端通过这个端口连接Canal获取数据。 canal.port 11111 # Canal实例的配置目录。默认指向conf/下的子目录每个子目录代表一个实例。 canal.destinations example # 这里定义了一个名为“example”的实例。你可以定义多个用逗号分隔如 example,test1,test2 # 实例的配置自动扫描周期毫秒用于热加载配置。 canal.auto.scan true canal.auto.scan.interval 5000 # Canal数据持久化模式。推荐使用file将解析位点消费进度持久化到本地文件防止重启后重复消费或丢失数据。 canal.instance.global.mode memory canal.instance.global.lazy false canal.instance.global.manager.address ${canal.conf.dir} canal.instance.global.spring.xml classpath:spring/file-instance.xml # 注意上面这四行是配置管理方式对于简单的file模式通常使用内置的spring/file-instance.xml即可它会使用memory内存模式管理元数据但位点信息会写入conf/example/meta.dat文件。 # 网络参数根据实际情况调整 canal.instance.network.receiveBufferSize 16384 canal.instance.network.sendBufferSize 16384 canal.instance.network.soTimeout 30000实操心得canal.serverMode一开始用tcp是最简单的便于测试。但在生产环境为了解耦和保证数据不丢失通常会设置为kafka或rocketmq让Canal Server直接对接消息队列。canal.destinations支持多实例这意味着一个Canal服务可以同时监听多个MySQL数据库或者同一个库的不同逻辑非常灵活。3.2 实例配置instance.properties这个文件定义了一个具体的同步任务监听哪个MySQL、同步哪些表、从哪里开始同步等。文件位于conf/example/目录下。# conf/example/instance.properties ################################################# ## 1. MySQL数据源配置 ################################################# # 数据库地址主库地址 canal.instance.master.address127.0.0.1:3306 # 前面创建的Canal账号 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal # 字符集必须和MySQL服务端配置一致否则中文乱码 canal.instance.connectionCharsetUTF-8 # 默认连接的数据库解析Binlog时会自动切换这里可以设一个存在的库 canal.instance.defaultDatabaseNametest ################################################# ## 2. 订阅过滤规则核心 ################################################# # 1. 所有库所有表.*\\..* # 2. 同步test库的所有表test\\..* # 3. 同步test库的user表和order表test.user,test.order # 4. 同步test库下以t_开头的所有表test\\.t_.* canal.instance.filter.regex.*\\..* # 这个正则表达式是重中之重。.*\\..*表示所有库所有表。生产环境一定要根据业务需要精确配置避免同步无关数据造成资源浪费和安全风险。 # 黑名单过滤一般不用 canal.instance.filter.black.regex ################################################# ## 3. 位点信息从何处开始同步 ################################################# # 如果这是全新的同步且需要历史数据可以设置为从头开始。 # canal.instance.master.journal.name # canal.instance.master.position # canal.instance.master.timestamp # 更常见的做法是不配置这些项。Canal启动后会自动从MySQL主库当前最新的Binlog位置开始同步。 # 如果存在meta.dat文件记录了上次同步的位点Canal会优先从meta.dat记录的位置开始实现断点续传。 ################################################# ## 4. 其他高级配置 ################################################# # 每次从MySQL拉取数据的批量大小 canal.instance.transaction.size1024 # 投递给客户端的批次大小 canal.instance.batch.size1000 # 获取数据超时时间毫秒 canal.instance.get.timeout30000关键点解析过滤规则 (canal.instance.filter.regex): 这是控制同步范围的核心。正则表达式中的点.需要转义\\.。例如只想同步business库下的所有表应写为business\\..*。配置不当是导致“为什么没收到数据”或“收到太多数据”的常见原因。位点控制: 对于全新部署不配置journal.name和positionCanal会从当前最新的Binlog位置开始。这对于监控实时新增数据是没问题的。但如果你需要补历史数据就需要指定一个更早的Binlog文件名和位置或者时间戳。获取这些信息可以使用SHOW MASTER STATUS;和SHOW BINARY LOGS;命令。批量参数:batch.size和transaction.size影响同步的性能和实时性。值太小频繁网络交互值太大内存占用高且延迟可能增加。需要根据实际数据变更频率和网络状况调整。4. 启动Canal服务与基础验证配置完成后就可以启动Canal了。我们使用其自带的脚本。4.1 启动与停止# 进入Canal目录 cd /opt/canal # 启动服务 sh bin/startup.sh # 查看启动日志这是排查问题最重要的文件 tail -f logs/canal/canal.log # 查看实例日志 tail -f logs/example/example.log如果启动成功你会在canal.log中看到类似下面的信息... INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server. ... INFO com.alibaba.otter.canal.deployer.CanalStarter - ## the canal server is running now ......停止服务的命令是sh bin/stop.sh4.2 基础连通性测试Canal启动后默认在11111端口提供了Socket服务。我们可以使用简单的telnet命令测试服务是否正常监听。telnet 127.0.0.1 11111如果连接成功你会看到一个空白屏幕光标在闪烁这表明端口是通的Canal Server在等待客户端连接。按Ctrl]然后输入quit退出。4.3 模拟数据变更验证这是验证Canal是否正常工作的关键一步。我们需要让MySQL产生数据变更然后通过Canal客户端来抓取这个变更。1. 准备一个简单的测试表在你的MySQL测试库例如test中执行CREATE TABLE user ( id int(11) NOT NULL AUTO_INCREMENT, name varchar(50) DEFAULT NULL, email varchar(100) DEFAULT NULL, PRIMARY KEY (id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;2. 使用Canal自带的Java示例客户端Canal解压包里的example子目录下有一个简单的Java客户端示例。我们编译并运行它。# 进入示例目录 cd /opt/canal/example # 编译确保JAVA_HOME已配置 sh build.sh # 运行客户端 sh run.shrun.sh脚本会启动一个客户端连接到本地的Canal Server127.0.0.1:11111订阅example实例。它会持续运行等待接收变更事件。3. 触发数据变更并观察在MySQL客户端中对测试表进行一些操作USE test; INSERT INTO user (name, email) VALUES (测试用户, testexample.com); UPDATE user SET emailupdatedexample.com WHERE name测试用户; DELETE FROM user WHERE name测试用户;此时观察运行run.sh的终端窗口。如果一切正常你应该能看到控制台打印出详细的Binlog解析结果格式类似**************************************************** * Batch Id: [1] * Execute Time: [2023-10-27 10:00:00] * Schema: [test] * Table: [user] * Event Type: [INSERT] * Before: null * After: {id1, name测试用户, emailtestexample.com} ****************************************************对于UPDATE操作你会看到Before和After分别有值清晰地展示了数据变化。DELETE操作则只有Before值。看到这样的输出就证明Canal已经成功安装、部署并且能够正确捕获和解析MySQL的增量数据了。5. 生产环境部署进阶与高可用考量单机版的Canal可以用于测试和中小型场景但在生产环境中我们需要考虑可靠性、可维护性和性能。这部分分享几个进阶部署方案和配置技巧。5.1 集群化部署与HA方案Canal本身支持集群模式其核心思想是多个Canal Server节点组成集群通过ZooKeeper来协调共同消费一个或多个MySQL实例的Binlog。当一个Server宕机时ZK会将其负责的实例转移给其他存活节点实现高可用。1. 依赖ZooKeeper集群首先你需要部署一个ZooKeeper集群至少3节点。假设ZK地址为zk1:2181,zk2:2181,zk3:2181。2. 修改Canal全局配置 (canal.properties)# 启用集群模式 canal.zkServers zk1:2181,zk2:2181,zk3:2181 # 将Canal Server的运行模式改为集群 canal.instance.global.mode spring canal.instance.global.lazy false # 指定集群模式下使用的Spring配置文件 canal.instance.global.spring.xml classpath:spring/default-instance.xml # 注意集群模式下通常使用default-instance.xml它集成了ZK的位点管理。3. 实例配置的调整在集群模式下每个实例的配置instance.properties可以放在ZK上统一管理也可以沿用本地文件。更推荐使用ZK管理便于统一修改和分发。需要在canal.properties中指定实例配置的存储模式# 将实例配置存储在ZK上 canal.instance.config.spring.xml classpath:spring/memory-instance-config.xml # 或者使用本地文件模式但集群切换时可能不一致 # canal.instance.config.spring.xml classpath:spring/file-instance-config.xml然后通过Canal提供的管理工具或直接操作ZK将instance.properties的内容上传到ZK的指定路径下如/otter/canal/destinations/example/conf。4. 启动多个Canal Server在所有Canal Server节点上使用相同的canal.propertiesZK地址一致启动服务。它们会自动连接到ZK并竞争对example实例的消费权。最终只有一个节点会成为“工作节点”其他节点作为“备用节点”待命。实操心得集群部署的难点在于ZK的维护和网络稳定性。务必确保Canal Server与ZK之间、Canal Server与MySQL之间的网络延迟低且稳定。监控ZK上/otter/canal/destinations/example/running节点的数据可以看到当前是哪个Server在运行该实例。5.2 对接消息队列Kafka/RocketMQ直接使用TCP模式客户端需要自己处理连接重试、负载均衡等问题。更生产化的做法是让Canal Server将解析后的数据直接投递到消息队列下游系统再从MQ消费。这样实现了彻底的解耦。1. 修改canal.properties# 将serverMode改为kafka或rocketmq canal.serverMode kafka # 如果是kafka kafka.bootstrap.servers kafka1:9092,kafka2:9092,kafka3:9092 # 如果是rocketmq # rocketmq.namesrv.addr rmq1:9876,rmq2:9876 # rocketmq.producer.group canal-producer-group # 其他相关MQ配置如重试次数、批量大小等 canal.mq.retries 3 canal.mq.batchSize 502. 配置实例的MQ投递规则 (instance.properties)# 指定投递到Kafka的Topic。支持动态Topic例如按库名表名划分 canal.mq.topic canal_test # 或者更精细的动态路由 # canal.mq.dynamicTopic test,.*,.* test_db , mydb,user user_topic # 分区规则 canal.mq.partition 0 # 或者按主键哈希 # canal.mq.partitionsNum 3 # canal.mq.partitionHash .*\\..*:$pk$3. 数据格式Canal投递到MQ的消息体默认是Protobuf序列化格式节省空间但可读性差。你也可以配置为JSON格式canal.mq.flatMessage true这样下游系统更容易解析。注意事项切换到MQ模式后原有的TCP客户端就无法连接了。你需要编写消费MQ的客户端或者使用Canal官方提供的canal-adapter等组件它内置了将MQ消息同步到ES、HBase等目标存储的能力。5.3 性能调优与监控随着同步数据量的增大可能需要对Canal进行调优。1. 关键性能参数canal.instance.transaction.size/canal.instance.batch.size: 如前所述调整批次大小。对于高吞吐场景可以适当调大如2048/2000但需要监控客户端消费速度避免堆积。canal.instance.parser.parallel/canal.instance.parser.parallelThreadSize: 启用并行解析利用多核CPU提升解析Binlog的速度。适用于表多、变更频繁的场景。canal.instance.parser.parallel true canal.instance.parser.parallelThreadSize 8 # 根据CPU核心数调整canal.instance.memory.rawEntry 是否在内存中缓存原始的Binlog Entry对象。默认true能提升性能但在内存受限或单次Batch非常大的情况下可能引起OOM可考虑设为false。2. 监控指标Canal通过JMX暴露了丰富的运行时指标。你可以使用JConsole、VisualVM等工具连接Canal进程JMX端口在canal.properties中通过canal.jmx.port配置监控消费延迟canal.instance.delay表示当前解析的Binlog时间戳与当前时间的差值。这是最重要的健康指标延迟持续增大说明消费跟不上生产。TPS每秒处理的事务数。内存使用canal.instance.memory.rawEntry缓存的大小。网络IO接收和发送的字节数。3. 日志分析与告警务必配置日志轮转并定期检查logs/example/example.log和logs/canal/canal.log。关注ERROR和WARN级别的日志。可以设置日志监控对“连接MySQL失败”、“解析异常”、“投递MQ失败”等关键错误进行告警。6. 常见问题排查与解决实录在实际部署和运维中我遇到过各种各样的问题。这里把一些典型问题和排查思路记录下来希望能帮你快速定位。6.1 启动失败类问题问题1Canal启动后立刻退出canal.log无错误信息或只有简短报错。排查首先检查logs/canal/canal.log和logs/example/example.log。更详细的日志可能在logs/目录下的stdout.log或stderr.log文件中。常见原因Java版本不兼容确认使用的是JDK 8或11。用java -version检查。端口被占用Canal默认使用11111端口。用netstat -tlnp | grep 11111检查。配置文件语法错误特别是canal.properties或instance.properties中有错误的配置项或格式。仔细核对。目录权限不足Canal进程需要对logs、conf等目录有写权限。用ps -ef | grep canal查看进程用户并检查目录权限。问题2日志显示com.alibaba.otter.canal.parse.exception.CanalParseException: connect failed: ...排查这是连接MySQL失败。检查instance.properties中的数据库地址、端口、用户名、密码是否正确。深入检查用mysql -ucanal -p -h127.0.0.1 -P3306命令手动测试连接。检查MySQL的bind-address配置确保不是只绑定了127.0.0.1如果Canal不在本机。检查防火墙是否放行了3306端口。确认canal用户是否有REPLICATION SLAVE权限。6.2 运行中数据同步异常问题3Canal客户端收不到任何数据变更消息。排查步骤 checklist 步骤检查项命令/方法预期结果/解决方案1. 源端是否有变更确认MySQL确实执行了DML操作。SELECT * FROM test.user;数据已变更。2. Canal实例运行正常查看实例日志有无错误。tail -f logs/example/example.log日志无ERROR有start successful或dump now等信息。3. 过滤规则是否正确检查canal.instance.filter.regex。核对instance.properties文件。正则表达式能匹配到你操作的表如test\\..*匹配test库所有表。4. Binlog格式是否为ROW确认MySQL配置。SHOW VARIABLES LIKE binlog_format;结果为ROW。5. 位点是否正常检查Canal是否从正确位置开始同步。查看logs/example/meta.log或ZK上记录的位点。位点时间戳接近当前时间或与SHOW MASTER STATUS;结果接近。6. 客户端连接和订阅正常确认客户端代码正确连接了Canal Server并订阅了实例。检查客户端代码确认destination参数为example。客户端无报错且Canal Server日志显示有客户端连接。问题4收到重复的数据变更消息。原因这通常是位点管理出了问题。Canal将消费进度binlog filename position持久化在conf/example/meta.dat文件模式或ZooKeeper集群模式。如果这个位点信息丢失或回退Canal就会从旧的位置重新拉取数据。解决文件模式检查meta.dat文件是否被误删或损坏。停止Canal备份并删除meta.dat文件重启Canal会从当前最新的Binlog位置开始注意这会丢失断点可能导致数据重复或遗漏慎用。集群模式检查ZK上对应实例的位点节点数据是否异常。网络分区也可能导致多个Canal Server同时认为自己是在线节点从而重复消费。确保ZK集群稳定。问题5同步延迟delay越来越大。分析这表示Canal消费Binlog的速度跟不上MySQL产生Binlog的速度。排查与优化检查Canal Server负载使用top命令查看Canal进程的CPU和内存使用率。如果持续很高可能是解析或投递遇到瓶颈。检查网络Canal Server与MySQL之间、与MQ/客户端之间的网络是否有瓶颈或丢包。调整批次参数适当增大canal.instance.transaction.size和canal.instance.batch.size减少网络交互次数。启用并行解析如果CPU有多余核心设置canal.instance.parser.paralleltrue。下游消费能力如果使用MQ模式检查Kafka/RocketMQ的消费者下游应用消费速度是否太慢。如果使用TCP模式检查你的客户端处理消息的速度。Binlog量激增检查MySQL是否在执行大批量数据操作如全表更新、历史数据迁移。可以考虑在业务低峰期进行这类操作。6.3 数据解析与格式问题问题6解析出的中文是乱码。原因字符集配置不一致。解决确保三处字符集统一为UTF-8。MySQL表/库的字符集CREATE TABLE ... CHARSETutf8mb4;MySQL连接字符集在instance.properties中设置canal.instance.connectionCharset UTF-8。Canal客户端或MQ消费者在解析消息体时指定正确的字符集。问题7收到的数据格式不符合预期或者缺少字段。原因表结构发生了变更ALTER TABLE而Canal缓存的表元数据没有及时更新。解决Canal会定期或触发式从MySQL拉取最新的表结构schema。如果发现问题可以重启对应的Canal实例强制刷新元数据缓存。在instance.properties中可以设置canal.instance.filter.table.auto.update true默认就是true来启用自动更新。部署和运维Canal是一个需要细致耐心的工作尤其是在生产环境。核心思路就是“大胆假设小心求证”从日志入手沿着数据流MySQL - Canal - Client/MQ逐段排查重点关注连接、权限、配置、资源这四个方面。把上面这些常见问题和解决方法备在手边能解决大部分初期遇到的技术障碍。