Flink车联网远控预警系统实战 —— 第 5 章 聚合层KeyedProcessFunction 状态与双定时器章节定位第三篇核心。解析后的CtrlNode按 requestId 聚合成一条RemoteControlRecord。本章是整个链路最核心的状态管理逻辑——用ValueStateAggState存六个锚点用两个定时器兜底 90s 延迟 3s保证一次远控只输出一条同时处理中间态过滤、三级失败判定、超时标记。理解了这一章就理解了有状态流处理的精髓。版本基线Apache Flink 2.x章节导读5.1 聚合层要解决什么问题5.2 AggState六锚点状态模型5.3 六锚点写入规则5.4 双定时器机制5.5 中间态过滤与最终态延迟输出5.6 三级失败判定与超时5.7 onTimer输出与清状态5.8 State TTL 与状态清理5.9 本章小结与下章预告5.1 聚合层要解决什么问题5.1.1 输入与输出输入: union 后的 CtrlNode 流(L3 L2 混合),按 requestId keyBy 同一个 requestId 的节点可能有 5~10 条,到达顺序不确定 输出: 一条 RemoteControlRecord(一次远控一行) 含: requestId, vin, startMs, ctrlTimeMs, ctrlResult, fail, timeout, failSource5.1.2 三个核心问题问题 1: 节点乱序到达,怎么组装成一次完整远控? → 用 KeyedState 缓存所有节点,按 forward 语义写入对应锚点 问题 2: 什么时候输出? 等不到所有节点怎么办? → 双定时器: 收到最终态等 3s 输出; 90s 还没最终态就兜底输出 问题 3: 一次远控可能输出多条吗? → 不会: 任一定时器触发即输出并清状态,保证只出一条5.2 AggState六锚点状态模型每个 requestId 一份ValueStateAggState包含六个时间锚点 各级结果码 两个定时器时间戳。publicstaticclassAggStateimplementsSerializable{// ── 六个时间锚点 ──LongstartMs;// APP_TO_L3(最早)LongfirstL3ToL2;// L3_TO_L2(最早,多轮取首)LonglastL2ToL3;// L2_TO_L3(最晚,多轮取尾)LongendMs;// L3_TO_APP 最终态Longtl0;// T_L0(最后一轮)Longfl0;// F_L0(最后一轮)// ── 三级结果码 ──StringctrlResult;// L3 最终结果(1/其他/null)StringctrlMsg;StringtboxResult;// 车机结果(F_L0)StringtboxMsg;Stringl2ResultMain;// L2 结果主来源(L2_TO_L3,已归一化)Stringl2MsgMain;Stringl2ResultAux;// L2 结果辅助来源(T_L3)Stringl2MsgAux;// ── 其它 ──Stringvin;LongtimeoutTimerTs;// 90s 兜底定时器时间戳LongfinalTimerTs;// 最终态 3s 延迟定时器时间戳}5.2.1 状态设计思路六个锚点对应第 1 章讲的耗时计算: startMs → ctrlTime 的起点 firstL3ToL2 → L3 前段 lastL2ToL3 → L3 后段 endMs → ctrlTime 的终点 tl0 / fl0 → tboxTime 三级结果码对应第 1 章的失败判定: ctrlResult → 一级(L3 最终态) tboxResult → 二级(车机) l2ResultMain/Aux→ 三级(L2,主辅) 两个定时器时间戳: 用于 onTimer 时区分是哪个定时器触发,以及清理另一个5.3 六锚点写入规则processElement里按side和forward把节点写入对应锚点。5.3.1 L3 侧节点写入if(isL3){switch(forward){caseAPP_TO_L3:state.startMsmin(state.startMs,ts);// 最早break;caseL3_TO_L2:state.firstL3ToL2min(state.firstL3ToL2,ts);// 最早(多轮取首)break;caseL2_TO_L3:state.lastL2ToL3max(state.lastL2ToL3,ts);// 最晚(多轮取尾)state.l2ResultMainnode.getResultCode();// L2 主结果state.l2MsgMainnode.getResultMsg();break;caseL3_TO_APP:handleL3ToApp(state,node,ctx);// 见 5.5break;}}5.3.2 L2 侧节点写入else{switch(forward){caseT_L0:state.tl0max(state.tl0,ts);// 最后一轮break;caseF_L0:state.fl0max(state.fl0,ts);// 最后一轮state.tboxResultnode.getResultCode();// 车机结果state.tboxMsgnode.getResultMsg();break;caseT_L3:state.l2ResultAuxnode.getResultCode();// L2 辅助结果state.l2MsgAuxnode.getResultMsg();break;// F_L3 / T_SMS / F_SMS: 不影响锚点,直接跳过}}5.3.3 min/max 取值规则一览锚点取值原因startMsmin最早APP_TO_L3 可能重发取最早那条firstL3ToL2min最早多轮往返第一条标记 L3 开始交接lastL2ToL3max最晚多轮往返最后一条标记 L2 结束交接endMs覆盖最终态最终态只有一条直接覆盖tl0max最晚多轮取最后一轮的下发时刻fl0max最晚多轮取最后一轮的返回时刻为什么 start/firstL3ToL2 取 min,而 lastL2ToL3/tl0/fl0 取 max? start 和 firstL3ToL2 是起点类锚点,越早就越准确。 lastL2ToL3、tl0、fl0 是终点类锚点,越晚就越接近最终状态。 本质: 起点取最早,终点取最晚。5.3.4 VIN 补全// L3 symbol 优先(先到先填,L3/L2 同 requestId 的 VIN 一致)if((state.vinnull||state.vin.isEmpty())node.getVin()!null!node.getVin().isEmpty()){state.vinnode.getVin();}VIN 补全规则: 先到先填,不为 null 就不覆盖。 L3 的 data.symbol 和 L2 的 dcm_no 对同一个 requestId 是一致的, 所以谁先到用谁,不用纠结优先级。5.4 双定时器机制这是聚合层最核心的设计——用两个定时器保证一次远控只输出一条。5.4.1 两个定时器定时器注册时机时长作用兜底定时器timeoutTimerTs首条消息到达90s到点仍无最终态 → 降级输出最终态定时器finalTimerTs收到 L3_TO_APP 最终态3s等迟到的 F_L0/T_L3 再输出5.4.2 兜底定时器注册AggStatestatestateState.value();if(statenull){statenewAggState();// 首条消息即注册兜底长定时器longfireTsctx.timerService().currentProcessingTime()aggregateTimeoutMs;ctx.timerService().registerProcessingTimeTimer(fireTs);state.timeoutTimerTsfireTs;}为什么首条消息就注册兜底定时器? 因为不知道这次远控有没有最终态。 如果 L3_TO_APP 最终态丢了,作业不能无限等下去, 90s 后必须输出一个降级结果(失败或未知),释放状态。5.4.3 最终态定时器注册privatevoidhandleL3ToApp(AggStatestate,CtrlNodenode,Contextctx){Stringmsgnode.getResultMsg();Stringcodenode.getResultCode();if(codenull||INTERIM_MSG.equals(msg))return;// 中间态丢弃state.endMsnode.getTimestamp();state.ctrlResultcode;state.ctrlMsgmsg;// 只注册一次(重复最终态只更新结果,不重复注册)if(state.finalTimerTsnull){longfireTsctx.timerService().currentProcessingTime()finalDelayMs;ctx.timerService().registerProcessingTimeTimer(fireTs);state.finalTimerTsfireTs;}}为什么收到最终态后还要等 3s? L3_TO_APP 到了,但 F_L0(车机结果)、T_L3(L2结果)可能因为网络延迟还在路上。 等 3s 让这些节点拱上来,这样: ① 失败判定可以用更完整的信息 ② 耗时计算的锚点更全(tl0/fl0 用于 tboxTime) 3s 是经验值,足够覆盖跨 topic 的网络延迟,又不会让输出太慢。5.4.4 双定时器的协作关系时间轴: T0s 首条节点到达 → 注册兜底定时器(T90s) T3s L3_TO_APP 最终态到达 → 注册最终态定时器(T6s) T6s 最终态定时器触发 → 输出 → 清理两个定时器 → 清状态 (兜底定时器 T90s 被删除,不会再触发) 异常情况: T0s 首条节点到达 → 注册兜底定时器(T90s) T90s 兜底定时器触发 → 无最终态,降级输出 → 清状态 (最终态定时器从未注册)关键: 任一定时器触发即输出并清状态,另一个定时器被删除。 所以一次远控只会输出一条记录,无论正常还是异常。5.5 中间态过滤与最终态延迟输出5.5.1 中间态指令已下发L3_TO_APP 可能发两条消息消息msgresult处理第 1 条指令已下发“1”丢弃中间态第 2 条指令执行成功/远控失败“1”/“0”记录为最终态privatestaticfinalStringINTERIM_MSG指令已下发;if(codenull||INTERIM_MSG.equals(msg)){return;// 中间态或无 result,直接丢弃}5.5.2 为什么在聚合层过滤而不是解析层解析层是无状态的,看到一条 L3_TO_APP 就输出一条, 不知道后面还有没有第二条。 如果在解析层丢了第一条,而第二条因故没到, 聚合层就拿不到任何 L3_TO_APP,会走兜底逻辑, 判定可能出错(本应成功的被判为未知)。 正确的做法: 解析层原样透出,聚合层按 msg 判断是否中间态。 聚合层有状态,知道最终态是否已到达。5.5.3 最终态延迟 3s 输出T3s: 收到 L3_TO_APP 最终态,记录 endMs/ctrlResult/ctrlMsg, 注册 finalTimerTs T6s T4s: 收到 F_L0(迟到的车机结果),写入 tboxResult T6s: finalTimer 触发,buildRecord 时 tboxResult 已在状态里, 虽然有 L3 最终态不用 tboxResult 判定,但状态完整5.6 三级失败判定与超时buildRecord方法里完成失败判定和超时标记。5.6.1 三级判定逻辑if(ctrlResult!null){// 一级: L3 最终态为准fail!SUCCESS.equals(ctrlResult);failSourcefail?L3:null;}else{// 最终态缺失,从车机→L2 自下而上查Stringl2Resultstate.l2ResultMain!null?state.l2ResultMain:state.l2ResultAux;if(state.tboxResult!null!SUCCESS.equals(state.tboxResult)){failtrue;failSourceTBOX;ctrlResult0;}elseif(l2Result!null!SUCCESS.equals(l2Result)){failtrue;failSourceL2;ctrlResult0;}else{failfalse;failSourcenull;// 都成功或缺失 → 未知}}5.6.2 判定流程图ctrlResult ! null? │ ├─ 是 → ctrlResult 1? │ ├─ 是 → 成功(failfalse, failSourcenull) │ └─ 否 → 失败(failtrue, failSourceL3) │ └─ 否(最终态缺失) │ tboxResult ! null 且 ! 1? ├─ 是 → 失败(failtrue, failSourceTBOX) └─ 否 │ l2Result(main ?? aux) ! null 且 ! 1? ├─ 是 → 失败(failtrue, failSourceL2) └─ 否 → 未知(failfalse, failSourcenull, ctrlResultnull)5.6.3 超时判定booleantimeoutfailctrlTime!nullctrlTimebusinessTimeoutMs;超时 失败 且 总耗时 30s 注意: 兜底失败(无 endMs)的 ctrlTimenull,不计超时。 因为没有终点,无法计算总耗时。 businessTimeoutMs 默认 30000(30s),外置配置, 改动前需与 TSP 对齐,否则两侧判定口径分叉。5.6.4 startMs 缺失直接丢弃if(state.startMsnull){log.warn([聚合丢弃] requestId{} 无APP_TO_L3起点,无法归属采样区间,requestId);returnnull;}为什么 startMs 缺失要丢弃? 统计层按 startMs 归属分钟桶(第 10 章)。 没有 startMs,这条记录不知道算哪个分钟的, 留着只会污染统计,直接丢弃。5.7 onTimer输出与清状态5.7.1 定时器触发处理OverridepublicvoidonTimer(longtimestamp,OnTimerContextctx,CollectorRemoteControlRecordout){AggStatestatestateState.value();if(statenull)return;booleanisFinalTimerstate.finalTimerTs!nullstate.finalTimerTstimestamp;booleanisTimeoutTimerstate.timeoutTimerTs!nullstate.timeoutTimerTstimestamp;if(!isFinalTimer!isTimeoutTimer)return;// 不是我的定时器StringrequestIdctx.getCurrentKey();StringtriggerTypeisFinalTimer?FINAL:FALLBACK;RemoteControlRecordrecordbuildRecord(state,triggerType,requestId);if(record!null)out.collect(record);clearTimers(ctx,state);// 删除另一个定时器stateState.clear();// 清状态}5.7.2 定时器区分onTimer 被调用时,只知道 timestamp,不知道是哪个定时器注册的。 通过比对 state.finalTimerTs 和 state.timeoutTimerTs 来区分: timestamp finalTimerTs → 最终态定时器 timestamp timeoutTimerTs → 兜底定时器 如果两个都不等于 timestamp,说明是过期的定时器(已被清理但还没触发), 直接 return 不处理。5.7.3 清理定时器与状态privatevoidclearTimers(OnTimerContextctx,AggStatestate){if(state.timeoutTimerTs!null){ctx.timerService().deleteProcessingTimeTimer(state.timeoutTimerTs);}if(state.finalTimerTs!null!state.finalTimerTs.equals(state.timeoutTimerTs)){ctx.timerService().deleteProcessingTimeTimer(state.finalTimerTs);}}输出后必须: ① 删除另一个还没触发的定时器(否则会重复输出) ② 清空 stateState(否则同一 requestId 的新数据会叠加到旧状态) 这两步缺一不可。5.8 State TTL 与状态清理5.8.1 TTL 配置StateTtlConfigttlConfigStateTtlConfig.newBuilder(Duration.ofMinutes(stateTtlMinutes)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build();ValueStateDescriptorAggStatedescriptornewValueStateDescriptor(remote-agg-state,AggState.class);descriptor.enableTimeToLive(ttlConfig);5.8.2 TTL 参数参数默认值说明stateTtlMinutes15 分钟状态保留时长UpdateTypeOnCreateAndWrite创建和写入时刷新 TTLTTL 15 分钟的考量: 兜底定时器 90s,正常输出后状态立即清除。 TTL 15 分钟是给状态没被正常清除的兜底(如作业重启后定时器丢失), 避免状态无限堆积。 15 分钟远大于 90s 兜底,不会误伤正常流程。5.8.3 三重清理保障保障 1: 正常输出后 stateState.clear() —— 绝大多数情况 保障 2: State TTL 15 分钟 —— 定时器丢失等异常情况 保障 3: Flink 状态后端 ForSt —— 大状态支撑 三重保障确保状态不会无限膨胀。5.9 本章小结与下章预告本章小结┌──────────────────────────────────────────────────────────────┐ │ 第 5 章 要点回顾 │ └──────────────────────────────────────────────────────────────┘ ✓ AggState: 每 requestId 一份 ValueState,含 6 锚点 3 级结果码 2 个定时器时间戳 ✓ 六锚点写入: start/firstL3ToL2 取 min(最早), lastL2ToL3/tl0/fl0 取 max(最晚),endMs 覆盖(最终态) ✓ 双定时器: 兜底 90s(首条注册) 最终态延迟 3s(最终态注册), 任一触发即输出并清状态,保证一次远控只出一条 ✓ 中间态过滤: L3_TO_APP 的 msg指令已下发丢弃, 只在聚合层做(解析层无状态) ✓ 三级判定: L3最终态 → 车机 → L2(主??辅), 都成功/缺失为未知;startMs 缺失直接丢弃 ✓ 超时: fail ctrlTime30s;兜底失败无 endMs 不计超时 ✓ 三重清理: 正常输出 clear TTL 15min ForSt 状态后端下章预告第 6 章 业务规则引擎耗时计算与失败判定聚合层输出了RemoteControlRecord但还没讲透耗时怎么算、失败怎么定。本章把第 1 章的业务规则落到代码里——四段耗时公式、三级失败判定的边界情况、节点缺失的连锁影响。这章是业务规则和代码实现的桥梁。官方参考资料Apache Flink KeyedProcessFunctionApache Flink State TTL本章实战代码清单序号内容说明1AggState 六锚点状态模型5.22六锚点写入规则min/max5.33双定时器机制兜底 90s 延迟 3s5.44中间态指令已下发过滤5.55三级失败判定 超时标记5.66onTimer 输出与清状态5.77State TTL 15 分钟配置5.8