企业官网建设流程全解析

KeyedProcessFunction 这个类,我前后在三四个项目里用到过,每次复盘的感觉都不一样:第一次觉得它就是个能注册定时器的 ProcessFunction,第二次发现坑全在状态清理上,第三次才真正意识到它的价值不是"定时器",而是"把 key 作为一等公民之后的自由编排能力"。前面两篇聊过基础的 ProcessFunction 和窗口相关的处理函数,这一篇专门啃 KeyedProcessFunction,把它拆到能直接抄作业的程度。

它适合谁看?如果你已经能跑通一个 keyBy + 窗口的作业,但遇到"下单 30 分钟未支付自动关单""同一用户 10 分钟内重复点击合并""设备连续 5 分钟无心跳就判离线"这类跨事件、跨时间的需求时,窗口表达式写得很别扭,那这篇就是给你的。文章里的场景全部虚构,代码基于常见版本(1.17 / 1.18)写成,较新版本里open()的参数类型有变化我会单独标注。

1. 先把定位摆正:它到底比别的处理函数多给了什么

1.1 三个处理函数的边界差异

很多人第一次接触KeyedProcessFunction是在翻官方文档,看到"可以访问 keyed state 和定时器"就直接跳过去了,然后在真实需求里来回纠结到底该用ProcessFunction还是它。我用一张对照表把边界说清楚,这张表是我自己在做方案评审时随手画的,后来成了团队里的默认判断依据。

看懂这张表的关键在一行:ProcessFunction里的timerService其实也能调,但因为流没有按 key 分区,定时器触发时你根本不知道它属于哪个逻辑实体,等于拿到一个没有主语的回调。而KeyedProcessFunction之所以叫这个名字,是因为它和 keyBy 是强绑定关系:同一个 key 的所有元素一定进同一个并行子任务,定时器也按 key 分开存储,触发时你能通过ctx.getCurrentKey()拿回这个 key。这句话是整个类的地基,后面所有关于状态一致性的讨论都建立在它上面。

窗口函数为什么不能替代它?因为窗口是"预先切好的时间格子",元素必须落进某个格子才能被处理。而真实业务里的时间边界往往是"跟着事件走的":一个订单的超时起算点是它自己创建的那一刻,不是整点也不是 5 分钟分界点。用窗口去模拟这种需求,得先滚动再合并,代码量和出错率都翻倍。

1.2 它真正解决的几类问题

抛开文档上的抽象描述,我把实际用到它的场景归成四类,你对着自己的需求往里套基本不会错。

第一类是单元素级别的超时控制。订单未支付关单、工单超时升级、验证码过期失效,这些需求的共同点是"从某个事件开始计时,到点没等到另一个事件就执行兜底逻辑"。用KeyedProcessFunction加一个ValueState存原始事件,一个ValueState存定时器时间戳,逻辑极其直白。

第二类是自定义会话切分。官方的会话窗口依赖 gap 参数,但真实业务里"会话结束"的判定经常带业务条件,比如只有浏览商品才算会话内活动,搜索不算。这种就得自己维护一个"最后一次有效活动时间",每次来元素时比较差值,超过阈值就先输出上一段会话再开新的。

第三类是乱序到达的事件配对。支付回调和订单创建谁先到不确定,用状态缓存先到的那个,等到另一个时配对输出,超过容忍时间还没配上就走侧输出流落到人工兜底。

第四类是延迟触发的旁路逻辑。比如用户下单后延迟 10 秒推送一条推荐消息,或者写入一条风控观察记录。这类需求用外部消息队列的延迟消息也能做,但如果逻辑依赖流内状态,放在onTimer里做能省掉一次状态外部化的开销。

注意:如果你发现自己想用它做"大批量数据的定时批处理",先停下来想想。定时器是存在状态后端里的,数量上去以后 checkpoint 体积和恢复时间都会很难看,这种需求用外部调度系统更合适。

1.3 一个能跑起来的最小骨架

先给一个不带任何业务的最小版本,把方法的调用顺序摸清楚,比一上来就塞业务逻辑好理解得多。

public class MinimalKeyedFunction extends KeyedProcessFunction<String, String, String> { private transient ValueState<Long> tsState; @Override public void open(Configuration parameters) { tsState = getRuntimeContext().getState( new ValueStateDescriptor<>("ts", Long.class)); } @Override public void processElement(String value, Context ctx, Collector<String> out) throws Exception { String key = ctx.getCurrentKey(); Long ts = ctx.timestamp(); long fireAt = (ts == null ? ctx.timerService().currentProcessingTime() : ts) + 5000L; ctx.timerService().registerEventTimeTimer(fireAt); tsState.update(fireAt); out.collect("registered: " + key + " at " + fireAt); } @Override public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception { out.collect("fired: " + ctx.getCurrentKey() + " at " + timestamp); tsState.clear(); } }

这段代码里有三个细节值得单独拎出来。第一,open()是唯一能安全拿RuntimeContext的地方,构造器里调用getRuntimeContext()会直接抛空指针,因为那时候算子还没被运行时初始化。第二,ctx.timestamp()在处理时间语义下可能是null,必须判空,我见过不止一个人在这里吃空指针。第三,onTimer的第一个参数就是你注册时传进去的那个时间戳原值,不是"当前时间",这一点在做幂等判断时特别重要。

2. 机制拆解:状态、定时器、时间语义三件套

2.1 keyed state 的声明位置与序列化陷阱

KeyedProcessFunction里能用的状态类型有ValueState、ListState、MapState、ReducingState、AggregatingState五种,声明方式都是getRuntimeContext().getState(descriptor),并且必须放在open()方法内。这一点和你在普通算子里的直觉不太一样:不是成员变量初始化时就能拿到,因为描述符需要注册到运行时上下文里。

状态的生命周期跟 key 绑定,key 变了就是另一份状态。这个设计的好处是你完全不用操心并发,同一个 key 的元素在同一个子任务里串行处理,不需要加锁。坏处是如果 key 的基数特别大(比如用 sessionId 或者设备指纹当 key,一天几千万个),状态就会线性膨胀,必须配合清理策略。

序列化是最容易出问题的地方。如果你用 POJO 作为状态类型,一定要满足三个条件:类是public的、有公开无参构造、字段有标准的 getter/setter。不满足的话 Flink 会退化到 Kryo 序列化,不但性能差,而且在类结构变更后反序列化容易出问题,恢复作业时一片报错。我自己的习惯是在open()之后立刻打一行日志确认状态类型,或者干脆在测试环境用StateTtlConfig配一个极短的过期时间,把序列化路径跑一遍。

ValueStateDescriptor<OrderEvent> desc = new ValueStateDescriptor<>("order", OrderEvent.class); // OrderEvent 需要是 public 类 + 无参构造 + getter/setter orderState = getRuntimeContext().getState(desc);

提示:用ListState存"一次会话内的所有事件"时,注意别无限追加。若会话跨度可能很长,优先用MapState按事件 ID 去重后再存,或者只存聚合所需的中间结果。

2.2 定时器的注册、去重与删除

定时器分两种:registerProcessingTimeTimer和registerEventTimeTimer。前者的时间基准是机器的系统时钟,注册后到点就触发,和数据的乱序程度无关,适合"我就是要等 10 秒"这种不依赖事件时间的场景;后者依赖水位线推进,只有watermark >= 注册的时间戳时才触发,乱序数据下更准确,但前提是你的水位线生成策略要合理。

关于去重,有个必须记住的规则:同一个 key 上,相同时间戳的定时器只会注册一个。你重复注册不会报错,也不会触发两次。这个特性在做"刷新超时时间"的逻辑时很有用,但也容易造成误解,比如有人以为注册了两次会触发两次,在onTimer里写了累加逻辑,结果发现只加了一次。

删除定时器要注意传参必须和注册时的时间戳完全一致:

Long fireAt = tsState.value(); if (fireAt != null) { ctx.timerService().deleteEventTimeTimer(fireAt); tsState.clear(); }

如果注册时用的是 A 时间戳,删除时传了 B,那次删除就是空操作,原来的定时器照样会触发,你的业务逻辑就得多处理一次"本不该来的回调"。我在做订单关单时踩过这个坑:第一版用的是相对时间偏移量计算,后来改成绝对时间戳,删除逻辑忘记同步改,导致一笔已支付的订单在半小时后收到了超时通知。

关于底层存储,HashMapStateBackend把定时器放在堆内存的优先队列里,数量级在几十万以内表现很好,超过就容易给 GC 压力;EmbeddedRocksDBStateBackend把定时器存进本地数据库,数量大时更稳,代价是每次触发多一次读盘。选型时不要只看峰值 QPS,要看"s 时刻存活的定时器总数",这个才是真正的压力指标。

2.3 时间语义混用带来的隐形 bug

作业里同时用处理时间和事件时间,是排查成本最高的一类问题。举个具体的例子:你的水位线生成策略配了 30 秒的最大乱序容忍,注册事件时间定时器时用的却是System.currentTimeMillis(),那么在数据回放场景下,历史数据的eventTime可能是几小时前,而定时器注册在未来,结果就是"数据都处理完了,定时器一个都没触发"。

我现在的做法是定死一条规矩:只要上游数据带业务时间字段,定时器一律用事件时间,并显式用元素携带的时间戳计算触发点。只有在完全无业务时间概念(比如纯粹的限流、延迟投递)时,才用处理时间,并且在代码注释里写清楚为什么。

// 好:触发时间来自业务字段,可回放、可复现 long fireAt = value.getCreateTime() + 30 * 60 * 1000L; ctx.timerService().registerEventTimeTimer(fireAt); // 危险:依赖机器时钟,回放和重跑时行为不一致 long fireAt2 = ctx.timerService().currentProcessingTime() + 30 * 60 * 1000L;

还有一个细节是水位线本身要推进才会触发定时器。如果你按 key 分区的数据稀疏,某个子任务长时间没数据,水位线就卡住,这个子任务上的定时器即使时间到了也不触发。解决办法是给空闲数据源配withIdleness,让分片在没有数据时也参与水位线计算,否则你会看到"大部分 key 正常关单,少数 key 卡住不动"的诡异现象。

3. 实战一:超时未支付订单的自动关单

3.1 需求拆解与选型判断

需求本身很朴素:订单创建后 30 分钟没收到支付事件,就发出一条关单消息。但要把边界理清楚:如果支付事件比创建事件先到怎么办(上游是两条独立的消息流,顺序无保证)?如果订单被取消了呢?如果同一订单收到重复的创建消息呢?

先做选型判断。用定时器还是用外部延迟队列?如果关单逻辑只需要订单 ID 和创建时间,不需要感知流里的其他状态,用延迟队列更省资源;如果需要结合用户等级(等级高的多给 10 分钟)、结合风控标记(风控命中则提前关单),那状态必须存在流里,KeyedProcessFunction就是更自然的选择。这个项目的规则是分层的,所以走定时器方案。

再定 key。订单 ID 是天然唯一的,但它会导致 key 数量巨大且几乎无热点,适合用ValueState存储;如果按用户 ID 当 key,能做"用户维度的统一超时管理",但同一个用户的多笔订单会挤在一个 key 上,状态互相牵连。这里有取舍:按订单 ID 做简单、状态小、清理容易;按用户 ID 做能做聚合逻辑但复杂度上升。我选订单 ID。

最后定时间语义。订单事件自带业务时间,必须用事件时间定时器,否则回放历史数据时行为不可复现。

3.2 完整实现代码

public class OrderTimeoutFunction extends KeyedProcessFunction<Long, OrderEvent, OrderResult> { private transient ValueState<OrderEvent> orderState; private transient ValueState<Long> fireAtState; private transient ValueState<Boolean> paidState; private static final long TIMEOUT_MS = 30 * 60 * 1000L; @Override public void open(Configuration parameters) { orderState = getRuntimeContext().getState( new ValueStateDescriptor<>("orderInfo", OrderEvent.class)); fireAtState = getRuntimeContext().getState( new ValueStateDescriptor<>("fireAt", Long.class)); paidState = getRuntimeContext().getState( new ValueStateDescriptor<>("paid", Boolean.class)); } @Override public void processElement(OrderEvent value, Context ctx, Collector<OrderResult> out) throws Exception { String type = value.getType(); if ("CREATE".equals(type)) { // 幂等:重复创建只保留第一次 if (orderState.value() != null) { return; } orderState.update(value); paidState.update(false); long fireAt = value.getEventTime() + TIMEOUT_MS; ctx.timerService().registerEventTimeTimer(fireAt); fireAtState.update(fireAt); } else if ("PAY".equals(type)) { Long fireAt = fireAtState.value(); if (fireAt != null) { ctx.timerService().deleteEventTimeTimer(fireAt); } paidState.update(true); out.collect(new OrderResult(value.getOrderId(), "PAID", value.getEventTime())); // 注意:这里不清 state,交给定时器统一清,见 3.3 的分析 } else if ("CANCEL".equals(type)) { Long fireAt = fireAtState.value(); if (fireAt != null) { ctx.timerService().deleteEventTimeTimer(fireAt); } orderState.clear(); fireAtState.clear(); paidState.clear(); } } @Override public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderResult> out) throws Exception { Boolean paid = paidState.value(); OrderEvent order = orderState.value(); if (order != null && (paid == null || !paid)) { out.collect(new OrderResult(order.getOrderId(), "TIMEOUT", timestamp)); } orderState.clear(); fireAtState.clear(); paidState.clear(); } }

3.3 逐行解读与几个必须注意的点

先说"CREATE"分支里的幂等判断。上游消息重复是常态,MQ 的至少一次投递、上游重跑、网络重试都会造成重复。如果不判空就直接registerEventTimeTimer,虽然同 key 同时间戳会去重,不会重复触发,但状态会被覆盖,一旦两次创建消息的时间字段有微小差异,就会注册出两个不同时间戳的定时器,触发两次关单。所以幂等必须做在前面。

再说"PAY"分支为什么不清状态。第一版我是在支付成功时直接把三个状态全清掉的,结果测试时发现:一个订单支付后又来了一条CANCEL,由于fireAtState已经是空,删除定时器成了空操作,原来的定时器还在队列里,到点触发onTimer时paidState也是空,被当成未支付,又发了一条超时关单消息,把已经支付的订单给关了。修复方式是保留paidState标记,让定时器自己判断并清理。这个坑非常隐蔽,只在"支付后又取消"这种路径下出现,而这条路径恰好覆盖率极低。

第三个点是onTimer里的清理。所有状态必须在这里清干净,否则 key 永远不会被释放,状态会一直累积到作业重启为止。我建议把清理逻辑写成一段独立方法,onTimer和业务分支里都调用,避免漏字段。

第四个点是侧输出流。真实场景里超时关单不能直接丢主输出,要落到一个单独的OutputTag,方便下游做差异处理。定义OutputTag时有个经典陷阱:匿名内部类写法会丢失泛型信息,导致运行时TypeInformation推断失败。

// 正确:显式子类,保留泛型信息 private static final OutputTag<OrderResult> TIMEOUT_TAG = new OutputTag<OrderResult>("order-timeout") {};

第五个是状态过期兜底策略。如果某些订单的支付事件永远不来,也没有水位线推进(比如上游断流),onTimer就永远不触发,状态会一直挂着。我的做法是给orderState加 TTL,配置成超时时间的 2 倍,让状态后端在压缩时顺手清掉过期条目,作为第二道保险。

4. 实战二:自定义会话切分与乱序去重

4.1 业务规则与数据流设计

第二个场景更有意思:一个内容平台要统计"用户一次连续浏览行为"的时长和深度。规则是,同一用户相邻两次有效浏览之间的间隔超过 10 分钟,就认为会话断开,输出上一段会话的统计结果。有效浏览指的是商品详情页和搜索页,首页曝光不算。

这个需求用官方会话窗口做不了,因为"哪些事件计入会话"是业务条件,窗口的 gap 只能基于时间。用KeyedProcessFunction就自然很多:key 用用户 ID,状态存"当前会话开始时间 + 已浏览条数 + 最后活动时间",每次来事件先判断类型再判断间隔。

乱序是这里的主要矛盾。用户 10:00 和 10:05 各浏览了一次,但 10:05 的数据先到,10:00 的后到。如果只比较"当前元素时间"和"最后活动时间",会把 10:00 的当成一个新会话,统计就错了。处理思路是:维护"本会话已见过的事件时间集合"和"最后活动时间(取最大值)",新元素到达时,如果它的时间早于会话起点超过容忍阈值,直接丢到侧输出作为迟到数据;如果落在容忍范围内,并入会话并更新时间边界,但不改变会话起点。

4.2 会话统计的代码骨架

public class SessionizeFunction extends KeyedProcessFunction<String, BrowseEvent, SessionResult> { private transient ValueState<SessionState> sessionState; private static final long GAP_MS = 10 * 60 * 1000L; @Override public void open(Configuration parameters) { sessionState = getRuntimeContext().getState( new ValueStateDescriptor<>("session", SessionState.class)); } @Override public void processElement(BrowseEvent value, Context ctx, Collector<SessionResult> out) throws Exception { if (!value.isValid()) { return; } SessionState s = sessionState.value(); long et = value.getEventTime(); if (s == null) { s = newSession(value, et); sessionState.update(s); ctx.timerService().registerEventTimeTimer(et + GAP_MS); return; } // 迟到但仍在容忍范围内:并入,不重建 if (et <= s.getLastActivity()) { s.setLastActivity(Math.max(s.getLastActivity(), et)); s.setCount(s.getCount() + 1); sessionState.update(s); return; } if (et - s.getLastActivity() <= GAP_MS) { s.setLastActivity(et); s.setCount(s.getCount() + 1); sessionState.update(s); ctx.timerService().registerEventTimeTimer(et + GAP_MS); } else { // 会话断裂:先输出旧的,再开新的 out.collect(s.toResult()); SessionState ns = newSession(value, et); sessionState.update(ns); ctx.timerService().registerEventTimeTimer(et + GAP_MS); } } @Override public void onTimer(long timestamp, OnTimerContext ctx, Collector<SessionResult> out) throws Exception { SessionState s = sessionState.value(); if (s != null && timestamp >= s.getLastActivity() + GAP_MS) { out.collect(s.toResult()); sessionState.clear(); } // 否则说明会话已被续期,什么都不做 } private SessionState newSession(BrowseEvent e, long et) { SessionState s = new SessionState(); s.setUserId(e.getUserId()); s.setStartTime(et); s.setLastActivity(et); s.setCount(1); return s; } }

这段代码里有个容易被忽略的设计:每次会话续期都重新注册一个定时器,但旧的定时器没有删除。这不是疏忽,而是刻意为之。因为定时器的去重是按时间戳的,续期注册的是新时间戳,旧的定时器到期时,onTimer里的timestamp >= lastActivity + GAP判断不成立,自然什么都不做,相当于空转一次。代价是定时器数量变多,收益是不需要在删除时小心翼翼地对齐时间戳。在会话场景里这个取舍是划算的,因为会话跨度短、续期次数有限;但在超时时间很长的场景里(比如 24 小时),我建议还是老实删旧的,否则定时器堆积会很可观。

4.3 水位线与迟到数据的配合

会话场景对水位线非常敏感。如果你的水位线策略是"周期性插入当前最大事件时间减 1 分钟",那么一个冷门用户的会话可能在水位线推进到很久以后才触发,统计延迟会很明显。反过来,如果水位线推得太激进,乱序数据会被当成迟到数据丢掉。

我的经验是分两步调:先看监控里"事件时间与处理时间的差值分布",确定乱序的 P99 大概是多少秒;再把这个值加上一点余量作为forBoundedOutOfOrderness的参数。不要凭感觉填 30 秒或者 1 分钟,这个值直接决定了状态驻留时间和输出延迟。

另外,onTimer里的判断条件timestamp >= s.getLastActivity() + GAP_MS是必须的。去掉它就会在最坏情况下提前关闭还没结束的会话,因为水位线推进可能让定时器在会话仍在续期时触发。这类"定时器触发但业务条件不满足"的分支,一定要显式处理,不能想当然认为"注册了就一定会在我期望的时刻触发"。

5. 常见问题排查:那些只在生产环境冒出来的问题

5.1 定时器不触发或触发次数不对

这是被问得最多的一类。我把排查顺序固化成了四步,基本能覆盖九成场景。

第一步看时间语义有没有搞混。作业的TimeCharacteristic(新版里是直接在env上配setStreamTimeCharacteristic或依赖水位线)和定时器注册方法必须匹配,事件时间定时器只能被水位线触发,处理时间定时器只能被系统时钟触发。两者混着用,不触发是必然的。

第二步看水位线是否推进。打开 Web UI 的 Watermark 指标,如果某个子任务的输入水位线长期不动,那它上面的定时器也就不会动。常见原因是数据源分片空闲、上游算子阻塞、或者数据倾斜导致个别子任务没数据。

第三步看 key 是否真的落在你预期的子任务上。有个很经典的误解:以为定时器是"全局队列"。实际上它是按 key 归属到子任务的,onTimer只会在这个子任务的线程里执行,你在onTimer里访问的外部连接池必须是线程安全的、并且已经被这个子任务初始化过。

第四步看时间戳单位。这个听起来很低级,但我见过两次:一次是业务字段是秒级时间戳,有人直接当毫秒用注册,结果定时器在几十年后才触发;另一次是时区转换把时间戳整体偏移了 8 小时。

注意:定时器触发的前提是水位线越过注册时间戳,是"越过"不是"相等"。如果你的注册时间戳恰好比水位线大一点点,而数据流已经结束,定时器就永远不会触发,直到作业被取消。

5.2 状态膨胀与 checkpoint 变大

状态膨胀的典型症状是 checkpoint 耗时从几百毫秒涨到几秒、恢复时间从分钟级涨到十分钟以上、RocksDB 的liveSstFilesSize持续上涨。遇到这种情况,先别急着扩资源,按下面的顺序查。

先查有没有清理状态。onTimer里清了没?异常分支里清了没?"CANCEL"这类提前结束的分支清了没?我见过最典型的一段代码是在processElement里update了状态,但在删除定时器的分支里只调了deleteEventTimeTimer而忘了clear,跑了三天状态涨了 40 个 G。

再查 key 的基数。如果你的 key 是"用户 ID + 日期"这种组合,每天自然会新增一批 key,历史 key 的状态如果不清理就是永久驻留。这种情况要么换成"用户 ID"当 key 配合状态内的时间窗口,要么给状态配 TTL。

最后查定时器数量。定时器也是状态的一部分,而且它的数量往往比业务状态更难直观感知。一个简单的方法是打开定时器相关的指标,或者临时在onTimer里打点统计,粗略估算单位时间的定时器增量和存量。存量超过千万级就该认真考虑方案调整了。

StateTtlConfig ttl = StateTtlConfig .newBuilder(Time.hours(2)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .cleanupInRocksdbCompactFilter(1000) .build(); ValueStateDescriptor<OrderEvent> desc = new ValueStateDescriptor<>("orderInfo", OrderEvent.class); desc.enableTimeToLive(ttl);

需要说清楚的是,TTL 不是即时删除,它依赖读取时判断和后台清理,所以"状态不会立刻变小"是正常的。NeverReturnExpired保证读不到过期数据,配上 RocksDB 的压缩过滤清理,才能在存储层面真正回收。如果你的阈值很紧,别指望 TTL 立刻见效。

5.3 问题速查表

提示:排查时尽量在测试环境复现,用最小数据集回放。定时器问题一旦上了生产,重启作业会导致状态恢复,而恢复过程中的行为和生产当前状态往往不一致,越查越乱。

6. 上线前的取舍:性能、并行度与清单

6.1 状态后端与定时器的规模匹配

定时器的存储位置和状态后端绑定,这个选型决策要在写代码之前就定下来,因为换后端意味着作业重启加状态迁移。我的判断标准比较粗暴:估算"峰值时刻同时存活的定时器数量",百万以下用堆内存后端,响应快、调试方便;超过千万毫不犹豫上 RocksDB 后端,避免 GC 抖动和 OOM。

估算方法很简单:超时时间是 T 秒,峰值 TPS 是 Q,那么稳态存活数量大约是Q * T。比如关单场景峰值 5 万 TPS、超时 30 分钟,那就是 900 万个定时器,这个量级已经该认真考虑后端选择了。如果业务允许,把超时时间从 30 分钟压到 10 分钟,数量直接降到三分之一,这是最省事的优化,比调参有效得多。

另外一个常被忽略的点是 key 的分布。KeyedProcessFunction的所有状态和定时器都跟着 key 走,如果某个大客户的订单占了总量的一半,那个子任务就会成为瓶颈,表现为 checkpoint 时长被一个子任务拖住(检查点的对齐等待)。这种倾斜不是靠加并行度能解决的,得在 key 上加盐或者拆分。

6.2 并行度、槽位与回调里的外部调用

并行度这块有一个容易犯的错误:因为有 keyBy,很多人以为并行度随便调都行。实际上maxParallelism一旦确定(默认或者显式设置),key 分组的数量就固定了,并行度只能是它的约数,而且调整并行度时 key 分组的分配会变化。如果你改过 key 的结构(比如从"用户ID"换成"用户ID+日期"),最好重新设定maxParallelism并做一次全量重跑,而不是直接在线改。

更值得注意的是onTimer里的外部调用。这里执行的是同步逻辑,任何一次慢查询都会直接阻塞整个子任务的处理,连带把水位线推进也拖住。我的做法是把外部调用全部改成异步客户端加回调,或者在onTimer里只做状态更新和发消息,把耗时操作推到下游算子。这一条是用血换来的:有一次在onTimer里同步调了一个风控接口,对方超时时间设了 2 秒,赶上大促那天接口抖动,整个子任务卡了十几分钟,消息积压到告警。

6.3 上线清单

下面这份清单是我自己每次上线前会过一遍的,按顺序检查能省掉大部分回滚。

最后分享一个我在实际项目里养成的小习惯:给每个KeyedProcessFunction加一个调试开关,通过open()里的参数读取,打开后在每次processElement和onTimer打一行带 key、时间戳、状态摘要的日志,默认关闭。上线后遇到疑难问题,临时把开关打开重跑一小段数据,比在代码里加删日志快得多。这个开关本身不占资源,但救急的时候特别好用。