java中利用mysql binlog配合canal实现实时增量同步，核心是让canal作为“伪从库”拉取解析binlog，java应用订阅消费变更事件；需开启row格式binlog、配置canal server并正确ack位点。
Java 中利用 MySQL binlog 配合 Canal 实现实时增量同步，核心是让 Canal 作为“伪从库”拉取并解析 binlog，再由 Java 应用订阅消费变更事件。整个过程不侵入业务代码，延迟低、可靠性高。
MySQL 端必须开启并配置好 binlog
Canal 依赖 MySQL 的 binlog 才能工作，必须确保以下几点已生效：
-
开启 binlog：在
my.cnf的[mysqld]段中添加：log-bin = mysql-binserver-id = 1（需全局唯一，不能与其他从库冲突） -
设置为 ROW 格式：
binlog-format = ROW—— 这是关键，只有 ROW 格式才能精确捕获每行数据的变更（INSERT/UPDATE/DELETE），STATEMENT 或 MIXED 无法满足增量同步要求。 -
可选但推荐：
sync-binlog = 1（每次事务都刷盘，提升数据安全性）、expire-logs-days = 7（自动清理旧日志，防磁盘爆满） - 配置完成后重启 MySQL，并执行
SHOW VARIABLES LIKE '%log_bin%';和SHOW VARIABLES LIKE 'binlog_format';确认状态为ON和ROW
Canal Server 配置与启动
下载 Canal（GitHub 官方最新版），解压后修改 conf/example/instance.properties（或新建 instance）：
-
连接 MySQL：
canal.instance.master.address = 127.0.0.1:3306，填入账号密码（需有REPLICATION SLAVE权限） -
指定同步库表：
canal.instance.filter.regex = order_db\..*（正则匹配，支持多库多表） -
位点初始化：首次运行可留空，Canal 自动从当前最新 position 开始；如需指定起点，设
canal.instance.binlog.offset和canal.instance.binlog.filename - 启动服务：
./bin/startup.sh，查看logs/canal/canal.log和logs/example/example.log确认连接成功、无报错
Java 应用端订阅与解析变更事件
引入 Canal 客户端依赖（Maven）：
<dependency><groupid>com.alibaba.otter</groupid><artifactid>canal.client</artifactid><version>1.1.7</version></dependency>
编写消费逻辑示例：
- 创建
CanalConnector，连接 Canal Server（默认地址127.0.0.1:11111） - 调用
connector.subscribe()订阅目标 destination（即 instance 名，如example） - 循环调用
connector.getWithoutAck(100)拉取一批 Entry，每条 Entry 包含多个 RowData - 遍历
entry.getEntryType() == EntryType.ROWDATA，再解析rowChange.getEventType()（INSERT/UPDATE/DELETE） - 从
rowChange.getRowDatasList()提取前后镜像（beforeColumns/afterColumns），按字段名和值做业务处理（如更新 Redis、写入 ES） - 成功处理后调用
connector.ack(batchId)，失败则nack()触发重推
生产环境关键注意事项
实际部署时需关注几个易忽略但影响稳定性的点：
-
权限最小化：MySQL 账号仅授予
SELECT, REPLICATION SLAVE, REPLICATION CLIENT，禁用 super 或 DDL 权限 - 位点管理：Java 消费端务必正确 ack，否则 Canal store 会堆积；建议结合本地持久化 offset（如存入 Redis）实现断点续传
- 高可用设计：单 Canal Server 是瓶颈，可通过部署多个 instance + Kafka 中转（Canal 原生支持投递到 Kafka），再由 Java 消费 Kafka，解耦并提升吞吐
-
DDL 同步谨慎处理：默认 Canal 不解析 DDL（如 ALTER TABLE），如需支持，需开启
canal.instance.enable.dts = true并注意下游兼容性
Java免费学习笔记：立即使用
解锁 Java 大师之旅：从入门到精通的终极指南
