Flink CDC实时消费MySQL变更数据实战指南-开发者社区

1. 最简Flink CDC消费MySQL实战指南

今天给大家分享一个开箱即用的Flink CDC消费MySQL变更数据的完整解决方案。这个Demo基于Flink 1.17和Flink-CDC 2.4版本，采用Java实现，包含所有必要配置和代码，复制粘贴就能直接运行。我会从原理到实践详细讲解每个环节，包括你可能遇到的坑和解决方案。

CDC（Change Data Capture）是实时数据同步的核心技术，而Flink CDC是目前最成熟的实现方案之一。相比传统的Canal或Debezium方案，Flink CDC最大的优势是与Flink引擎深度集成，可以直接作为Source使用，无需额外维护组件。我们这次要实现的场景是：实时捕获MySQL中指定表的insert/update/delete操作，并将变更事件打印到控制台。

2. 环境准备与依赖配置

2.1 项目依赖配置

首先来看pom.xml中的关键依赖。除了Flink基础依赖外，重点需要添加flink-connector-mysql-cdc和fastjson：

<properties> <flink.version>1.17.0</flink.version> <flink.cdc.version>2.4.0</flink.cdc.version> </properties> <dependencies> <!-- Flink核心依赖 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_2.12</artifactId> <version>${flink.version}</version> </dependency> <!-- Flink CDC MySQL连接器 --> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>${flink.cdc.version}</version> </dependency> <!-- JSON序列化 --> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.83</version> </dependency> </dependencies>

注意：Flink CDC 2.4版本开始要求Flink最低版本为1.16，这里我们使用1.17确保兼容性。fastjson用于后续的JSON解析，如果项目中有其他JSON库也可以替换。

2.2 MySQL配置要求

MySQL必须开启binlog并配置为ROW模式，这是CDC工作的基础。修改MySQL配置文件my.cnf（Linux）或my.ini（Windows）：

[mysqld] server-id = 1 log_bin = mysql-bin binlog_format = ROW binlog_row_image = FULL expire_logs_days = 7

配置说明：

server-id：MySQL主从复制需要的唯一标识
log_bin：启用binlog并指定前缀
binlog_format=ROW：必须设置为ROW模式才能捕获行级变更
binlog_row_image=FULL：记录变更前后的完整数据
expire_logs_days：binlog保留天数，根据磁盘空间设置

修改后重启MySQL服务，并验证配置是否生效：

SHOW VARIABLES LIKE 'log_bin'; -- 应为ON SHOW VARIABLES LIKE 'binlog_format'; -- 应为ROW

常见问题：如果MySQL是8.0+版本且运行在Docker中，可能需要额外配置binlog过期策略，否则可能导致磁盘爆满。

3. 核心代码实现解析

3.1 Flink CDC数据源构建

下面是完整的Java实现代码，我添加了详细注释：

import com.ververica.cdc.connectors.mysql.source.MySqlSource; import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class FlinkMysqlCdcDemo { public static void main(String[] args) throws Exception { // 创建流执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 必须开启checkpoint以实现精确一次语义 env.enableCheckpointing(5000); // 每5秒一次checkpoint // 1. 构建MySQL CDC数据源 MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("localhost") // MySQL地址 .port(3306) // MySQL端口 .databaseList("test_db") // 监控的数据库 .tableList("test_db.user") // 监控的表（格式：db.table） .username("root") // 用户名 .password("root") // 密码 // 使用JSON格式反序列化变更事件 .deserializer(new JsonDebeziumDeserializationSchema()) .build(); // 2. 从数据源读取并打印 env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source") .setParallelism(1) // 设置并行度为1保证顺序 .print(); // 打印到控制台 env.execute("Flink MySQL CDC Demo"); } }

关键配置说明：

enableCheckpointing(5000)：必须开启，否则无法保证精确一次语义
databaseList和tableList：支持监控多个库表，用逗号分隔
JsonDebeziumDeserializationSchema：将变更事件转为JSON字符串
setParallelism(1)：保证变更事件的顺序性

3.2 变更事件格式解析

程序运行后，控制台会输出类似如下的JSON格式变更事件：

{ "before": null, // 变更前的数据（insert时为null） "after": { // 变更后的数据（delete时为null） "id": 1, "name": "test", "age": 20 }, "source": { // 元数据（数据库、表等信息） "db": "test_db", "table": "user" }, "op": "c" // 操作类型：c=insert, u=update, d=delete }

字段说明：

before和after：分别表示变更前后的数据状态
op：操作类型标识符
source：包含数据库、表名等元信息

4. 生产环境注意事项

4.1 性能与稳定性优化

并行度设置：
- 单表消费建议并行度为1保证顺序
- 多表消费可以按表设置不同并行度

Checkpoint调优：

env.enableCheckpointing(10000); // 生产环境建议10-30秒 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);

断点续传：

mySqlSourceBuilder .startupOptions(StartupOptions.initial()) // 或使用特定binlog位置 // .startupOptions(StartupOptions.specificOffset("mysql-bin.000003", 4L))

4.2 常见问题排查

连接问题：
- 错误：Failed to connect to MySQL server
- 检查：MySQL用户权限（需要REPLICATION CLIENT和REPLICATION SLAVE权限）
```
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'username'@'%';
```
Binlog问题：
- 错误：The connector is trying to read binlog starting at...
- 检查：binlog是否被清除（设置合理的expire_logs_days）
时区问题：
- 现象：时间字段与数据库不一致
- 解决：在JDBC URL中添加时区参数
```
.jdbcUrl("jdbc:mysql://localhost:3306/test_db?serverTimezone=Asia/Shanghai")
```

5. 扩展应用场景

5.1 解析JSON转实体对象

如果需要将JSON转为Java对象，可以添加如下处理：

env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source") .map(json -> JSON.parseObject(json, UserChangeEvent.class)) .print(); // 定义变更事件类 @Data public static class UserChangeEvent { private User before; private User after; private String op; @Data public static class User { private Integer id; private String name; private Integer age; } }

5.2 写入Kafka

将变更事件发送到Kafka的示例：

// 添加Kafka依赖 <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.12</artifactId> <version>${flink.version}</version> </dependency> // 写入Kafka代码 env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source") .addSink(new FlinkKafkaProducer<>( "kafka-topic", new SimpleStringSchema(), kafkaProps ));

5.3 使用SQL API实现

对于偏好SQL的用户，Flink CDC也提供了SQL API：

-- 注册MySQL CDC表 CREATE TABLE user_source ( id INT, name STRING, age INT ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'localhost', 'port' = '3306', 'username' = 'root', 'password' = 'root', 'database-name' = 'test_db', 'table-name' = 'user' ); -- 查询变更数据 SELECT * FROM user_source;

6. 实际应用中的经验分享

在多个生产项目中实施Flink CDC后，我总结出以下几点经验：

监控binlog延迟：通过source.ts_ms字段可以计算处理延迟，建议设置告警
处理DDL变更：表结构变更可能导致解析失败，需要：
- 监控binlog解析异常
- 准备回滚方案（如从checkpoint重启）
大数据量初始化：全量同步阶段可能对源库造成压力，建议：
- 在业务低峰期启动
- 分批次同步（通过scan.incremental.snapshot.chunk.size控制）

高可用配置：

env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 3, // 重启尝试次数 Time.of(10, TimeUnit.SECONDS) // 重启间隔 ));

这套方案已经在我们的订单、用户等多个核心业务系统稳定运行，相比传统的双写或定时同步方案，数据一致性得到了显著提升，延迟控制在秒级以内。

Flink CDC实时消费MySQL变更数据实战指南