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 主流方案对比

我们对比了四种常见的数据同步方案：

方案类型	典型实现	优点	缺点
应用层双写	代码硬编码	实现简单	耦合业务代码，一致性难保证
定时任务扫描	Quartz+SQL查询	解耦业务逻辑	高延迟，源库压力大
数据库触发器	MySQL Trigger	实时性较好	影响主库性能，难以维护
日志解析	Canal/Flink CDC	完全解耦，低延迟	需要维护中间件

2.2 为什么选择Flink CDC

Flink CDC（Change Data Capture）相较于传统方案的核心优势在于：

无侵入性：通过解析MySQL binlog获取变更，对业务代码零改造
全增量一体化：既能捕获增量变更，也支持全量历史数据初始化
Exactly-Once语义：通过checkpoint机制保证数据不丢不重
强大的转换能力：可以在数据管道中灵活进行ETL处理

我们的最终架构设计如下图所示：

MySQL Binlog → Flink CDC → (流式ETL) → Kafka → ↘ Elasticsearch Sink ↘ MongoDB Sink ↘ ClickHouse Sink

3. 核心实现细节

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_center

3.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' = '500'

4. 生产环境落地实践

4.1 上线迁移步骤

全量数据初始化：

# 使用Flink CDC的snapshot功能 flink run \ -Dexecution.checkpointing.interval=0 \ -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_id=xxx
处理吞吐：flink_taskmanager_job_numRecordsInPerSecond
Checkpoint成功率：flink_jobmanager_job_lastCheckpointDuration

Grafana监控看板配置示例：

{ "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: 10

5.2 网络闪断处理

现象：Debezium连接器报Communications link failure

优化方案：

debezium.source.connect.keep.alive=true debezium.source.connect.keep.alive.interval.ms=30000 debezium.source.connect.timeout.ms=60000 debezium.source.connection.pool.size=15

5.3 大事务处理

现象：单事务修改10w+行数据导致内存OOM

优化策略：

调整snapshot分块大小：

debezium.source.snapshot.fetch.size=2000 debezium.source.max.batch.size=2048

在Flink侧启用缓冲：

'table.exec.state.ttl' = '36 h' 'execution.buffer-timeout' = '1s'

6. 最终效果与收益

指标对比如下：

指标项	双写方案	Flink CDC方案	提升幅度
写入延迟	5000±1200ms	80±30ms	98.4%↓
数据一致性	99.2%	99.999%	3个9提升
资源消耗	32核/64G	8核/16G	75%↓
扩展新数据源	需要代码发布	修改配置即可	效率10倍↑