当前位置：首页 > news >正文

Flink Processing Timer Service 用处理时间把“准点任务”写进流里

news 2025/10/31 7:41:18

1. 处理时间定时器是什么？

处理时间（Processing Time） 是 TaskManager 机器的墙钟时间。处理时间定时器允许你在“某个具体的处理时刻”触发回调来执行计算。与 事件时间（Event Time） 不同，它不依赖数据里的事件时间戳和水位线，而是“到点就执行”。

典型用途

周期性 flush 内存聚合结果（比如每 1 分钟批量写一次外部库）。
防抖/节流：延迟一小段时间后再提交，聚合短时间内的多次更新。
空闲/心跳检测：X 分钟未收到该 Key 的数据则触发告警/回收状态。
缓存/状态清理：给状态设置软 TTL，到期清理，节省内存。

2. API 速览：ProcessingTimeManager

在 ProcessFunction 里，通过 PartitionedContext 获取 ProcessingTimeManager：

@Experimental
public interface ProcessingTimeManager {void registerTimer(long timestamp);  // 注册处理时间定时器（毫秒时间戳）void deleteTimer(long timestamp);    // 删除尚未触发的定时器long currentTime();                  // 当前处理时间（毫秒）
}

重点规则

同一 Key、同一目标时间只保留一个定时器；
只能在 Keyed Partitioned Stream 中使用（需要有“当前 Key”的概念）。

当达到 registerTimer 的目标时间时，Flink 会调用你的
ProcessFunction#onProcessingTimer(long ts, Collector<T> out, PartitionedContext<T> ctx) 回调。

3. 最小可用骨架

public class CustomProcessFunction implements OneInputStreamProcessFunction<String, String> {@Overridepublic void processRecord(String record, Collector<String> out, PartitionedContext<String> ctx) throws Exception {// 1) 业务计算（可选）// ...// 2) 注册一个“当前时间 + 1 分钟”的处理时间定时器long now = ctx.getProcessingTimeManager().currentTime();ctx.getProcessingTimeManager().registerTimer(now + Duration.ofMinutes(1).toMillis());}@Overridepublic void onProcessingTimer(long timestamp, Collector<String> out, PartitionedContext<String> ctx) {// 3) 到点回调：执行你的定时逻辑（聚合 flush / 清理 / 告警）// out.collect(...);}
}

4. 常见模式（拿来即用）

4.1 周期性 Flush（自驱动定时）

在回调里再次注册“下一次”的定时器，实现稳定的固定周期任务。

static final long PERIOD = Duration.ofSeconds(30).toMillis();@Override
public void processRecord(Rec r, Collector<Out> out, PartitionedContext<Rec> ctx) throws Exception {// 首次收到数据时，若无周期定时器，则启动一个long now = ctx.getProcessingTimeManager().currentTime();long next = (now / PERIOD + 1) * PERIOD; // 对齐到下一个 30s 边界ctx.getProcessingTimeManager().registerTimer(next);// ... 更新内存聚合/状态
}@Override
public void onProcessingTimer(long ts, Collector<Out> out, PartitionedContext<Rec> ctx) {// flush 当前 Key 的聚合结果// out.collect(...)// 安排下一次ctx.getProcessingTimeManager().registerTimer(ts + PERIOD);
}

要点：用“回调里再注册下一次”来实现长期稳态周期，而不是在 processRecord 里反复注册。

4.2 防抖（Debounce）：只在“沉默一段时间后”触发

多次更新密集到来，只在最后一次后的 X 毫秒触发一次。

// 假设我们用 ValueState<Long> 存储“已注册的触发时间”
static final ValueStateDeclaration<Long> DEBOUNCE_TS =StateDeclarations.valueState("debounce-ts", TypeDescriptors.LONG);@Override
public Set<StateDeclaration> usesStates() { return Set.of(DEBOUNCE_TS); }static final long QUIET = 2_000L; // 2 秒静默期@Override
public void processRecord(Rec r, Collector<Out> out, PartitionedContext<Rec> ctx) throws Exception {ValueState<Long> tsState = ctx.getStateManager().getState(DEBOUNCE_TS);ProcessingTimeManager tm = ctx.getProcessingTimeManager();// 1) 若之前注册过定时器，先撤销Long oldTs = tsState.value();if (oldTs != null) tm.deleteTimer(oldTs);// 2) 以“当前时间 + 静默期”注册新的定时器long triggerTs = tm.currentTime() + QUIET;tm.registerTimer(triggerTs);tsState.update(triggerTs);// 3) 累加/变更状态（用于回调时一次性处理）// ...
}@Override
public void onProcessingTimer(long ts, Collector<Out> out, PartitionedContext<Rec> ctx) throws Exception {// 真正的提交/落库// ...// 清理记录的触发时间ctx.getStateManager().getState(DEBOUNCE_TS).clear();
}

要点：借助 deleteTimer 实现“只保留最后一次”。

4.3 Key 空闲检测（Inactivity）

若某个 Key 在 X 分钟内没数据，就触发清理或报警。

static final ValueStateDeclaration<Long> EXPIRE_TS =StateDeclarations.valueState("expire-ts", TypeDescriptors.LONG);
static final long IDLE = Duration.ofMinutes(5).toMillis();@Override
public Set<StateDeclaration> usesStates() { return Set.of(EXPIRE_TS); }@Override
public void processRecord(Rec r, Collector<Out> out, PartitionedContext<Rec> ctx) throws Exception {ValueState<Long> exp = ctx.getStateManager().getState(EXPIRE_TS);ProcessingTimeManager tm = ctx.getProcessingTimeManager();Long old = exp.value();if (old != null) tm.deleteTimer(old);long next = tm.currentTime() + IDLE;tm.registerTimer(next);exp.update(next);// 更新该 Key 的业务状态...
}@Override
public void onProcessingTimer(long ts, Collector<Out> out, PartitionedContext<Rec> ctx) throws Exception {// X 分钟未活跃，做清理/告警// out.collect(alert(...));// 清理业务状态与过期时间ctx.getStateManager().getState(EXPIRE_TS).clear();// ... clear other states
}