欢迎访问我的GitHub
https://github.com/zq2599/blog\_demos
内容：所有原创文章分类汇总及配套源码，涉及Java、Docker、Kubernetes、DevOPS等；
Flink处理函数实战系列链接
- 深入了解ProcessFunction的状态操作(Flink-1.10)；
- ProcessFunction；
- KeyedProcessFunction类；
- ProcessAllWindowFunction(窗口处理)；
- CoProcessFunction(双流处理)；
本篇概览
本文是《Flink处理函数实战》系列的第三篇，上一篇《Flink处理函数实战之二：ProcessFunction类》学习了最简单的ProcessFunction类，今天要了解的KeyedProcessFunction，以及该类带来的一些特性；
关于KeyedProcessFunction
通过对比类图可以确定，KeyedProcessFunction和ProcessFunction并无直接关系：
KeyedProcessFunction用于处理KeyedStream的数据集合，相比ProcessFunction类，KeyedProcessFunction拥有更多特性，官方文档如下图红框，状态处理和定时器功能都是KeyedProcessFunction才有的：
介绍完毕，接下来通过实例来学习吧；
版本信息
- 开发环境操作系统：MacBook Pro 13寸， macOS Catalina 10.15.3
- 开发工具：IDEA ULTIMATE 2018.3
- JDK：1.8.0_211
- Maven：3.6.0
- Flink：1.9.2
源码下载
如果您不想写代码，整个系列的源码可在GitHub下载到，地址和链接信息如下表所示(https://github.com/zq2599/blog\_demos)：
名称
链接
备注
项目主页
https://github.com/zq2599/blog\_demos
该项目在GitHub上的主页
git仓库地址(https)
https://github.com/zq2599/blog\_demos.git
该项目源码的仓库地址，https协议
git仓库地址(ssh)
git@github.com:zq2599/blog_demos.git
该项目源码的仓库地址，ssh协议
这个git项目中有多个文件夹，本章的应用在flinkstudy文件夹下，如下图红框所示：
实战简介
本次实战的目标是学习KeyedProcessFunction，内容如下：
- 监听本机9999端口，获取字符串；
- 将每个字符串用空格分隔，转成Tuple2实例，f0是分隔后的单词，f1等于1；
- 上述Tuple2实例用f0字段分区，得到KeyedStream；
- KeyedSteam转入自定义KeyedProcessFunction处理；
- 自定义KeyedProcessFunction的作用，是记录每个单词最新一次出现的时间，然后建一个十秒的定时器，十秒后如果发现这个单词没有再次出现，就把这个单词和它出现的总次数发送到下游算子；
编码
继续使用《Flink处理函数实战之二：ProcessFunction类》一文中创建的工程flinkstudy；
创建bean类CountWithTimestamp，里面有三个字段，为了方便使用直接设为public：
package com.bolingcavalry.keyedprocessfunction;public class CountWithTimestamp { public String key; public long count; public long lastModified;}
创建FlatMapFunction的实现类Splitter，作用是将字符串分割后生成多个Tuple2实例，f0是分隔后的单词，f1等于1：
package com.bolingcavalry;import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.util.Collector;import org.apache.flink.util.StringUtils;public class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> { @Override public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception { if(StringUtils.isNullOrWhitespaceOnly(s)) { System.out.println("invalid line"); return; } for(String word : s.split(" ")) { collector.collect(new Tuple2<String, Integer>(word, 1)); } }}
最后是整个逻辑功能的主体：ProcessTime.java，这里面有自定义的KeyedProcessFunction子类，还有程序入口的main方法，代码在下面列出来之后，还会对关键部分做介绍：
package com.bolingcavalry.keyedprocessfunction;import com.bolingcavalry.Splitter;import org.apache.flink.api.common.state.ValueState;import org.apache.flink.api.common.state.ValueStateDescriptor;import org.apache.flink.api.java.tuple.Tuple;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.TimeCharacteristic;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;import org.apache.flink.streaming.api.functions.KeyedProcessFunction;import org.apache.flink.streaming.api.watermark.Watermark;import org.apache.flink.util.Collector;import java.text.SimpleDateFormat;import java.util.Date;/** * @author will * @email zq2599@gmail.com * @date 2020-05-17 13:43 * @description 体验KeyedProcessFunction类(时间类型是处理时间) /public class ProcessTime { /* * KeyedProcessFunction的子类，作用是将每个单词最新出现时间记录到backend，并创建定时器， * 定时器触发的时候，检查这个单词距离上次出现是否已经达到10秒，如果是，就发射给下游算子 */ static class CountWithTimeoutFunction extends KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Long>> { // 自定义状态 private ValueState
state; @Override public void open(Configuration parameters) throws Exception { // 初始化状态，name是myState state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class)); } @Override public void proces.........
