欢送拜访我的GitHub
https://github.com/zq2599/blog_demos
内容：所有原创文章分类汇总及配套源码，波及Java、Docker、Kubernetes、DevOPS等；
本篇概览
- 本文是《CoProcessFunction实战三部曲》的第二篇，咱们要实战的是双流连贯场景下，解决一号流中的数据时，还要联合该key在二号流中的状况；
- 最简略的例子：<font color="blue">aaa</font>在一号流中的value和二号流的value相加，再输入到上游，如下图所示，一号流中的value存入state，在二号流中取出并相加，将后果输入给上游：
本篇的内容就是编码实现上图的性能；
参考文章
了解状态：《深刻理解ProcessFunction的状态操作(Flink-1.10)》
源码下载
如果您不想写代码，整个系列的源码可在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>文件夹下，如下图红框所示：
编码
- 字符串转Tuple2的Map函数，以及抽象类<font color="blue">AbstractCoProcessFunctionExecutor</font>都和上一篇《CoProcessFunction实战三部曲之一：基本功能》截然不同；
- 新增AbstractCoProcessFunctionExecutor的子类<font color="blue">AddTwoSourceValue.java</font>，源码如下，稍后会阐明几个关键点：
package com.bolingcavalry.coprocessfunction;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.slf4j.Logger;import org.slf4j.LoggerFactory;/** * @author will * @email zq2599@gmail.com * @date 2020-11-11 09:48 * @description 性能介绍 */public class AddTwoSourceValue extends AbstractCoProcessFunctionExecutor { private static final Logger logger = LoggerFactory.getLogger(AddTwoSourceValue.class); @Override protected CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> getCoProcessFunctionInstance() { return new CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>>() { // 某个key在processElement1中存入的状态 private ValueState<Integer> state1; // 某个key在processElement2中存入的状态 private ValueState<Integer> state2; @Override public void open(Configuration parameters) throws Exception { // 初始化状态 state1 = getRuntimeContext().getState(new ValueStateDescriptor<>("myState1", Integer.class)); state2 = getRuntimeContext().getState(new ValueStateDescriptor<>("myState2", Integer.class)); } @Override public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception { logger.info("解决元素1：{}", value); String key = value.f0; Integer value2 = state2.value(); // value2为空，就示意processElement2还没有解决或这个key， // 这时候就把value1保存起来 if(null==value2) { logger.info("2号流还未收到过[{}]，把1号流收到的值[{}]保存起来", key, value.f1); state1.update(value.f1); } else { logger.info("2号流收到过[{}]，值是[{}]，当初把两个值相加后输入", key, value2); // 输入一个新的元素到上游节点 out.collect(new Tuple2<>(key, value.f1 + value2)); // 把2号流的状态清理掉 state2.clear(); } } @Override public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception { logger.info("解决元素2：{}", value); String key = value.f0; Integer value1 = state1.value(); // value1为空，就示意processElement1还没有解决或这个key， // 这时候就把value2保存起来 if(null==value1) { logger.info("1号流还未收到过[{}]，把2号流收到的值[{}]保存起来", key, value.f1); state2.update(value.f1); } else { logger.info("1号流收到过[{}]，值是[{}]，当初把两个值相加后输入", key, value1); // 输入一个新的元素到上游节点 out.collect(new Tuple2<>(key, value.f1 + value1)); // 把1号流的状态清理掉 state1.clear(); } } }; } public static void main(String[] args) throws Exception { new AddTwoSourceValue().execute(); }}- 关键点之一：对于<font color="blue">aaa</font>这个key，无奈确定会先呈现在一号源还是二号源，如果先呈现在一号源，就应该在processElement1中将value保留在state1中，这样等到aaa再次出现在二号源时，processElement2就能够从state1中取出一号源的value，相加后输入到上游；
- 关键点之二：如果输入到上游，就示意数据曾经处理完毕，此时要把保留的状态清理掉；
- 如果您想理解低阶函数中的状态存取的更多细节，请参考《深刻理解ProcessFunction的状态操作(Flink-1.10)》
验证
- 别离开启本机的<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">AddTwoSourceValue.main</font>办法即可（如果是windows电脑，我这没试过，不过做成jar在线部署也是能够的）；
- 在监听9998端口的控制台输出<font color="blue">aaa,111</font>，此时flink控制台输入如下，可见processElement1办法中，读取state2为空，示意aaa在二号流还未呈现过，此时的aaa是首次呈现，应该放入state中保留：
22:35:12,135 INFO AddTwoSourceValue - 解决元素1：(aaa,111)22:35:12,136 INFO AddTwoSourceValue - 2号流还未收到过[aaa]，把1号流收到的值[111]保存起来- 在监听9999端口的控制台输出<font color="blue">bbb,123</font>，flink日志如下所示，示意bbb也是首次呈现，把值保留在state中：
22:35:34,473 INFO AddTwoSourceValue - 解决元素2：(bbb,123)22:35:34,473 INFO AddTwoSourceValue - 1号流还未收到过[bbb]，把2号流收到的值[123]保存起来- 在监听9999端口的控制台输出<font color="blue">aaa,222</font>，flink日志如下，很显著，之前保留在state中的值被取出来了，因而processElement2办法中，aaa在两个数据源的值111和222会被相加后输入到上游，上游是print，间接打印进去了：
22:35:38,072 INFO AddTwoSourceValue - 解决元素2：(aaa,222)22:35:38,072 INFO AddTwoSourceValue - 1号流收到过[aaa]，值是[111]，当初把两个值相加后输入(aaa,333)- 至此，双流场景下的状态互通实际咱们曾经实现了，接下来的文章，会加上定时器和旁路输入，将双流场景的数据处理思考得更加全面；
你不孤独，欣宸原创一路相伴
- Java系列
- Spring系列
- Docker系列
- kubernetes系列
- 数据库+中间件系列
- DevOps系列
欢送关注公众号：程序员欣宸
微信搜寻「程序员欣宸」，我是欣宸，期待与您一起畅游Java世界...
https://github.com/zq2599/blog_demos