Canal是阿里巴巴开源的基于MySQL binlog的增量日志解析组件,Java项目通过客户端连接Canal Server订阅表变更,拉取并解析RowChange获取INSERT/UPDATE/DELETE数据,需注意位点确认、幂等处理及配置匹配。

Canal 是阿里巴巴开源的 MySQL 数据库增量日志解析组件,基于 MySQL 的 binlog 实现数据变更实时捕获。在 Java 项目中使用 Canal,核心是启动 Canal 客户端(Client)连接 Canal Server,订阅指定数据库表的变更事件,并在回调中处理 INSERT/UPDATE/DELETE 操作。
1. 环境与前置准备
确保以下条件已满足:
- MySQL 开启 binlog,格式为 ROW(
binlog_format=ROW),且binlog_row_image=FULL - 创建用于 Canal 连接的专用账号,并赋予
SELECT, REPLICATION SLAVE, REPLICATION CLIENT权限 - 下载并启动 Canal Server(如 canal.deployer),配置好 instance(对应目标 MySQL 实例)
- Java 项目引入 Canal 客户端依赖(Maven):
2. 编写 Canal Client 订阅逻辑
典型 Java 客户端代码结构如下:
- 创建
CanalConnector,连接 Canal Server(默认地址127.0.0.1:11111) - 调用
connect()建立连接,再用subscribe("pattern")订阅表名正则(如test\.user或.*\..*) - 循环调用
getMessages(100)拉取批量消息(Message对象包含多个Entry) - 遍历每个
Entry,过滤EntryType == ROWDATA,再解析RowChange获取具体 DML 类型和字段值
注意:需手动控制 ack() 确认消费成功,否则重复拉取;异常时建议 rollback() 避免位点丢失。
立即学习“Java免费学习笔记(深入)”;
面向 CRM 系统的 MySQL 8 数据库模式设计。当用户需要为客户关系管理(CRM)系统设计、评审、优化或生成数据库模式时,应使用此项技能。
3. 解析 RowChange 提取变更数据
从 RowChange 中可获取:
-
eventType:枚举值INSERT/UPDATE/DELETE -
isDdl:是否为 DDL 操作(如 ALTER TABLE),一般忽略 -
rowDatas:每条记录含beforeColumns(旧值)、afterColumns(新值) - 通过
column.getName()和column.getValue()提取字段名与值,注意空值、时间类型(可能为字符串或 long)
示例:UPDATE 场景下,beforeColumns 有原主键/版本字段,afterColumns 有更新后全量字段,可用于构建对比或同步语句。
4. 生产环境关键注意事项
实际部署需关注:
- 连接池管理:避免频繁新建
CanalConnector,推荐单例 + 手动重连机制 - 位点可靠性:开启
flatMessage=true(Canal Server 配置)可简化 JSON 格式输出;也可接入 Kafka 由 Canal Server 直接投递,Java 侧改用 Kafka Consumer 消费,提升吞吐与容错 - 幂等处理:因网络或重启可能导致重复消息,业务层应基于主键+操作类型+时间戳做去重或状态校验
- 异常兜底:监听器中捕获所有异常,记录日志并跳过当前批次,防止阻塞后续消息
不复杂但容易忽略:务必确认 Canal Server 的 destination 名称、MySQL 实例的 serverId、以及客户端订阅 pattern 三者严格匹配,否则收不到任何数据。

















