书史足自悦，安用勤与劬。这篇文章主要讲述结合案例总结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(

推荐阅读

• 

  蚌埠是哪个省的城市
• 

  模具平衡块的作用 平衡块的作用
• 

  网络技术|实现内网通过外网域名访问NAT映射的内网服务器
• 

  数据分析好找工作吗,python数据分析好找工作吗
• 

  lg冰箱236不制冷是什么故障,看完以后终于明白
• 

  佳期如梦意思 佳期如梦说的是什么
• 

  技术管理进阶——第三个五年，独立思考与落地实操
• 

  萝卜的做法大全家常菜
• 

  fly100,FLY100%打魔兽时喜欢听的歌
• 

  lols6盲僧李青瞎子打野天赋符文加点图
• 

  chatgpt数学建模，数学建模clear
• 

  老年人春季养生4秘诀 老年人春季如何养生
• 

  去眼袋手术的价格表
• 

  什么是茶艺、茶道、茶文化？三者之间又有什么关系呢？
• 

  mongodb唯一键 mongodb主键保存
• 

  吃桑葚有什么副作用
• 

  幽灵漏洞原理分析,源代码漏洞分析原理
• 

  分泌性中耳炎能转变为化脓性中耳炎吗？
• 

  奶茶妹妹南外6年、清华4年、哥大1年,你觉得她的英语口语怎么样？
• 

  齐白石四季山水十二条屏 山水十二条屏

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