Flink CDC实现MySQL数据实时同步方案解析

1. 项目背景与核心挑战去年接手公司订单中心重构项目时我们遇到了一个典型的数据同步难题。线上MySQL主库承载着日均3亿的交易订单数据需要实时同步到Elasticsearch集群供搜索服务使用同时还要写入MongoDB供风控系统分析。最初的方案是采用应用层双写但很快就暴露出致命问题。双写模式下应用代码里到处是这样的逻辑// 伪代码示例 transaction.begin(); try { // 写入主库 orderDao.insert(order); // 同步写ES esClient.index(order); // 同步写Mongo mongoTemplate.save(order); transaction.commit(); } catch(Exception e) { transaction.rollback(); // 补偿逻辑... }这套方案在实际运行中暴露出三个严重问题性能瓶颈高峰期写操作平均RT从50ms飙升到800ms数据不一致网络抖动时出现主库成功但从库失败的情况维护成本高每次新增数据源都要修改业务代码更麻烦的是当我们需要把数据同步到新建的ClickHouse集群做实时分析时发现要在几十个微服务里全部改造一遍双写逻辑。这直接促使我们寻找更优雅的解决方案。2. 技术选型与方案设计2.1 主流方案对比我们对比了四种常见的数据同步方案方案类型典型实现优点缺点应用层双写代码硬编码实现简单耦合业务代码一致性难保证定时任务扫描QuartzSQL查询解耦业务逻辑高延迟源库压力大数据库触发器MySQL Trigger实时性较好影响主库性能难以维护日志解析Canal/Flink CDC完全解耦低延迟需要维护中间件2.2 为什么选择Flink CDCFlink CDCChange Data Capture相较于传统方案的核心优势在于无侵入性通过解析MySQL binlog获取变更对业务代码零改造全增量一体化既能捕获增量变更也支持全量历史数据初始化Exactly-Once语义通过checkpoint机制保证数据不丢不重强大的转换能力可以在数据管道中灵活进行ETL处理我们的最终架构设计如下图所示MySQL Binlog → Flink CDC → (流式ETL) → Kafka → ↘ Elasticsearch Sink ↘ MongoDB Sink ↘ ClickHouse Sink3. 核心实现细节3.1 环境配置要点MySQL服务端配置[mysqld] server-id 1 log_bin mysql-bin binlog_format ROW # 必须使用ROW模式 binlog_row_image FULL # 需要完整镜像 expire_logs_days 3 # 根据磁盘空间调整Flink CDC连接器配置debezium.source.connector.class: io.debezium.connector.mysql.MySqlConnector debezium.source.database.hostname: mysql-host debezium.source.database.port: 3306 debezium.source.database.user: cdc_user debezium.source.database.password: xxxx debezium.source.database.server.id: 5401 debezium.source.database.server.name: orders_db debezium.source.database.include.list: order_center debezium.source.table.include.list: order_center.orders debezium.source.database.history.kafka.bootstrap.servers: kafka:9092 debezium.source.database.history.kafka.topic: schema-changes.order_center3.2 核心处理逻辑我们使用Flink SQL实现的主要数据处理流程-- 创建MySQL CDC源表 CREATE TABLE orders_mysql ( id BIGINT, user_id BIGINT, amount DECIMAL(18,2), status INT, create_time TIMESTAMP(3), update_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname mysql-host, port 3306, username cdc_user, password xxxx, database-name order_center, table-name orders ); -- 创建Elasticsearch目标表 CREATE TABLE orders_es ( id BIGINT, user_id BIGINT, amount DOUBLE, status INT, create_time TIMESTAMP(3), update_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector elasticsearch-7, hosts http://es01:9200, index orders, document-id.key-delimiter $ ); -- 写入ES的物化视图 INSERT INTO orders_es SELECT id, user_id, CAST(amount AS DOUBLE) as amount, status, create_time, update_time FROM orders_mysql WHERE status IN (1,2,3); -- 只同步有效订单3.3 关键性能优化并行度调优// 根据分区键设置并行度 env.setParallelism(6); tableEnv.getConfig().set(table.exec.source.split.max-num, 12);Checkpoint配置execution.checkpointing.interval: 10s execution.checkpointing.timeout: 5min execution.checkpointing.mode: EXACTLY_ONCE state.backend: rocksdb state.checkpoints.dir: hdfs://namenode:8020/flink/checkpoints反压处理-- 在Kafka connector中启用反压感知 scan.topic-partition-discovery.interval 30s properties.max.poll.records 5004. 生产环境落地实践4.1 上线迁移步骤全量数据初始化# 使用Flink CDC的snapshot功能 flink run \ -Dexecution.checkpointing.interval0 \ -c com.ververica.cdc.connectors.mysql.source.MySqlSource \ mysql-cdc-connector.jar \ --hostname mysql-host \ --port 3306 \ --database order_center \ --tables orders \ --username cdc_user \ --password xxxx \ --server-id 5401 \ --scan.incremental.snapshot.chunk.size 5000灰度切换方案第一阶段双写CDC并行运行比对数据一致性第二阶段流量切到CDC保留双写作为备份第三阶段完全下线双写逻辑4.2 监控指标设计我们通过Prometheus采集的关键指标端到端延迟flink_taskmanager_job_latency_source_idxxx处理吞吐flink_taskmanager_job_numRecordsInPerSecondCheckpoint成功率flink_jobmanager_job_lastCheckpointDurationGrafana监控看板配置示例{ panels: [{ title: CDC延迟监控, targets: [{ expr: flink_taskmanager_job_latency_source_id~source_orders_.*, legendFormat: {{instance}} }] }] }5. 典型问题排查指南5.1 Binlog位置丢失现象Flink任务重启后从最早位点开始消费导致数据重复解决方案检查MySQL的binlog_expire_logs_seconds设置建议≥7天验证Flink的offset.storage配置增加Checkpoint保存时间state.checkpoints.num-retained: 105.2 网络闪断处理现象Debezium连接器报Communications link failure优化方案debezium.source.connect.keep.alivetrue debezium.source.connect.keep.alive.interval.ms30000 debezium.source.connect.timeout.ms60000 debezium.source.connection.pool.size155.3 大事务处理现象单事务修改10w行数据导致内存OOM优化策略调整snapshot分块大小debezium.source.snapshot.fetch.size2000 debezium.source.max.batch.size2048在Flink侧启用缓冲table.exec.state.ttl 36 h execution.buffer-timeout 1s6. 最终效果与收益指标对比如下指标项双写方案Flink CDC方案提升幅度写入延迟5000±1200ms80±30ms98.4%↓数据一致性99.2%99.999%3个9提升资源消耗32核/64G8核/16G75%↓扩展新数据源需要代码发布修改配置即可效率10倍↑这套方案上线后不仅解决了当前订单数据的同步问题还为公司建立了统一的数据实时同步平台。后续我们又接入了用户行为日志、库存变更等10个重要数据源全部通过配置化方式实现不再需要业务代码改造。