&&大数据学习&&

🔥系列专栏： 👑哲学语录: 承认自己的无知，乃是开启智慧的大门
💖如果觉得博主的文章还不错的话，请点赞👍+收藏⭐️+留言📝支持一下博主哦🤞

---

之前提到，只有在KeyedStream中才支持使用TimerService设置定时器的操作。所以一般情况下，我们都是先做了keyBy分区之后，再去定义处理操作；代码中更加常见的处理函数是KeyedProcessFunction。

1 定时器（Timer）和定时服务（TimerService）

在.onTimer()方法中可以实现定时处理的逻辑，而它能触发的前提，就是之前曾经注册过定时器、并且现在已经到了触发时间。注册定时器的功能，是通过上下文中提供的“定时服务”来实现的。

定时服务与当前运行的环境有关。前面已经介绍过，ProcessFunction的上下文（Context）中提供了.timerService()方法，可以直接返回一个TimerService对象。TimerService是Flink关于时间和定时器的基础服务接口，包含以下六个方法：

// 获取当前的处理时间long currentProcessingTime();// 获取当前的水位线（事件时间）long currentWatermark();// 注册处理时间定时器，当处理时间超过time时触发void registerProcessingTimeTimer(long time);// 注册事件时间定时器，当水位线超过time时触发void registerEventTimeTimer(long time);// 删除触发时间为time的处理时间定时器void deleteProcessingTimeTimer(long time);// 删除触发时间为time的处理时间定时器void deleteEventTimeTimer(long time);

六个方法可以分成两大类：基于处理时间和基于事件时间。而对应的操作主要有三个：获取当前时间，注册定时器，以及删除定时器。需要注意，尽管处理函数中都可以直接访问TimerService，不过只有基于KeyedStream的处理函数，才能去调用注册和删除定时器的方法；未作按键分区的DataStream不支持定时器操作，只能获取当前时间。

TimerService会以键（key）和时间戳为标准，对定时器进行去重；也就是说对于每个key和时间戳，最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次。

2 KeyedProcessFunction案例

基于keyBy之后的KeyedStream，直接调用.process()方法，这时需要传入的参数就是KeyedProcessFunction的实现类。

stream.keyBy( t -> t.f0 ).process(new MyKeyedProcessFunction())

类似地，KeyedProcessFunction也是继承自AbstractRichFunction的一个抽象类，与ProcessFunction的定义几乎完全一样，区别只是在于类型参数多了一个K，这是当前按键分区的key的类型。同样地，我们必须实现一个.processElement()抽象方法，用来处理流中的每一个数据；另外还有一个非抽象方法.onTimer()，用来定义定时器触发时的回调操作。

代码如下：

public class KeyedProcessTimerDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 7777).map(new WaterSensorMapFunction()).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((element, ts) -> element.getTs() * 1000L));KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());// TODO Process:keyedSingleOutputStreamOperator<String> process = sensorKS.process(new KeyedProcessFunction<String, WaterSensor, String>() {/*** 来一条数据调用一次* @param value* @param ctx* @param out* @throws Exception*/@Overridepublic void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {//获取当前数据的keyString currentKey = ctx.getCurrentKey();// TODO 1.定时器注册TimerService timerService = ctx.timerService();// 1、事件时间的案例Long currentEventTime = ctx.timestamp(); // 数据中提取出来的事件时间timerService.registerEventTimeTimer(5000L);System.out.println("当前key=" + currentKey + ",当前时间=" + currentEventTime + ",注册了一个5s的定时器");// 2、处理时间的案例//                        long currentTs = timerService.currentProcessingTime();//                        timerService.registerProcessingTimeTimer(currentTs + 5000L);//                        System.out.println("当前key=" + currentKey + ",当前时间=" + currentTs + ",注册了一个5s后的定时器");// 3、获取 process的 当前watermark//                        long currentWatermark = timerService.currentWatermark();//                        System.out.println("当前数据=" + value + ",当前watermark=" + currentWatermark);// 注册定时器： 处理时间、事件时间//                        timerService.registerProcessingTimeTimer();//                        timerService.registerEventTimeTimer();// 删除定时器： 处理时间、事件时间//                        timerService.deleteEventTimeTimer();//                        timerService.deleteProcessingTimeTimer();// 获取当前时间进展： 处理时间-当前系统时间，  事件时间-当前watermark//                        long currentTs = timerService.currentProcessingTime();//                        long wm = timerService.currentWatermark();}/*** TODO 2.时间进展到定时器注册的时间，调用该方法* @param timestamp 当前时间进展，就是定时器被触发时的时间* @param ctx       上下文* @param out       采集器* @throws Exception*/@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {super.onTimer(timestamp, ctx, out);String currentKey = ctx.getCurrentKey();System.out.println("key=" + currentKey + "现在时间是" + timestamp + "定时器触发");}});process.print();env.execute();}}