1. 为什么需要实时数据同步想象一下你正在运营一个电商平台用户在下单支付后订单数据需要实时同步到库存系统、物流系统和财务系统。如果采用传统的定时任务批量同步可能会出现库存扣减延迟、物流发货滞后等问题。这种场景下实时数据同步就成了刚需。Canal作为阿里巴巴开源的MySQL数据库增量日志解析工具能够完美解决这类问题。它通过伪装成MySQL的slave节点实时获取主库的binlog变更再将变更事件推送给下游消费者。相比传统的轮询查询或触发器方案Canal具有零侵入性不改动业务代码、低延迟秒级同步和高可靠性断点续传三大优势。我在去年参与的一个金融项目中就深有体会。当时需要将核心交易系统的数据实时同步到风控系统最初尝试用Spring Batch定时跑批结果风控规则总是延迟5分钟触发。切换到Canal方案后不仅实现了秒级同步还减少了70%的服务器资源消耗。2. 环境准备与组件部署2.1 组件选型与下载首先需要从GitHub获取Canal的最新稳定版本当前为1.1.7。这里重点需要两个核心组件canal.deployer负责连接MySQL并解析binlogcanal.adapter将变更事件适配到目标存储建议在Linux服务器上新建两个目录分别存放这两个组件。我习惯用/opt/canal作为根目录下面建立deployer和adapter子目录。解压时注意保持目录结构清晰mkdir -p /opt/canal/{deployer,adapter} tar -zxvf canal.deployer-1.1.7.tar.gz -C /opt/canal/deployer tar -zxvf canal.adapter-1.1.7.tar.gz -C /opt/canal/adapter2.2 MySQL配置检查确保源MySQL数据库已开启binlog并配置为ROW模式。这是我踩过的第一个坑——很多开发环境的MySQL默认使用STATEMENT格式的binlog-- 检查当前配置 SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format; -- 如需修改在my.cnf中添加 [mysqld] log-binmysql-bin binlog-formatROW server-id1重要提醒修改配置后必须重启MySQL服务。曾经有同事忘记重启排查了半天为什么Canal收不到事件。3. Canal核心组件配置详解3.1 deployer配置实战进入deployer的conf目录修改example/instance.properties# MySQL连接配置 canal.instance.mysql.slaveId1234 canal.instance.master.address127.0.0.1:3306 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal123 # 过滤规则同步哪些库表 canal.instance.filter.regex.*\\..*这里特别强调三点经验不要直接使用root账号建议创建专属账号并授予REPLICATION SLAVE权限slaveId不要与现有MySQL集群中的ID冲突过滤规则使用正则表达式.*\\..*表示同步所有库表启动deployer时建议用nohup挂到后台nohup ./bin/startup.sh logs/canal.log 21 3.2 adapter配置技巧adapter的配置主要在conf/application.yml。分享一个多目标库的配置模板canal.conf: mode: tcp srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/source_db?useSSLfalse username: source_user password: source_pass canalAdapters: - instance: example groups: - groupId: g1 outerAdapters: - name: rdb key: target1 properties: jdbc.url: jdbc:mysql://10.0.0.2:3306/target_db1 jdbc.username: target_user jdbc.password: target_pass - name: rdb key: target2 properties: jdbc.url: jdbc:postgresql://10.0.0.3:5432/target_db2 jdbc.username: pg_user jdbc.password: pg_pass踩坑警示当需要同步到不同类型的数据库如MySQL到PostgreSQL时务必在lib目录下添加对应的JDBC驱动包否则启动时会报ClassNotFound异常。4. SpringBoot集成实战4.1 客户端依赖配置在pom.xml中添加以下依赖时要注意版本兼容性dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.5/version exclusions exclusion groupIdch.qos.logback/groupId artifactIdlogback-classic/artifactId /exclusion /exclusions /dependency dependency groupIdcom.google.protobuf/groupId artifactIdprotobuf-java/artifactId version3.17.3/version /dependency4.2 事件处理核心代码下面是一个增强版的Canal客户端实现增加了异常处理和状态监控Component public class CanalClient { private static final Logger logger LoggerFactory.getLogger(CanalClient.class); Value(${canal.server.host}) private String canalHost; Value(${canal.server.port}) private int canalPort; PostConstruct public void init() { ThreadFactory factory new ThreadFactoryBuilder() .setNameFormat(canal-worker-%d) .setUncaughtExceptionHandler((t, e) - logger.error(Thread {} got exception, t.getName(), e)) .build(); ExecutorService executor Executors.newCachedThreadPool(factory); executor.submit(this::process); } private void process() { CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(canalHost, canalPort), example, , ); try { connector.connect(); connector.subscribe(your_db.your_table); connector.rollback(); while (true) { Message message connector.getWithoutAck(1000); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { Thread.sleep(1000); continue; } processEntries(message.getEntries()); connector.ack(batchId); } } catch (Exception e) { logger.error(Canal processing error, e); } finally { connector.disconnect(); } } private void processEntries(ListEntry entries) { for (Entry entry : entries) { if (entry.getEntryType() EntryType.ROWDATA) { RowChange rowChange RowChange.parseFrom(entry.getStoreValue()); for (RowData rowData : rowChange.getRowDatasList()) { if (rowChange.getEventType() EventType.INSERT) { // 处理新增数据 MapString, String afterColumns rowData.getAfterColumnsList() .stream() .collect(Collectors.toMap( Column::getName, Column::getValue)); logger.info(INSERT: {}, afterColumns); } } } } } }5. 高级配置与性能优化5.1 多表同步的优雅方案原始文章提到多表同步需要复制多个yml文件其实可以通过动态配置实现更优雅的管理。在application.yml中添加canal: sync: tables: - source: db1.table1 target: db2.table1_copy - source: db1.table2 target: db2.table2_copy然后通过Spring的ConfigurationProperties读取配置动态生成adapter需要的yml文件。这种方式在表结构变更时只需修改配置无需重启服务。5.2 性能调优参数在高并发场景下这些参数能显著提升吞吐量# deployer调优 canal.instance.memory.buffer.size32m canal.instance.memory.buffer.memunit1024 # adapter调优 canal.conf.canalAdapters[0].groups[0].outerAdapters[0].properties.jdbc.batchSize500 canal.conf.canalAdapters[0].groups[0].outerAdapters[0].properties.jdbc.flushInterval1000在最近的压力测试中通过这些优化使单节点处理能力从2000TPS提升到了8000TPS。不过要注意batchSize过大会增加内存消耗需要根据服务器配置找到平衡点。6. 监控与异常处理建议通过Spring Boot Actuator暴露监控端点关键指标包括最后同步时间积压事件数量最近错误信息对于网络闪断等临时故障可以实现重试机制private void processWithRetry() { int maxRetries 3; int retryCount 0; while (retryCount maxRetries) { try { process(); break; } catch (CanalClientException e) { retryCount; logger.warn(Process failed, retry {}/{}, retryCount, maxRetries); if (retryCount maxRetries) { alertService.notifyAdmin(Canal同步持续失败); } Thread.sleep(5000); } } }在实际生产环境中我们还会将同步状态持久化到Redis这样即使服务重启也能从断点继续。
SpringBoot 整合 Canal：构建 MySQL 实时数据同步的实战指南
1. 为什么需要实时数据同步想象一下你正在运营一个电商平台用户在下单支付后订单数据需要实时同步到库存系统、物流系统和财务系统。如果采用传统的定时任务批量同步可能会出现库存扣减延迟、物流发货滞后等问题。这种场景下实时数据同步就成了刚需。Canal作为阿里巴巴开源的MySQL数据库增量日志解析工具能够完美解决这类问题。它通过伪装成MySQL的slave节点实时获取主库的binlog变更再将变更事件推送给下游消费者。相比传统的轮询查询或触发器方案Canal具有零侵入性不改动业务代码、低延迟秒级同步和高可靠性断点续传三大优势。我在去年参与的一个金融项目中就深有体会。当时需要将核心交易系统的数据实时同步到风控系统最初尝试用Spring Batch定时跑批结果风控规则总是延迟5分钟触发。切换到Canal方案后不仅实现了秒级同步还减少了70%的服务器资源消耗。2. 环境准备与组件部署2.1 组件选型与下载首先需要从GitHub获取Canal的最新稳定版本当前为1.1.7。这里重点需要两个核心组件canal.deployer负责连接MySQL并解析binlogcanal.adapter将变更事件适配到目标存储建议在Linux服务器上新建两个目录分别存放这两个组件。我习惯用/opt/canal作为根目录下面建立deployer和adapter子目录。解压时注意保持目录结构清晰mkdir -p /opt/canal/{deployer,adapter} tar -zxvf canal.deployer-1.1.7.tar.gz -C /opt/canal/deployer tar -zxvf canal.adapter-1.1.7.tar.gz -C /opt/canal/adapter2.2 MySQL配置检查确保源MySQL数据库已开启binlog并配置为ROW模式。这是我踩过的第一个坑——很多开发环境的MySQL默认使用STATEMENT格式的binlog-- 检查当前配置 SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format; -- 如需修改在my.cnf中添加 [mysqld] log-binmysql-bin binlog-formatROW server-id1重要提醒修改配置后必须重启MySQL服务。曾经有同事忘记重启排查了半天为什么Canal收不到事件。3. Canal核心组件配置详解3.1 deployer配置实战进入deployer的conf目录修改example/instance.properties# MySQL连接配置 canal.instance.mysql.slaveId1234 canal.instance.master.address127.0.0.1:3306 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal123 # 过滤规则同步哪些库表 canal.instance.filter.regex.*\\..*这里特别强调三点经验不要直接使用root账号建议创建专属账号并授予REPLICATION SLAVE权限slaveId不要与现有MySQL集群中的ID冲突过滤规则使用正则表达式.*\\..*表示同步所有库表启动deployer时建议用nohup挂到后台nohup ./bin/startup.sh logs/canal.log 21 3.2 adapter配置技巧adapter的配置主要在conf/application.yml。分享一个多目标库的配置模板canal.conf: mode: tcp srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/source_db?useSSLfalse username: source_user password: source_pass canalAdapters: - instance: example groups: - groupId: g1 outerAdapters: - name: rdb key: target1 properties: jdbc.url: jdbc:mysql://10.0.0.2:3306/target_db1 jdbc.username: target_user jdbc.password: target_pass - name: rdb key: target2 properties: jdbc.url: jdbc:postgresql://10.0.0.3:5432/target_db2 jdbc.username: pg_user jdbc.password: pg_pass踩坑警示当需要同步到不同类型的数据库如MySQL到PostgreSQL时务必在lib目录下添加对应的JDBC驱动包否则启动时会报ClassNotFound异常。4. SpringBoot集成实战4.1 客户端依赖配置在pom.xml中添加以下依赖时要注意版本兼容性dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.5/version exclusions exclusion groupIdch.qos.logback/groupId artifactIdlogback-classic/artifactId /exclusion /exclusions /dependency dependency groupIdcom.google.protobuf/groupId artifactIdprotobuf-java/artifactId version3.17.3/version /dependency4.2 事件处理核心代码下面是一个增强版的Canal客户端实现增加了异常处理和状态监控Component public class CanalClient { private static final Logger logger LoggerFactory.getLogger(CanalClient.class); Value(${canal.server.host}) private String canalHost; Value(${canal.server.port}) private int canalPort; PostConstruct public void init() { ThreadFactory factory new ThreadFactoryBuilder() .setNameFormat(canal-worker-%d) .setUncaughtExceptionHandler((t, e) - logger.error(Thread {} got exception, t.getName(), e)) .build(); ExecutorService executor Executors.newCachedThreadPool(factory); executor.submit(this::process); } private void process() { CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(canalHost, canalPort), example, , ); try { connector.connect(); connector.subscribe(your_db.your_table); connector.rollback(); while (true) { Message message connector.getWithoutAck(1000); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { Thread.sleep(1000); continue; } processEntries(message.getEntries()); connector.ack(batchId); } } catch (Exception e) { logger.error(Canal processing error, e); } finally { connector.disconnect(); } } private void processEntries(ListEntry entries) { for (Entry entry : entries) { if (entry.getEntryType() EntryType.ROWDATA) { RowChange rowChange RowChange.parseFrom(entry.getStoreValue()); for (RowData rowData : rowChange.getRowDatasList()) { if (rowChange.getEventType() EventType.INSERT) { // 处理新增数据 MapString, String afterColumns rowData.getAfterColumnsList() .stream() .collect(Collectors.toMap( Column::getName, Column::getValue)); logger.info(INSERT: {}, afterColumns); } } } } } }5. 高级配置与性能优化5.1 多表同步的优雅方案原始文章提到多表同步需要复制多个yml文件其实可以通过动态配置实现更优雅的管理。在application.yml中添加canal: sync: tables: - source: db1.table1 target: db2.table1_copy - source: db1.table2 target: db2.table2_copy然后通过Spring的ConfigurationProperties读取配置动态生成adapter需要的yml文件。这种方式在表结构变更时只需修改配置无需重启服务。5.2 性能调优参数在高并发场景下这些参数能显著提升吞吐量# deployer调优 canal.instance.memory.buffer.size32m canal.instance.memory.buffer.memunit1024 # adapter调优 canal.conf.canalAdapters[0].groups[0].outerAdapters[0].properties.jdbc.batchSize500 canal.conf.canalAdapters[0].groups[0].outerAdapters[0].properties.jdbc.flushInterval1000在最近的压力测试中通过这些优化使单节点处理能力从2000TPS提升到了8000TPS。不过要注意batchSize过大会增加内存消耗需要根据服务器配置找到平衡点。6. 监控与异常处理建议通过Spring Boot Actuator暴露监控端点关键指标包括最后同步时间积压事件数量最近错误信息对于网络闪断等临时故障可以实现重试机制private void processWithRetry() { int maxRetries 3; int retryCount 0; while (retryCount maxRetries) { try { process(); break; } catch (CanalClientException e) { retryCount; logger.warn(Process failed, retry {}/{}, retryCount, maxRetries); if (retryCount maxRetries) { alertService.notifyAdmin(Canal同步持续失败); } Thread.sleep(5000); } } }在实际生产环境中我们还会将同步状态持久化到Redis这样即使服务重启也能从断点继续。
