企业官网建设流程全解析

Flink车联网远控预警系统实战 —— 第 9 章 Flink CDC：预警配置表实时同步

章节定位：第四篇核心。预警作业是"配置驱动"的——采样区间、样本门槛、是否启用全部来自配置表t_cn_remote_warning_config。当前实现用 JDBC 每 60 秒轮询一次 MySQL，能用，但不够好：生效慢、全量扫描、对删除/禁用感知靠下一轮快照覆盖。本章讲清轮询方案的原理与痛点，并给出 Flink CDC 升级方案：把 MySQL binlog 变成一条实时配置流，实现秒级生效、增量捕获、天然感知删除。

版本基线：Apache Flink 2.x + Flink CDC 3.x + MySQL 8.x（JDK 17）

章节导读

• 9.1 配置表：预警系统的大脑
• 9.2 当前方案：JDBC 定时轮询
• 9.3 轮询方案的四个痛点
• 9.4 Flink CDC 升级方案
• 9.5 ConfigSnapshot：快照语义不变
• 9.6 一致性保证与运维要点
• 9.7 本章小结与下章预告

9.1 配置表：预警系统的大脑

9.1.1 配置表管什么

-- MySQL (TSP iov 库)CREATETABLEt_cn_remote_warning_config(idINTUNSIGNEDPRIMARYKEY,min_sample_countINT,-- 最少样本数(不足不写入)sample_interval_valueINT,-- 采样区间数值,如 30sample_interval_unitTINYINT,-- 单位: 1=分钟 2=小时 3=天remote_monitor_frequencyDECIMAL(5,2),-- 失败率阈值(TSP 用)remote_durationVARCHAR(50),-- 超时率阈值(TSP 用)push_alert_valueINT,-- 推送频率数值(TSP 用)push_alert_unitTINYINT,-- 推送频率单位(TSP 用)push_typeVARCHAR(255),-- 推送类型(TSP 用)push_accountVARCHAR(255),-- 推送账号(TSP 用)is_enableTINYINT-- 1=启用 0=禁用);

9.1.2 大数据侧只取 4 个字段

字段使用分工(第 1 章已讲,这里再强调): 大数据侧(Flink 作业)消费: · min_sample_count 样本校验 · sample_interval_value 采样区间长度 · sample_interval_unit 区间单位换算 · is_enable 是否参与统计 TSP 应用侧消费: · remote_monitor_frequency / remote_duration 阈值比对 · push_alert_* / push_type / push_account 推送控制 职责边界硬约束: 大数据侧只做统计与样本校验,不做阈值判定、不推送; 阈值判定与推送是 TSP 的职责。 所以 Flink 作业的查询 SQL 只 SELECT 自己需要的 4 列 + id。

9.1.3 配置的流转路径

MySQL(配置表) ──→ 同步通道 ──→ Flink 广播流 ──→ 统计函数(每分钟整点读取) 本章的命题: 这条"同步通道"用 JDBC 轮询还是 Flink CDC?

9.2 当前方案：JDBC 定时轮询

9.2.1 实现结构

// RemoteWarningConfigSource: SourceFunction 实现publicclassRemoteWarningConfigSourceextendsRichSourceFunction<ConfigSnapshot>{privatevolatilebooleanrunning=true;@Overridepublicvoidrun(SourceContext<ConfigSnapshot>ctx)throwsException{while(running){try{// 每 60s 查一次全量Map<String,RemoteWarningConfig>configs=queryAll(jdbcUrl);ConfigSnapshotsnapshot=newConfigSnapshot(configs);ctx.collect(snapshot);// 整体快照下发}catch(Exceptione){// 查询失败不抛异常,沿用上一轮快照,保 Source 存活log.warn("[配置查询失败] 沿用上一轮快照 err={}",e.getMessage());}Thread.sleep(60_000);}}privateMap<String,RemoteWarningConfig>queryAll(StringjdbcUrl){Stringsql="SELECT id, min_sample_count, sample_interval_value,"+" sample_interval_unit, is_enable "+" FROM tsp_ods.ods_iov_t_cn_remote_warning_config";// JDBC 查询 → 组装 Map<id, RemoteWarningConfig>}}

9.2.2 关键设计点

① 全量快照,不带 WHERE SQL 不加 "WHERE is_enable=1",查全量。 enabled = (is_enable == 1) 在统计侧判断。 原因: 禁用配置也要进广播态(标记不启用), 否则禁用操作永远生效不了(下详)。 ② 快照整体替换 下游 processBroadcastElement: state.clear() + putAll()。 语义: 这一轮快照 = 配置表的"当前真相", 配置被删/被禁 → 下一轮快照没有它 → 广播态自然消失。 这规避了"幽灵配置"问题(配置已删但广播态还残留)。 ③ 失败容错 查询失败不抛异常,沿用上一轮快照。 设计取舍: 宁可配置短暂过期,也不能让 Source 挂掉导致作业重启。

9.2.3 数据流时序

T+0s Source 启动,首次查询全量配置 → 广播 T+60s 第 2 次查询 → 快照替换 T+120s 第 3 次查询 → 快照替换 ... 配置变更生效延迟: 最坏 60s(变更刚发生后,要等下一轮轮询)

9.3 轮询方案的四个痛点

9.3.1 痛点一：生效延迟

场景: 运维发现某车型远控故障率飙升,紧急配置一条预警, 期望 1 分钟内生效。 实际: 最坏要等 60s 轮询周期, 加上统计触发周期(每分钟整点), 整体生效延迟可能 2 分钟。 后果: 紧急止血场景下,2 分钟可能错过关键窗口。

9.3.2 痛点二：全量扫描

每次轮询都 SELECT 全表,即使没有任何变更: · MySQL 侧: 无意义的 QPS(配置表通常几十行,压力可忽略,但不优雅) · Flink 侧: 每 60s 一次广播,即使配置没变 真正的成本不在压力,而在"广播风暴": 每次快照下发都会 clear+putAll 广播态, 配置行数 × 广播频率 = 状态操作次数。 行数小时无感,但模式不优雅。

9.3.3 痛点三：删除感知靠覆盖

幽灵配置问题(轮询方案是靠"整体替换"兜住的): 配置被物理删除后,下一轮快照里没有它, clear+putAll 后广播态里自然消失——语义正确。 但这依赖"每轮都全量覆盖"这一实现细节。 一旦有人把 putAll 改成增量 put(合并而非替换), 幽灵配置就出现了: 删掉的配置永远留在广播态, 继续参与统计——静默错误,极难排查。

9.3.4 痛点四：无变更感知

轮询不知道"发生了什么变化",只知道"现在是什么样"。 无法回答的问题: · 这条配置是刚改的还是一直如此? · 过去 1 小时配置变了几次? · 是谁/什么时间改的?(需另建审计,轮询给不了) 对于配置驱动系统,变更事件本身是有价值的: · 触发审计日志 · 触发缓存失效 · 触发下游联动(如通知 TSP 重新拉取)

9.3.5 痛点总结表

9.4 Flink CDC 升级方案

9.4.1 Flink CDC 是什么

Flink CDC(Change Data Capture): 把数据库的变更日志(MySQL binlog)直接变成 Flink 流。 核心能力: · 读取 binlog,捕获 INSERT/UPDATE/DELETE · 启动时可先做一次快照(全量),再增量跟随(Exactly-Once) · 不需要业务双写,不需要轮询,秒级生效 对本场景的适配: MySQL 配置表 ──binlog──→ MySqlSource(Flink CDC) ↓ ConfigSnapshot 组装 ↓ 广播流 → 统计函数

9.4.2 CDC Source 代码骨架

// 依赖(注意: 不打 -j,jar 进 lib 目录,硬约束)// flink-connector-mysql-cdc 3.xMySqlSource<ConfigChange>mySqlSource=MySqlSource.<ConfigChange>builder().hostname(config.get("mysql.host")).port(config.getInt("mysql.port",3306)).databaseList("tsp_iov").tableList("tsp_iov.t_cn_remote_warning_config").username(config.get("mysql.username")).password(config.get("mysql.password"))// env 注入,禁硬编码.serverId("5401-5499")// 每个并行实例一个 id 段.serverTimeZone("Asia/Shanghai")// 时区硬约束// 启动模式: 先快照全量,再增量跟随(Exactly-Once).startupMode(StartupMode.initial()).deserializer(newConfigChangeDeserializer())// 自定义反序列化.build();DataStream<ConfigChange>cdcStream=env.fromSource(mySqlSource,WatermarkStrategy.noWatermarks(),"mysql-cdc-config");

9.4.3 变更事件的组装

// 自定义反序列化: 把 binlog 行变成 ConfigChangepublicclassConfigChange{publicenumOp{INSERT,UPDATE,DELETE}publicOpop;// 变更类型publicRemoteWarningConfigconfig;// after 映像(DELETE 时为 before)}// 组装为快照(保持与轮询方案下游兼容)publicclassConfigSnapshotAssemblerextendsKeyedProcessFunction<String,ConfigChange,ConfigSnapshot>{privateMapState<String,RemoteWarningConfig>latest;// 全量最新状态@OverridepublicvoidprocessElement(ConfigChangec,Contextctx,Collector<ConfigSnapshot>out){switch(c.op){caseINSERT:caseUPDATE:latest.put(c.config.getId(),c.config);break;caseDELETE:latest.remove(c.config.getId());// 删除事件直接感知break;}// 每次变更都发出一份"当前全量快照",下游语义不变out.collect(newConfigSnapshot(toMap(latest)));}}

9.4.4 与轮询方案的对比

选型判断: · 配置表行数少、变更频率低 → 轮询已够用 · 追求秒级生效、需要变更审计、配置表将承载更多联动逻辑 → 升级 CDC · 本课程把 CDC 作为"升级方案"讲解,轮询作为"现状方案"保留, 两者下游广播语义完全一致(快照整体替换)

9.5 ConfigSnapshot：快照语义不变

9.5.1 为什么下游语义不变

关键设计: 无论上游是轮询还是 CDC, 下发给统计函数的都是"ConfigSnapshot 全量快照"。 统计函数 processBroadcastElement 只做一件事: state.clear() + state.putAll(snapshot.getConfigs()) 所以上游换成 CDC 后: · 统计函数零改动 · 广播态语义零变化 · 升级只是替换 Source,不动下游

9.5.2 is_enable 的处理位置

硬约束: is_enable 过滤不在 Source 层做,在统计层做。 Source 层: 查/收全量(不过滤 is_enable) 统计层(onTimer): cfg.isEnabled() ? 参与统计 : 跳过 为什么: 如果在 Source 层过滤 is_enable=0, 那么"禁用某配置"这个操作在 Source 层就被吞掉了, 广播态里那条配置永远不会被移除 → 幽灵配置。 在统计层判断,禁用配置进广播态但不参与统计, 语义清晰: 配置存在,只是不生效。

9.5.3 快照替换 vs 增量合并

幽灵配置问题的根源: 增量合并(put)而非整体替换(clear+putAll)。 ✓ 整体替换: state.clear(); state.putAll(snapshot) 广播态 = 快照 = 配置表当前真相 ✗ 增量合并: state.put(id, cfg) 删除的配置永远残留,幽灵配置参与统计 无论是轮询还是 CDC,下发快照后必须整体替换, 这是写广播态的铁律。

9.6 一致性保证与运维要点

9.6.1 启动一致性

作业重启场景: 轮询方案: 启动后 60s 内发起首次查询,快照恢复 CDC 方案: StartupMode.initial() 先做一次全量快照(锁表/读快照点), 再从快照点位增量跟随 binlog 共同点: 启动后广播态都是"当前全量真相",语义一致。 差异: CDC 的初始快照+增量跟随由 connector 保证 Exactly-Once, 不会出现"快照后到增量前"的变更丢失。

9.6.2 Exactly-Once 语义

硬约束组合(复用第 2 章): Checkpoint 60s / EXACTLY_ONCE / unaligned 重启策略: fixedDelay 无限重试 × 10s Source: group-offsets(Kafka)/ initial(CDC) CDC Source 的位点管理: binlog 位点随 Checkpoint 持久化, 故障恢复后从上次位点继续读, 配合快照整体替换语义,配置变更不丢不重。

9.6.3 运维要点

□ MySQL 侧: 必须开启 binlog(ROW 模式),binlog 保留期 ≥ 作业最大停机时间 □ server_id: 每个 Flink 并行实例一个 id 段,避免与其他 CDC 任务冲突 □ 时区: serverTimeZone="Asia/Shanghai",与 JVM 时区一致(硬约束) □ 凭证: 用户名/密码环境变量注入,禁硬编码进代码/DDL □ 权限: CDC 账号需 SELECT + RELOAD + SHOW DATABASES + REPLICATION SLAVE + REPLICATION CLIENT(读快照+读 binlog) □ 验证: 改一行配置,观察秒级生效;删一行配置,观察广播态消失; 禁一行配置,观察统计跳过但广播态保留

9.6.4 回退策略

升级 CDC 后保留轮询代码,配置开关切换: remote.config.source.type = jdbc # 现状 remote.config.source.type = cdc # 升级后 切回轮询: 改配置重启作业即可。 风险窗口: CDC 上线初期双跑观察(影子期),确认口径一致后单跑。

9.7 本章小结与下章预告

本章小结

┌──────────────────────────────────────────────────────────────┐ │ 第 9 章 要点回顾 │ └──────────────────────────────────────────────────────────────┘ ✓ 配置表是预警系统大脑: 大数据侧只取 4 字段 (min_sample_count/sample_interval/is_enable), 阈值与推送字段归 TSP ✓ 轮询方案: 60s 全量 SELECT + 快照整体替换, 失败沿用旧快照保 Source 存活; 四痛点: 生效慢/全量扫/删除感知靠覆盖/无变更感知 ✓ CDC 升级: MySqlSource 读 binlog,StartupMode.initial() 快照+增量 Exactly-Once,秒级生效,DELETE 天然感知 ✓ 下游语义不变: 无论轮询还是 CDC,都下发 ConfigSnapshot, 统计函数零改动;广播态必须 clear+putAll 整体替换 ✓ is_enable 在统计层过滤,不在 Source 层, 否则幽灵配置永存 ✓ 运维: binlog ROW 模式、server_id 段、时区 Asia/Shanghai、 凭证环境变量、CDC 账号最小权限、影子期双跑

下章预告

第 10 章 广播状态：预警配置动态驱动统计

配置到了广播流，怎么驱动统计？本章深入KeyedBroadcastProcessFunction三大入口——processElement（落桶）、processBroadcastElement（快照替换）、onTimer（整点回溯），以及分钟桶共享池、区间左闭右开、样本校验、过期清理的实现细节。这是预警作业的"计算中枢"。

官方参考资料

• Apache Flink CDC Connectors
• MySQL CDC Connector
• Flink Broadcast State

本章实战代码清单

序号	内容	说明
1	JDBC 轮询 Source 完整实现	9.2
2	轮询四痛点分析	9.3
3	MySqlSource（Flink CDC）代码骨架	9.4
4	ConfigSnapshotAssembler 变更事件组装	9.4
5	快照整体替换铁律（clear+putAll）	9.5
6	运维要点与回退策略	9.6