书史足自悦，安用勤与劬。这篇文章主要讲述结合案例总结Flink框架中的最底层API（ProcessFunction)用法相关的知识，希望能为你提供帮助。

概述在之前总结的文章中有提到过，Flink框架提供了三层API完成流处理任务。至此已经学习了DataStream API ，ProcessFunction API 是Flink中最底层的API，可以访问时间戳、watermark 以及注册定时事件。还可以输出特定的一些事件。、

文章图片

【结合案例总结Flink框架中的最底层API（ProcessFunction)用法】Process Function 用来构建事件驱动的应用以及实现自定义的业务逻辑，若窗口函数以及转换算子都无法满足业务的要求时，需要请出ProcessFunction 去完成开发任务。Flink SQL 就是使用 Process Function 实现的。
Flink 提供了 8 个 Process Function如下：??ProcessFunction???、??KeyedProcessFunction???、??CoProcessFunction???、??ProcessJoinFunction???、??BroadcastProcessFunction???、??KeyedBroadcastProcessFunction???、??ProcessWindowFunction???、??ProcessAllWindowFunction??。接下来我们以KeyedProcessFunction为例来进行学习。
KeyedProcessFunction< K, I, O> 它主要用来操作KeyedStream，会处理流的每一个元素，输出为 0 个、1 个或者多个元素。所有的 Process Function 都继承自RichFunction 接口，所以都有 open()、close()和 getRuntimeContext()等方法。除此之外还提供了两个方法：

• 数据流中的每一个元素都会调用这个方法，调用结果将会放在 Collector 数据类型中输出。Context 可以访问元素的时间戳，元素的key，以及 TimerService 时间服务。Context 还可以将结果输出到别的流(side outputs)。

processElement(I& nbsp; value,& nbsp; Context& nbsp; ctx,& nbsp; Collector& lt; O& gt; & nbsp; out)

• 当之前注册的定时器触发时调用。参数 timestamp 为定时器所设定的触发的时间戳。Collector为输出结果的集合。OnTimerContext 和processElement 的 Context 参数一样，提供了上下文的一些信息。例如定时器触发的时间信息(事件时间或者处理时间)。

onTimer(long& nbsp; timestamp,& nbsp; OnTimerContext& nbsp; ctx,& nbsp; Collector& lt; O& gt; & nbsp; out)

定时器Context 中 TimerService对象方汇总：

• 返回当前处理时间

long& nbsp; currentProcessingTime()

• 返回当前 watermark 的时间戳

long& nbsp; currentWatermark()

• 注册当前 key 的processing time 的定时器，当 processing time 到达定时时间时，触发 timer。

void& nbsp; registerProcessingTimeTimer(long& nbsp; timestamp)

• 注册当前 key 的 event time 定时器。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。

void& nbsp; registerEventTimeTimer(long& nbsp; timestamp)

• 删除之前注册处理时间定时器。如果没有这个时间戳的定时器，则不执行

void& nbsp; deleteProcessingTimeTimer(long& nbsp; timestamp)

• 删除之前注册的事件时间定时器，如果没有此时间戳的定时器，则不执行。

void& nbsp; deleteEventTimeTimer(long& nbsp; timestamp)

测试代码：

public& nbsp; class& nbsp; ProcessTest1_KeyedProcessFunction& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; static& nbsp; void& nbsp; main(String[]& nbsp; args)& nbsp; throws& nbsp; Exception
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; StreamExecutionEnvironment& nbsp; env& nbsp; =& nbsp; StreamExecutionEnvironment.getExecutionEnvironment();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; env.setParallelism(1);

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; DataStreamSource& lt; String& gt; & nbsp; inputStream& nbsp; =& nbsp; env.socketTextStream(& quot; localhost& quot; ,& nbsp; 7777);
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; DataStream& lt; SensorReading& gt; & nbsp; dataStream& nbsp; =& nbsp; inputStream.map(line& nbsp; -& gt; & nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; String[]& nbsp; fields& nbsp; =& nbsp; line.split(& quot; ,& quot; );
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; return& nbsp; new& nbsp; SensorReading(fields[0],& nbsp; new& nbsp; Long(fields[1]),& nbsp; new& nbsp; Double(fields[2]));
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; );

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //测试keyedprocessFunction  先分组，自定义处理
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; dataStream.keyBy(& quot; id& quot; )
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; .process(new& nbsp; MyProcess())
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; .print();


& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; env.execute();
& nbsp; & nbsp; & nbsp; & nbsp;

& nbsp; & nbsp; & nbsp; & nbsp; //  实现自定义处理函数
& nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; static& nbsp; class& nbsp; MyProcess& nbsp; extends& nbsp; KeyedProcessFunction& lt; Tuple,SensorReading,Integer& gt; & nbsp;

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ValueState& lt; Long& gt; & nbsp; tsTimeState;

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; @Override
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; void& nbsp; open(Configuration& nbsp; parameters)& nbsp; throws& nbsp; Exception& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; tsTimeState& nbsp; =& nbsp; getRuntimeContext().getState(new& nbsp; ValueStateDescriptor& lt; Long& gt; (
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & quot; tsTimeState& quot; ,Long.class
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ));
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp;

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; @Override
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; void& nbsp; processElement(SensorReading& nbsp; value,& nbsp; Context& nbsp; ctx,& nbsp; Collector& lt; Integer& gt; & nbsp; out)& nbsp; throws& nbsp; Exception& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; out.collect(value.getId().length());

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //  Context操作
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.timestamp();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.getCurrentKey();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //  侧流
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //ctx.output();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //  获取当前系统处理时间
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.timerService().currentProcessingTime();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //  获取当前事件时间
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.timerService().currentWatermark();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //  注册系统处理时间定时器
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.timerService().registerProcessingTimeTimer(& nbsp; ctx.timerService().currentProcessingTime()& nbsp; +& nbsp; 1000L);
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; tsTimeState.update(& nbsp; ctx.timerService().currentProcessingTime()& nbsp; +& nbsp; 1000L);

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //  注册事件时间定时器
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //ctx.timerService().registerEventTimeTimer((value.getTimestamp()  +  10)  *  1000L);
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //  删除时间
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //ctx.timerService().deleteProcessingTimeTimer(tsTimeState.value());
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp;

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; @Override
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; void& nbsp; onTimer(long& nbsp; timestamp,& nbsp; OnTimerContext& nbsp; ctx,& nbsp; Collector& lt; Integer& gt; & nbsp; out)& nbsp; throws& nbsp; Exception& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; System.out.println(timestamp+& quot; 定时器触发& quot; );

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.getCurrentKey();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //ctx.output();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.timeDomain();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp;

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; @Override
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; void& nbsp; close()& nbsp; throws& nbsp; Exception& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; tsTimeState.clear();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp;
& nbsp; & nbsp; & nbsp; & nbsp;

一个案例：

监控温度传感器的温度值，如果温度值在 10 秒钟之内连续上升，则报警。

public& nbsp; class& nbsp; ProcessTest2_ApplicationCase& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; static& nbsp; void& nbsp; main(String[]& nbsp; args)& nbsp; throws& nbsp; Exception& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; StreamExecutionEnvironment& nbsp; env& nbsp; =& nbsp; StreamExecutionEnvironment.getExecutionEnvironment();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; env.setParallelism(1);

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; DataStreamSource& lt; String& gt; & nbsp; inputStream& nbsp; =& nbsp; env.socketTextStream(& quot; localhost& quot; ,& nbsp; 7777);
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; DataStream& lt; SensorReading& gt; & nbsp; dataStream& nbsp; =& nbsp; inputStream.map(line& nbsp; -& gt; & nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; String[]& nbsp; fields& nbsp; =& nbsp; line.split(& quot; ,& quot; );
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; return& nbsp; new& nbsp; SensorReading(fields[0],& nbsp; new& nbsp; Long(fields[1]),& nbsp; new& nbsp; Double(fields[2]));
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; );


& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; dataStream.keyBy(& quot; id& quot; )
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; .process(

推荐阅读

• 

  单机的经营游戏，单机经营游戏推荐
• 

  黑夜使者
• 

  淮北市第49周楼市成交数据来了，备案83套！
• 

  苏轼的轼组词 轼出色组词
• 

  新买的毛衣怎么洗
• 

  面包冰箱里能放几天？解析面包在冰箱中的保鲜期
• 

  android android:duplicateParentState=&amp;quot;true&amp;quot; &amp;quot;false&amp;quot;
• 

  红烧鲳鱼怎么做好吃又简单 红烧鲳鱼的家常做法简单窍门
• 

  做整鼻子痛吗
• 

  残差收敛多大视为有限元计算收敛（）
• 

  哪里能学做蛋糕？
• 

  迷你世界2月26日激活码 迷你世界2022年2月26日礼包兑换码
• 

  骐达曲轴传感器位置图 骐达曲轴传感器位置图解
• 

  高跟鞋后跟怎么换鞋钉 高跟鞋后跟如何换鞋钉
• 

  小公司与大公司的Java程序员有什么差别（如何才能进大公司？）
• 

  苹果12怎么设置密码锁屏（苹果12密码锁屏设置）
• 

  怎么设置苹果主屏幕快捷键，苹果手机的主屏幕键怎么设置
• 

  大量微信群推广代发广告，哪里有代发群广告的工作室信誉一定要好的 有过合作或者正在合作
• 

  丧假工资怎么发
• 

  痘疤痘坑要怎么修复

• bat脚本系列---批量修改文件名
• jenkins系列---jenkins+gitee+vue持续集成到另一台服务器
• linux之同时监控多个日志文件变化
• elk
• web服务Tomcat优化之多实例
• 更改AD DC的电脑名称
• Mac 终端窗口的美化
• CentOS安装zip unzip命令
• CS162操作系统课程第二课-4个核心OS概念