ToB企服应用市场:ToB评测及商务社交产业平台

标题: Flink中定时器的使用 [打印本页]

---

作者: 大连密封材料    时间: 2024-7-18 13:16
标题: Flink中定时器的使用
定时器

让 Flink 流处理步伐对处理时间和变乱时间的变化作出反应。通常在KeyProcessFunction中使用。
基于处理时间或者变乱时间处理过一个元素之后，注册一个定时器，然后在指定的时间运行。
最常见的显式使用Timer的地方就是KeyedProcessFunction。我们在其processElement()方法中注册Timer，然后覆写其onTimer()方法作为Timer触发时的回调逻辑。根据时间特征的差别：

• 处理时间——调用Context.timerService().registerProcessingTimeTimer()注册；onTimer()在体系时间戳到达Timer设定的时间戳时触发。
  
• 变乱时间——调用Context.timerService().registerEventTimeTimer()注册；onTimer()在Flink内部水印到达或凌驾Timer设定的时间戳时触发。
  

Context和OnTimerContext所持有的TimerService对象拥有以下方法：

• currentProcessingTime(): Long 返回当前处理时间
  
• currentWatermark(): Long 返回当前watermark的时间戳
  
• registerProcessingTimeTimer(timestamp: Long): Unit 会注册当前key的processing time的定时器。当processing time到达指定时间时，触发timer。
  
• registerEventTimeTimer(timestamp: Long): Unit 会注册当前key的event time 定时器。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。
  
• deleteProcessingTimeTimer(timestamp: Long): Unit 删除之前注册处理时间定时器。假如没有这个时间戳的定时器，则不执行。
  
• deleteEventTimeTimer(timestamp: Long): Unit 删除之前注册的变乱时间定时器，假如没有此时间戳的定时器，则不执行。
  

1.基于处理时间的定时器

1. SingleOutputStreamOperator<WaterSensor> stream = env
2.   .socketTextStream("hadoop102", 9999)  // 在socket终端只输入毫秒级别的时间戳
3.   .map(value -> {
4.       String[] datas = value.split(",");
5.       return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
7.   });
9. stream
10.   .keyBy(WaterSensor::getId)
11.   .process(new KeyedProcessFunction<String, WaterSensor, String>() {
12.       @Override
13.       public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
14.           // 处理时间过后5s后触发定时器
15.           ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 5000);
16.           out.collect(value.toString());
17.       }
20.       // 定时器被触发之后, 回调这个方法
21.       // 参数1: 触发器被触发的时间
22.       @Override
23.       public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
24.           System.out.println(timestamp);
25.           out.collect("我被触发了....");
26.       }
27.   })
28.   .print();

复制代码

2.基于变乱时间的定时器
变乱进展依据的是watermark

1. SingleOutputStreamOperator<WaterSensor> stream = env
2.   .socketTextStream("hadoop102", 9999)  // 在socket终端只输入毫秒级别的时间戳
3.   .map(value -> {
4.       String[] datas = value.split(",");
5.       return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
7.   });
9. WatermarkStrategy<WaterSensor> wms = WatermarkStrategy
10.   .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
11.   .withTimestampAssigner((element, recordTimestamp) -> element.getTs() * 1000);
12. stream
13.   .assignTimestampsAndWatermarks(wms)
14.   .keyBy(WaterSensor::getId)
15.   .process(new KeyedProcessFunction<String, WaterSensor, String>() {
16.       @Override
17.       public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
18.           System.out.println(ctx.timestamp());
19.           ctx.timerService().registerEventTimeTimer(ctx.timestamp() + 5000);
20.           out.collect(value.toString());
21.       }
23.       @Override
24.       public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
25.           System.out.println("定时器被触发.....");
26.       }
27.   })
28.   .print();

复制代码

3.定时器练习
监控水位传感器的水位值，假如水位值在（处理时间）5秒内一连上上，则报警。

1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
2.         env.setParallelism(1);
4.         env.socketTextStream("hadoop162",9999)
5.                 .map(value->{
6.                     String[] split = value.split(",");
7.                     return new WaterSensor(split[0],Long.valueOf(split[1])*1000,Integer.valueOf(split[2]));
8.                 })
9.                 .assignTimestampsAndWatermarks(WatermarkStrategy.
10.                         <WaterSensor>forMonotonousTimestamps()
11.                     .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>(){
12.                         @Override
13.                         public long extractTimestamp(WaterSensor element, long recordTimestamp) {
14.                             return element.getTs();
15.                         }
16.                     })
17.                 )
18.                 .keyBy(WaterSensor::getId)
19.                 .process(new KeyedProcessFunction<String, WaterSensor, String>() {
21.                     private  long  time;
22.                     private  int  lastVc = 0;
23.                     private  boolean  isFirst = true;
25.                     @Override
26.                     public void processElement(WaterSensor value,
27.                                                Context ctx,
28.                                                Collector<String> out) throws Exception {
29.                         if (value.getVc() > lastVc){
30.                             if (isFirst){
31.                                 //水位上升
32.                                 time =ctx.timestamp() +5000L;
33.                                 ctx.timerService().registerEventTimeTimer(time);
34.                                 isFirst=false;
35.                             }
36.                         }else {
37.                             //水位下降或者不变
38.                             ctx.timerService().deleteEventTimeTimer(time);
40.                             time = ctx.timestamp() + 5000L;
41.                             ctx.timerService().registerEventTimeTimer(time);
42.                         }
43.                         lastVc = value.getVc();
44.                     }
46.                     @Override
47.                     public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
48.                         out.collect(ctx.getCurrentKey()+"水位已经连续5上升。。。。");
49.                         isFirst=true;
50.                     }
51.                 })
52.                 .print();
53.                
54.         try {
55.             env.execute();
56.         } catch (Exception e) {
57.             e.printStackTrace();
58.         }
59.     }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4