flink离线任务定时器使用

Flink提供了两种类型的定时器：事件时间定时器和处理时间定时器。

事件时间定时器基于事件时间，可以在数据流中插入一个事件时间戳，一旦到达指定时间，就会触发定时器。事件时间定时器适用于需要等待一段时间以获取完整结果的场景，例如处理窗口。

处理时间定时器基于处理时间，可以在数据到达后的固定时间间隔后触发定时器。处理时间定时器适用于需要在一定时间间隔内执行某个操作的场景，例如清理过期数据。

下面是一个使用处理时间定时器的示例代码：

DataStreamSource<String> source = env.fromElements("a", "b", "c");

source
    .map(new MapFunction<String, Tuple2<String, Long>>() {
        @Override
        public Tuple2<String, Long> map(String value) throws Exception {
            return Tuple2.of(value, System.currentTimeMillis() + 10000L);
        }
    })
    .keyBy(0)
    .process(new KeyedProcessFunction<String, Tuple2<String, Long>, String>() {
        @Override
        public void processElement(Tuple2<String, Long> value, Context ctx, Collector<String> out) throws Exception {
            // 注册处理时间定时器，10秒后触发
            ctx.timerService().registerProcessingTimeTimer(value.f1);

            // 保存状态
            ValueState<String> state = getRuntimeContext().getState(new ValueStateDescriptor<>("value-state", String.class));
            state.update(value.f0);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            // 处理时间定时器触发，输出状态中的值
            ValueState<String> state = getRuntimeContext().getState(new ValueStateDescriptor<>("value-state", String.class));
            out.collect(state.value());
        }
    })
    .print();

env.execute();

在上面的示例中，我们先从一个字符串数据源中获取数据，并使用MapFunction将每个字符串与当前时间戳组成一个Tuple2。然后，我们使用KeyedProcessFunction对每个Tuple2进行处理，其中我们注册了处理时间定时器，并将状态保存在ValueState中。当定时器触发时，我们可以从状态中获取值并输出。

需要注意的是，处理时间定时器的触发时间是相对于 Flink JobManager 的机器时间的，而不是相对于数据流中的事件时间。因此，在使用处理时间定时器时应特别注意处理时间与事件时间之间的差异。