在更底层，可以不定义任何具体的算子（比如 map，filter，或者 window），而只是提炼出一个统一的“处理”（process）操作——它是所有转换算子的一个概括性的表达，可以自定义处理逻辑，所以这一层接口就被叫作“处理函数”（process function）。
在处理函数中，我们直面的就是数据流中最基本的元素：数据事件（event）、状态（state）以及时间（time）。这就相当于对流有了完全的控制权。处理函数比较抽象，没有具体的操作，所以对于一些常见的简单应用（比如求和、开窗口）会显得有些麻烦；不过正是因为它不限定具体做什么，所以理论上我们可以做任何事情，实现所有需求。所以可以说，处理函数是我们进行 Flink 编程的“大招”，轻易不用，一旦放出来必然会扫平一切。
一、基本处理函数（ProcessFunction）
处理函数主要是定义数据流的转换操作，所以也可以把它归到转换算子中。我们知道在
Flink 中几乎所有转换算子都提供了对应的函数类接口，处理函数也不例外；它所对应的函数类，就叫作 ProcessFunction。
1.1 处理函数的功能和使用
转换算子一般只是针对某种具体操作来定义的，能够拿到信息比较有限。比如map算子，获取转换之后形式；而像窗口聚合这样的复杂操作，AggregateFunction中除数据外，还可以获取到当前的状态（以累加器Accumulator形式出现）。
另外富函数类，可以获取运行时上下文的方法getRuntimeContext()，还可以拿到状态、并行度、任务名称之类的运行时信息。但这些算子无法获得访问事件的时间戳、当前的水位线等信息。
处理函数提供了一个定时服务（TimeService），我们可以通过它访问流中的事件（event）、时间戳（timestamp）、水位线（watermark），甚至可以注册“定时事件”。而且处理函数继承了AbstractRichFunction抽象类，故拥有富函数类的所有特性，同样可以访问状态（state）和其他运行时信息。此外，处理函数还可以直接将数据输出到侧输出流（side output）中。故处理函数是最为灵活的处理方法，还可以实现各种自定义的业务逻辑，同时也是整个DataStream API的底层基础。
处理函数的使用与基本的转换操作类似，只需要直接基于 DataStream 调用.process()方法就可以了。方法需要传入一个 ProcessFunction 作为参数，用来定义处理逻辑：
stream.process(new MyProcessFunction)这里 ProcessFunction 不是接口，而是一个抽象类，继承了 AbstractRichFunction；
MyProcessFunction 是它的一个具体实现。所以所有的处理函数，都是富函数（RichFunction），富函数可以调用的东西这里同样都可以调用。
下面是一个具体的应用示例：
public class ProcessFunctionTest {public static void main(String[] args) throws Exception {//创建执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//设置生成水位线的时间间隔env.getConfig().setAutoWatermarkInterval(100);//乱序流的Watermark生成SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())// 插入水位线的逻辑 设置 watermark 延迟时间，2 秒.assignTimestampsAndWatermarks(// 针对乱序流插入水位线，延迟时间设置为 2sWatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))// 抽取时间戳的逻辑.withTimestampAssigner((SerializableTimestampAssigner<Event>) (element, recordTimestamp) -> element.getTimestamp()));stream.process(new ProcessFunction<Event, String>() {@Overridepublic void processElement(Event event, Context context, Collector<String> out) throws Exception {if("Mary".equals(event.getUser())){out.collect(event.getUser() + "clicks: " + event.getUrl());}else if("Bobo".equals(event.getUser())){out.collect(event.getUser());out.collect(event.getUser());}out.collect(event.toString());Long timestamp = context.timestamp();TimerService timerService = context.timerService();System.out.println("timestamp: " + timestamp);System.out.println("Watermark: " + timerService.currentWatermark());}}).print();env.execute();}
}这里在 ProcessFunction 中重写了.processElement()方法，自定义了一种处理逻辑：当数据的 user 为“Mary”时，将其输出一次；而如果为“Bob”时，将 user 输出两次。这里的
输 出 ， 是 通 过 调 用 out.collect() 来实现的。另外我们还可以调用
ctx.timerService().currentWatermark() 来 获 取 当 前 的 水 位 线 打 印 输 出 。 所 以 可 以 看 到 ，ProcessFunction 函数有点像 FlatMapFunction 的升级版。可以实现 Map、Filter、FlatMap 的所有功能。很明显，处理函数非常强大，能够做很多之前做不到的事情。
1.1.1 ProcessFunction 解析
在源码中可以看到，抽象类 ProcessFunction 继承了 AbstractRichFunction，有两个泛型类型参数：I 表示 Input ，也就是输入的数据类型；O 表示 Output ，也就是处理完成之后输出的数据类型。内部单独定义了两个方法：一个是必须要实现的抽象方法 .processElement()；另一个是非抽象方法 .onTimer()。
@PublicEvolving
public abstract class ProcessFunction<I, O> extends AbstractRichFunction {private static final long serialVersionUID = 1L;public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception;public void onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) throws Exception {}......}抽象方法.processElement()
用于“处理元素”，定义了处理的核心逻辑。这个方法对于流中的每个元素都会调用一次，
参数包括三个：输入数据值 value，上下文 ctx，以及“收集器”（Collector）out。方法没有返回值，处理之后的输出数据是通过收集器 out 来定义的。
- value：当前流中的输入元素，也就是正在处理的数据，类型与流中数据类
型一致。 - ctx：类型是 ProcessFunction 中定义的内部抽象类 Context，表示当前运行的
上下文，可以获取到当前的时间戳，并提供了用于查询时间和注册定时器的“定时服
务”(TimerService)，以及可以将数据发送到“侧输出流”（side output）的方法.output()。
Context 抽象类定义如下：
public abstract class Context {public abstract Long timestamp();public abstract TimerService timerService();public abstract <X> void output(OutputTag<X> outputTag, X value);
}- out：“收集器”（类型为 Collector），用于返回输出数据。使用方式与 flatMap
算子中的收集器完全一样，直接调用 out.collect()方法就可以向下游发出一个数据。
这个方法可以多次调用，也可以不调用。
通过几个参数的分析不难发现，ProcessFunction 可以轻松实现 flatMap 这样的基本转换功能（当然 map、filter 更不在话下）；而通过富函数提供的获取上下文方法.getRuntimeContext()，也可以自定义状态（state）进行处理，这也就能实现聚合操作的功能了。
非抽象方法.onTimer()
用于定义定时触发的操作，这是一个非常强大、也非常有趣的功能。
这个方法只有在注册好的定时器触发的时候才会调用，而定时器是通过“定时服务”TimerService 来注册的。
打个比方，注册定时器（timer）就是设了一个闹钟，到了设定时间就会响；而.onTimer()中定义的，就是闹钟响的时候要做的事。所以它本质上是一个基于时间的“回调”（callback）方法，通过时间的进展来触发；在事件时间语义下就是由水位线（watermark）来触发了。
与.processElement()类似，定时方法.onTimer()也有三个参数：时间戳（timestamp），上下文（ctx），以及收集器（out）。这里的 timestamp 是指设定好的触发时间，事件时间语义下当然就是水位线了。另外这里同样有上下文和收集器，所以也可以调用定时服务（TimerService），以及任意输出处理之后的数据。
既然有.onTimer()方法做定时触发，我们用 ProcessFunction 也可以自定义数据按照时间分组、定时触发计算输出结果；这其实就实现了窗口（window）的功能。所以说 ProcessFunction是真正意义上的终极奥义，用它可以实现一切功能。
我们也可以看到，处理函数都是基于事件触发的。水位线就如同插入流中的一条数据一样；只不过处理真正的数据事件调用的是.processElement()方法，而处理水位线事件调用的是.onTimer()。
这里需要注意的是，上面的.onTimer()方法只是定时器触发时的操作，而定时器（timer）
真正的设置需要用到上下文 ctx 中的定时服务。在 Flink 中，只有“按键分区流”KeyedStream才支持设置定时器的操作，所以之前的代码中我们并没有使用定时器。所以基于不同类型的流，可以使用不同的处理函数，它们之间还是有一些微小的区别的。
1.1.2 处理函数的分类
Flink 中的处理函数其实是一个大家族，ProcessFunction 只是其中一员。
我们知道，DataStream 在调用一些转换方法之后，有可能生成新的流类型；例如调
用.keyBy()之后得到 KeyedStream，进而再调用.window()之后得到 WindowedStream。对于不同类型的流，其实都可以直接调用.process()方法进行自定义处理，这时传入的参数就都叫作处理函数。当然，它们尽管本质相同，都是可以访问状态和时间信息的底层 API，可彼此之间也会有所差异。
Flink 提供了 8 个不同的处理函数：
- 1 ProcessFunction
最基本的处理函数，基于 DataStream 直接调用.process()时作为参数传入。 - 2 KeyedProcessFunction
对流按键分区后的处理函数，基于 KeyedStream 调用.process()时作为参数传入。要想使用定时器，比如基于 KeyedStream。 - 3 ProcessWindowFunction
开窗之后的处理函数，也是全窗口函数的代表。基于 WindowedStream 调用.process()时作为参数传入。 - 4 ProcessAllWindowFunction
同样是开窗之后的处理函数，基于 AllWindowedStream 调用.process()时作为参数传入。 - 5 CoProcessFunction
合并（connect）两条流之后的处理函数，基于 ConnectedStreams 调用.process()时作为参数传入。 - 6 ProcessJoinFunction
间隔连接（interval join）两条流之后的处理函数，基于 IntervalJoined 调用.process()时作为参数传入。 - 7 BroadcastProcessFunction
广播连接流处理函数，基于 BroadcastConnectedStream 调用.process()时作为参数传入。这里的“广播连接流”BroadcastConnectedStream，是一个未 keyBy 的普通 DataStream 与一个广播流（BroadcastStream）做连接（conncet）之后的产物。 - 8 KeyedBroadcastProcessFunction
按键分区的广播连接流处理函数，同样是基于 BroadcastConnectedStream 调用.process()时作为参数传入。与 BroadcastProcessFunction 不同的是，这时的广播连接流，是一个 KeyedStream与广播流（BroadcastStream）做连接之后的产物。
接下来，我们就对 KeyedProcessFunction 和 ProcessWindowFunction 的具体用法展开详细说明。
1.2 按键分区处理函数（KeyedProcessFunction）
在 Flink 程序中，为了实现数据的聚合统计，或者开窗计算之类的功能，我们一般都要先
用 keyBy 算子对数据流进行“按键分区”，得到一个 KeyedStream。也就是指定一个键（key），按照它的哈希值（hash code）将数据分成不同的“组”，然后分配到不同的并行子任务上执行计算；这相当于做了一个逻辑分流的操作，从而可以充分利用并行计算的优势实时处理海量数据。
另外只有在 KeyedStream 中才支持使用 TimerService 设置定时器的操作。所以一般情况下，都是先做了 keyBy 分区之后，再去定义处理操作；代码中更加常见的处理函数是 KeyedProcessFunction，最基本的 ProcessFunction 反而出镜率没那么高。
1.2.1 定时器（Timer）和定时服务（TimerService）
KeyedProcessFunction 的一个特色，就是可以灵活地使用定时器。
定时器（timers）是处理函数中进行时间相关操作的主要机制。在.onTimer()方法中可以实现定时处理的逻辑，而它能触发的前提，就是之前曾经注册过定时器、并且现在已经到了触发时间。注册定时器的功能，是通过上下文中提供的“定时服务”（TimerService）来实现的。
定时服务与当前运行的环境有关。前面已经介绍过，ProcessFunction 的上下文（Context）中提供了.timerService()方法，可以直接返回一个 TimerService 对象：
public abstract 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 内部会用一个优先队列将它们的时间戳（timestamp）保存起来，排队等待执行。可以认为，定时器其实是 KeyedStream上处理算子的一个状态，它以时间戳作为区分。所以 TimerService 会以键（key）和时间戳为标准，对定时器进行去重；也就是说对于每个 key 和时间戳，最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次。这样一来，我们在代码中就方便了很多，可以肆无忌惮地对一个 key 注册定时器，而不用担心重复定义——因为一个时间戳上的定时器只会触发一次。
基于 KeyedStream 注册定时器时，会传入一个定时器触发的时间戳，这个时间戳的定时器对于每个 key 都是有效的。这样，我们的代码并不需要做额外的处理，底层就可以直接对不同key 进行独立的处理操作了。
利用这个特性，有时我们可以故意降低时间戳的精度，来减少定时器的数量，从而提高处理性能。比如我们可以在设置定时器时只保留整秒数，那么定时器的触发频率就是最多 1 秒一次。
long coalescedTime = time / 1000 * 1000;
ctx.timerService().registerProcessingTimeTimer(coalescedTime);这里注意定时器的时间戳必须是毫秒数，所以我们得到整秒之后还要乘以 1000。定时器
默认的区分精度是毫秒。
另外 Flink 对.onTimer()和.processElement()方法是同步调用的（synchronous），所以也不会出现状态的并发修改。
Flink 的定时器同样具有容错性，它和状态一起都会被保存到一致性检查点（checkpoint）中。当发生故障时，Flink 会重启并读取检查点中的状态，恢复定时器。如果是处理时间的定时器，有可能会出现已经“过期”的情况，这时它们会在重启时被立刻触发。
1.2.2 KeyedProcessFunction 的使用
KeyedProcessFunction 是 ProcessFunction 的一个扩展。我们只要基于 keyBy 之后的 KeyedStream，直接调用.process()方法，这时需要传入的参数就是 KeyedProcessFunction 的实现类。
stream.keyBy( t -> t.f0 ).process(new MyKeyedProcessFunction())类似地，KeyedProcessFunction 也是继承自 AbstractRichFunction 的一个抽象类，源码中定义如下：
public abstract class KeyedProcessFunction<K, I, O> extends AbstractRichFunction {...public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception;public void onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) throws Exception {}public abstract class Context {...}...}可以看到与 ProcessFunction 的定义几乎完全一样，区别只是在于类型参数多了一个 K，这是当前按键分区的 key 的类型。同样地，我们必须实现一个.processElement()抽象方法，用来处理流中的每一个数据；另外还有一个非抽象方法.onTimer()，用来定义定时器触发时的回调操作。由于定时器只能在 KeyedStream 上使用，所以到了 KeyedProcessFunction 这里，我们才真正对时间有了精细的控制，定时方法.onTimer()才真正派上了用场。
下面是一个使用处理时间定时器的具体示例：
public class ProcessingTimeTimerTest {public static void main(String[] args) throws Exception {//创建执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//设置生成水位线的时间间隔env.getConfig().setAutoWatermarkInterval(100);//乱序流的Watermark生成SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())// 插入水位线的逻辑 设置 watermark 延迟时间，2 秒.assignTimestampsAndWatermarks(// 针对乱序流插入水位线，延迟时间设置为 2sWatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))// 抽取时间戳的逻辑.withTimestampAssigner((SerializableTimestampAssigner<Event>) (element, recordTimestamp) -> element.getTimestamp()));//处理时间定时器stream.keyBy(Event::getUser).process(new KeyedProcessFunction<String, Event, String>() {@Overridepublic void processElement(Event event, Context context, Collector<String> out) throws Exception {long processingTime = context.timerService().currentProcessingTime();out.collect(context.getCurrentKey() + "数据到达时间: " + new Timestamp(processingTime));//注册一个10秒后的定时器context.timerService().registerProcessingTimeTimer(processingTime + 10000);}@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {out.collect(ctx.getCurrentKey() + "定时器触发时间: " + new Timestamp(timestamp));}}).print();env.execute();}
}在上面的代码中，由于定时器只能在 KeyedStream 上使用，所以先要进行 keyBy；这里
的.keyBy(data -> true)是将所有数据的 key 都指定为了 true，其实就是所有数据拥有相同的 key，会分配到同一个分区。
之后我们自定义了一个 KeyedProcessFunction，其中.processElement()方法是每来一个数据都会调用一次，主要是定义了一个 10 秒之后的定时器；而.onTimer()方法则会在定时器触发时调用。所以我们会看到，程序运行后先在控制台输出“数据到达”的信息，等待 10 秒之后，又会输出“定时器触发”的信息，打印出的时间间隔正是 10 秒。
当然，上面的例子是处理时间的定时器，所以我们是真的需要等待 10 秒才会看到结果。
事件时间语义下，又会有什么不同呢？我们可以对上面的代码略作修改，做一个测试：
import com.atguigu.chapter05.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;public class EventTimeTimerTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<Event> stream = env.addSource(new CustomSource()).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<Event>() {@Overridepublic long extractTimestamp(Event element, long recordTimestamp) {return element.timestamp;}}));// 基于KeyedStream定义事件时间定时器stream.keyBy(data -> true).process(new KeyedProcessFunction<Boolean, Event, String>() {@Overridepublic void processElement(Event value, Context ctx, Collector<String> out) throws Exception {out.collect("数据到达，时间戳为：" + ctx.timestamp());out.collect("数据到达，水位线为：" + ctx.timerService().currentWatermark() + "\n -------分割线-------");// 注册一个10秒后的定时器ctx.timerService().registerEventTimeTimer(ctx.timestamp() + 10 * 1000L);}@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {out.collect("定时器触发，触发时间：" + timestamp);}}).print();env.execute();}// 自定义测试数据源public static class CustomSource implements SourceFunction<Event> {@Overridepublic void run(SourceContext<Event> ctx) throws Exception {// 直接发出测试数据ctx.collect(new Event("Mary", "./home", 1000L));// 为了更加明显，中间停顿5秒钟Thread.sleep(5000L);// 发出10秒后的数据ctx.collect(new Event("Mary", "./home", 11000L));Thread.sleep(5000L);// 发出10秒+1ms后的数据ctx.collect(new Event("Alice", "./cart", 11001L));Thread.sleep(5000L);}@Overridepublic void cancel() { }}
}由于是事件时间语义，所以我们必须从数据中提取出数据产生的时间戳。这里为了更清楚地看到程序行为，我们自定义了一个数据源，发出三条测试数据，时间戳分别为 1000、11000和 11001，并且发出数据后都会停顿 5 秒。
在代码中，我们依然将所有数据分到同一分区，然后在自定义的 KeyedProcessFunction 中使用定时器。同样地，每来一条数据，我们就将当前的数据时间戳和水位线信息输出，并注册一个 10 秒后（以当前数据时间戳为基准）的事件时间定时器。执行程序结果如下：
数据到达，时间戳为：1000
数据到达，水位线为：-9223372036854775808
-------分割线-------
数据到达，时间戳为：11000
数据到达，水位线为：999
-------分割线-------
数据到达，时间戳为：11001
数据到达，水位线为：10999
-------分割线-------
定时器触发，触发时间：11000
定时器触发，触发时间：21000
定时器触发，触发时间：21001每来一条数据，都会输出两行“数据到达”的信息，并以分割线隔开；两条数据到达的时
间间隔为 5 秒。当第三条数据到达后，随后立即输出一条定时器触发的信息；再过 5 秒之后，剩余两条定时器信息输出，程序运行结束。
我们可以发现，数据到来之后，当前的水位线与时间戳并不是一致的。当第一条数据到来，时间戳为 1000，可水位线的生成是周期性的（默认 200ms 一次），不会立即发生改变，所以依然是最小值 Long.MIN_VALUE；随后只要到了水位线生成的时间点（200ms 到了），就会依据当前的最大时间戳 1000 来生成水位线了。这里我们没有设置水位线延迟，默认需要减去 1 毫秒，所以水位线推进到了 999。而当时间戳为 11000 的第二条数据到来之后，水位线同样没有立即改变，仍然是 999，就好像总是“滞后”数据一样。
这样程序的行为就可以得到合理解释了。事件时间语义下，定时器触发的条件就是水位线推进到设定的时间。第一条数据到来后，设定的定时器时间为 1000 + 10 * 1000 = 11000；而当时间戳为 11000 的第二条数据到来，水位线还处在 999 的位置，当然不会立即触发定时器；而之后水位线会推进到 10999，同样是无法触发定时器的。必须等到第三条数据到来，将水位线真正推进到 11000，就可以触发第一个定时器了。第三条数据发出后再过 5 秒，没有更多的数据生成了，整个程序运行结束将要退出，此时 Flink 会自动将水位线推进到长整型的最大值（Long.MAX_VALUE）。于是所有尚未触发的定时器这时就统一触发了，我们就在控制台看到了后两个定时器的触发信息。
