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() 避免位点丢失。
3. 解析 RowChange 提取变更数据
从 RowChange 中可获取：
MySQL8 CRM Schema Design Expert
下载
面向 CRM 系统的 MySQL 8 数据库模式设计。当用户需要为客户关系管理（CRM）系统设计、评审、优化或生成数据库模式时，应使用此项技能。
-
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 三者严格匹配，否则收不到任何数据。
Java免费学习笔记：立即使用
解锁 Java 大师之旅：从入门到精通的终极指南
