欢送拜访我的GitHub
这里分类和汇总了欣宸的全副原创(含配套源码)：https://github.com/zq2599/blog_demos
本篇概览
- 本文是《CoProcessFunction实战三部曲》的终篇，次要内容是在CoProcessFunction中应用定时器和侧输入，对上一篇的性能进行加强；
- 回顾上一篇的性能：一号流收到<font color="blue">aaa</font>后保留在状态中，直到二号流收到aaa，把两个aaa的值相加后输入到上游；
- 上述性能有个问题：二号流如果始终收不到<font color="blue">aaa</font>，上游就始终没有aaa的输入，相当于进入一号流的aaa曾经杳无音信了；
明天的实战就是修复上述问题：aaa在一个流中呈现后，10秒之内如果呈现在另一个流中，就像以前那样值相加，输入到上游，如果10秒内没有呈现在另一个流，就流向侧输入，再将所有状态清理洁净；
参考文章
- 了解状态：《深刻理解ProcessFunction的状态操作(Flink-1.10)》
了解定时器：《了解ProcessFunction的Timer逻辑》
梳理流程
- 为了编码的逻辑正确，咱们把失常和异样的流程先梳理分明；
- 下图是失常流程：aaa在一号流呈现后，10秒内又在二号流呈现了，于是相加并流向上游：
- 再来看异样的流程，如下图，一号流在<font color="blue">16:14:01</font>收到aaa，但二号流始终没有收到aaa，等到10秒后，也就是<font color="blue">16:14:11</font>，定时器被触发，从状态1得悉10秒前一号流收到过aaa，于是将数据流向一号侧输入：
接下来编码实现下面的性能；
源码下载
如果您不想写代码，整个系列的源码可在GitHub下载到，地址和链接信息如下表所示(https://github.com/zq2599/blo...
| 名称 | 链接 | 备注 |
|---|---|---|
| 我的项目主页 | https://github.com/zq2599/blo... | 该我的项目在GitHub上的主页 |
| git仓库地址(https) | https://github.com/zq2599/blo... | 该我的项目源码的仓库地址，https协定 |
| git仓库地址(ssh) | git@github.com:zq2599/blog_demos.git | 该我的项目源码的仓库地址，ssh协定 |
这个git我的项目中有多个文件夹，本章的利用在<font color="blue">flinkstudy</font>文件夹下，如下图红框所示：
CoProcessFunction的子类
- 后面的两篇实战中，<font color="blue">CoProcessFunction</font>的子类都写成了匿名类，如下图红框：
- 本文中，CoProcessFunction子类会用到外部类的成员变量，因而不能再用匿名类了，新增CoProcessFunction的子类<font color="blue">ExecuteWithTimeoutCoProcessFunction.java</font>，稍后会阐明几个关键点：
package com.bolingcavalry.coprocessfunction;import com.bolingcavalry.Utils;import org.apache.flink.api.common.state.ValueState;import org.apache.flink.api.common.state.ValueStateDescriptor;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.functions.co.CoProcessFunction;import org.apache.flink.util.Collector;import org.apache.flink.util.OutputTag;import org.slf4j.Logger;import org.slf4j.LoggerFactory;/** * 实现双流业务逻辑的性能类 */public class ExecuteWithTimeoutCoProcessFunction extends CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> { private static final Logger logger = LoggerFactory.getLogger(ExecuteWithTimeoutCoProcessFunction.class); /** * 等待时间 */ private static final long WAIT_TIME = 10000L; public ExecuteWithTimeoutCoProcessFunction(OutputTag<String> source1SideOutput, OutputTag<String> source2SideOutput) { super(); this.source1SideOutput = source1SideOutput; this.source2SideOutput = source2SideOutput; } private OutputTag<String> source1SideOutput; private OutputTag<String> source2SideOutput; // 某个key在processElement1中存入的状态 private ValueState<Integer> state1; // 某个key在processElement2中存入的状态 private ValueState<Integer> state2; // 如果创立了定时器，就在状态中保留定时器的key private ValueState<Long> timerState; // onTimer中拿不到以后key，只能提前保留在状态中（KeyedProcessFunction的OnTimerContext有API能够取到，然而CoProcessFunction的OnTimerContext却没有） private ValueState<String> currentKeyState; @Override public void open(Configuration parameters) throws Exception { // 初始化状态 state1 = getRuntimeContext().getState(new ValueStateDescriptor<>("myState1", Integer.class)); state2 = getRuntimeContext().getState(new ValueStateDescriptor<>("myState2", Integer.class)); timerState = getRuntimeContext().getState(new ValueStateDescriptor<>("timerState", Long.class)); currentKeyState = getRuntimeContext().getState(new ValueStateDescriptor<>("currentKeyState", String.class)); } /** * 所有状态都清理掉 */ private void clearAllState() { state1.clear(); state2.clear(); currentKeyState.clear(); timerState.clear(); } @Override public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception { logger.info("processElement1：解决元素1：{}", value); String key = value.f0; Integer value2 = state2.value(); // value2为空，就示意processElement2还没有解决或这个key， // 这时候就把value1保存起来 if(null==value2) { logger.info("processElement1：2号流还未收到过[{}]，把1号流收到的值[{}]保存起来", key, value.f1); state1.update(value.f1); currentKeyState.update(key); // 开始10秒的定时器，10秒后会进入 long timerKey = ctx.timestamp() + WAIT_TIME; ctx.timerService().registerProcessingTimeTimer(timerKey); // 保留定时器的key timerState.update(timerKey); logger.info("processElement1：创立定时器[{}]，期待2号流接收数据", Utils.time(timerKey)); } else { logger.info("processElement1：2号流收到过[{}]，值是[{}]，当初把两个值相加后输入", key, value2); // 输入一个新的元素到上游节点 out.collect(new Tuple2<>(key, value.f1 + value2)); // 删除定时器（这个定时器应该是processElement2创立的） long timerKey = timerState.value(); logger.info("processElement1：[{}]的新元素已输入到上游，删除定时器[{}]", key, Utils.time(timerKey)); ctx.timerService().deleteProcessingTimeTimer(timerKey); clearAllState(); } } @Override public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception { logger.info("processElement2：解决元素2：{}", value); String key = value.f0; Integer value1 = state1.value(); // value1为空，就示意processElement1还没有解决或这个key， // 这时候就把value2保存起来 if(null==value1) { logger.info("processElement2：1号流还未收到过[{}]，把2号流收到的值[{}]保存起来", key, value.f1); state2.update(value.f1); currentKeyState.update(key); // 开始10秒的定时器，10秒后会进入 long timerKey = ctx.timestamp() + WAIT_TIME; ctx.timerService().registerProcessingTimeTimer(timerKey); // 保留定时器的key timerState.update(timerKey); logger.info("processElement2：创立定时器[{}]，期待1号流接收数据", Utils.time(timerKey)); } else { logger.info("processElement2：1号流收到过[{}]，值是[{}]，当初把两个值相加后输入", key, value1); // 输入一个新的元素到上游节点 out.collect(new Tuple2<>(key, value.f1 + value1)); // 删除定时器（这个定时器应该是processElement1创立的） long timerKey = timerState.value(); logger.info("processElement2：[{}]的新元素已输入到上游，删除定时器[{}]", key, Utils.time(timerKey)); ctx.timerService().deleteProcessingTimeTimer(timerKey); clearAllState(); } } @Override public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception { super.onTimer(timestamp, ctx, out); String key = currentKeyState.value(); // 定时器被触发，意味着此key只在一个中呈现过 logger.info("[{}]的定时器[{}]被触发了", key, Utils.time(timestamp)); Integer value1 = state1.value(); Integer value2 = state2.value(); if(null!=value1) { logger.info("只有1号流收到过[{}]，值为[{}]", key, value1); // 侧输入 ctx.output(source1SideOutput, "source1 side, key [" + key+ "], value [" + value1 + "]"); } if(null!=value2) { logger.info("只有2号流收到过[{}]，值为[{}]", key, value2); // 侧输入 ctx.output(source2SideOutput, "source2 side, key [" + key+ "], value [" + value2 + "]"); } clearAllState(); }}- 关键点之一：新增状态<font color="blue">timerState</font>，用于保留定时器的key；
- 关键点之二：CoProcessFunction的<font color="blue">onTimer</font>中拿不到以后key(KeyedProcessFunction能够，其OnTimerContext类提供了API)，因而新增状态<font color="blue">currentKeyState</font>，这样在onTimer中就晓得以后key了；
- 关键点之三：processElement1中，解决aaa时， 如果2号流还没收到过aaa，就存入状态，并启动10秒定时器；
- 关键点之四：processElement2解决aaa时，发现1号流收到过aaa，就相加再输入到上游，并且删除processElement1中创立的定时器，aaa相干的所有状态也全副清理掉；
- 关键点之五：如果10秒内aaa在两个流中都呈现过，那么肯定会流入上游并且定时器会被删除，因而，一旦<font color="blue">onTimer</font>被执行，意味着aaa只在一个流中呈现过，而且曾经过来10秒了，此时在<font color="blue">onTimer</font>中能够执行流向侧输入的操作；
- 以上就是双流解决的逻辑和代码，接下来编写<font color="blue">AbstractCoProcessFunctionExecutor</font>的子类；
业务执行类AddTwoSourceValueWithTimeout
- 负责执行整个性能的，是抽象类<font color="blue">AbstractCoProcessFunctionExecutor</font>的子类，如下，稍后会阐明几个关键点：
package com.bolingcavalry.coprocessfunction;import com.bolingcavalry.Utils;import org.apache.flink.api.java.tuple.Tuple;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.datastream.KeyedStream;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;import org.apache.flink.streaming.api.functions.co.CoProcessFunction;import org.apache.flink.streaming.api.watermark.Watermark;import org.apache.flink.util.OutputTag;import org.slf4j.Logger;import org.slf4j.LoggerFactory;/** * @author will * @email zq2599@gmail.com * @date 2020-11-11 09:48 * @description 将两个流中相通key的value相加，当key在一个流中呈现后， * 会在无限工夫内期待它在另一个流中呈现，如果超过等待时间任未呈现就在旁路输入 */public class AddTwoSourceValueWithTimeout extends AbstractCoProcessFunctionExecutor { private static final Logger logger = LoggerFactory.getLogger(AddTwoSourceValueWithTimeout.class); // 假如aaa流入1号源后，在2号源超过10秒没有收到aaa，那么1号源的aaa就会流入source1SideOutput final OutputTag<String> source1SideOutput = new OutputTag<String>("source1-sideoutput"){}; // 假如aaa流入2号源后，如果1号源超过10秒没有收到aaa，那么2号源的aaa就会流入source2SideOutput final OutputTag<String> source2SideOutput = new OutputTag<String>("source2-sideoutput"){}; /** * 重写父类的办法，放弃父类逻辑不变，仅减少了工夫戳分配器，向元素中退出工夫戳 * @param port * @return */ @Override protected KeyedStream<Tuple2<String, Integer>, Tuple> buildStreamFromSocket(StreamExecutionEnvironment env, int port) { return env // 监听端口 .socketTextStream("localhost", port) // 失去的字符串"aaa,3"转成Tuple2实例，f0="aaa"，f1=3 .map(new WordCountMap()) // 设置工夫戳分配器，用以后工夫作为工夫戳 .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() { @Override public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) { long timestamp = System.currentTimeMillis(); logger.info("增加工夫戳，值：{}，工夫戳：{}", element, Utils.time(timestamp)); // 应用以后零碎工夫作为工夫戳 return timestamp; } @Override public Watermark getCurrentWatermark() { // 本例不须要watermark，返回null return null; } }) // 将单词作为key分区 .keyBy(0); } @Override protected CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> getCoProcessFunctionInstance() { return new ExecuteWithTimeoutCoProcessFunction(source1SideOutput, source2SideOutput); } @Override protected void doSideOutput(SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream) { // 两个侧输入都间接打印 mainDataStream.getSideOutput(source1SideOutput).print(); mainDataStream.getSideOutput(source2SideOutput).print(); } public static void main(String[] args) throws Exception { new AddTwoSourceValueWithTimeout().execute(); }}- 关键点之一：增减成员变量<font color="blue">source1SideOutput</font>和<font color="blue">source2SideOutput</font>，用于侧输入；
- 关键点之二：重写父类的buildStreamFromSocket办法，加了个工夫戳分配器，这样每个元素都带有工夫戳；
- 关键点之三：重写父类的doSideOutput办法，这外面会把侧输入的数据打印进去；
- 以上就是所有代码了，接下来开始验证；
验证（不超时的操作）
- 别离开启本机的<font color="blue">9998</font>和<font color="blue">9999</font>端口，我这里是MacBook，执行<font color="blue">nc -l 9998</font>和<font color="blue">nc -l 9999</font>
- 启动Flink利用，如果您和我一样是Mac电脑，间接运行<font color="blue">AddTwoSourceValueWithTimeout.main</font>办法即可（如果是windows电脑，我这没试过，不过做成jar在线部署也是能够的）；
- 在监听9998端口的控制台输出<font color="blue">aaa,1</font>，此时flink控制台输入如下，可见processElement1办法中，读取state2为空，示意aaa在2号流还未呈现过，此时的aaa是首次呈现，应该放入state中保留，并且创立了定时器：
18:18:10,472 INFO AddTwoSourceValueWithTimeout - 增加工夫戳，值：(aaa,1)，工夫戳：2020-11-12 06:18:1018:18:10,550 INFO ExecuteWithTimeoutCoProcessFunction - processElement1：解决元素1：(aaa,1)18:18:10,550 INFO ExecuteWithTimeoutCoProcessFunction - processElement1：2号流还未收到过[aaa]，把1号流收到的值[1]保存起来18:18:10,553 INFO ExecuteWithTimeoutCoProcessFunction - processElement1：创立定时器[2020-11-12 06:18:20]，期待2号流接收数据- 尽快在监听9999端口的控制台输出<font color="blue">aaa,2</font>，flink日志如下所示，可见相加后输入到上游，并且定时器也删除了：
18:18:15,813 INFO AddTwoSourceValueWithTimeout - 增加工夫戳，值：(aaa,2)，工夫戳：2020-11-12 06:18:1518:18:15,887 INFO ExecuteWithTimeoutCoProcessFunction - processElement2：解决元素2：(aaa,2)18:18:15,887 INFO ExecuteWithTimeoutCoProcessFunction - processElement2：1号流收到过[aaa]，值是[1]，当初把两个值相加后输入(aaa,3)18:18:15,888 INFO ExecuteWithTimeoutCoProcessFunction - processElement2：[aaa]的新元素已输入到上游，删除定时器[2020-11-12 06:18:20]验证（超时的操作）
- 前面试过了失常流程，再来试试超时流程是否合乎预期；
- 在监听9998端口的控制台输出<font color="blue">aaa,1</font>，而后期待十秒，flink控制台输入如下，可见定时器被触发，并且aaa流向了1号流的侧输入：
18:23:37,393 INFO AddTwoSourceValueWithTimeout - 增加工夫戳，值：(aaa,1)，工夫戳：2020-11-12 06:23:3718:23:37,417 INFO ExecuteWithTimeoutCoProcessFunction - processElement1：解决元素1：(aaa,1)18:23:37,417 INFO ExecuteWithTimeoutCoProcessFunction - processElement1：2号流还未收到过[aaa]，把1号流收到的值[1]保存起来18:23:37,417 INFO ExecuteWithTimeoutCoProcessFunction - processElement1：创立定时器[2020-11-12 06:23:47]，期待2号流接收数据18:23:47,398 INFO ExecuteWithTimeoutCoProcessFunction - [aaa]的定时器[2020-11-12 06:23:47]被触发了18:23:47,399 INFO ExecuteWithTimeoutCoProcessFunction - 只有1号流收到过[aaa]，值为[1]source1 side, key [aaa], value [1]- 至此，CoProcessFunction实战三部曲曾经全副实现了，心愿这三次实战可能给您一些参考，帮您更快把握和了解CoProcessFunction；
你不孤独，欣宸原创一路相伴
- Java系列
- Spring系列
- Docker系列
- kubernetes系列
- 数据库+中间件系列
- DevOps系列
欢送关注公众号：程序员欣宸
微信搜寻「程序员欣宸」，我是欣宸，期待与您一起畅游Java世界...
https://github.com/zq2599/blog_demos