flink多次调用registerProcessingTimeTimer的行为

如果在 flink 中多次调用 registerProcessingTimeTimer，它会覆盖之前的定时器，并重新安排一次计时器事件。解决该问题的一个方法是在调用之前检查定时器是否已存在，如果是，则取消该定时器并重新安排一个新的定时器。

以下是一个示例，演示如何在 Flink 中处理多个计时器的情况：

public class MyProcessFunction extends ProcessFunction<Integer, String> {

  private MapState<Long, String> timerState;

  @Override
  public void open(Configuration config) {
    MapStateDescriptor<Long, String> descriptor = new MapStateDescriptor<>(
      "timers",
      LongSerializer.INSTANCE,
      StringSerializer.INSTANCE);
    timerState = getRuntimeContext().getMapState(descriptor);
  }

  @Override
  public void processElement(Integer value, Context ctx, Collector<String> out) throws Exception {
    // get current processing time
    long currentTime = ctx.timerService().currentProcessingTime();

    // set timer for one minute later
    long timerTime = currentTime + (60 * 1000);

    // check if there is already a timer set for this time
    if (timerState.contains(timerTime)) {
      // if yes, cancel the existing timer
      ctx.timerService().deleteProcessingTimeTimer(timerTime);
    }

    // register new timer
    ctx.timerService().registerProcessingTimeTimer(timerTime);

    // store timer in state
    timerState.put(timerTime, "timer set for " + timerTime);
  }

  @Override
  public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
    // get message from state and emit
    String message = timerState.get(timestamp);
    out.collect(message);

    // remove timer from state
    timerState.remove(timestamp);
  }
}