import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class AddTwoSourcevalue extends AbstractCoProcessFunctionExecutor {
private static final Logger logger = LoggerFactory.getLogger(AddTwoSourcevalue.class);
@Override
protected CoProcessFunction
return new CoProcessFunction
// 某个key在processElement1中存入的状态
private ValueState state1;
// 某个key在processElement2中存入的状态
private ValueState 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
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
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();
}
}
-
关键点之一：对于aaa这个key，无法确定会先出现在一号源还是二号源，如果先出现在一号源，就应该在processElement1中将value保存在state1中，这样等到aaa再次出现在二号源时，processElement2就可以从state1中取出一号源的value，相加后输出到下游；
-
关键点之二：如果输出到下游，就表示数据已经处理完毕，此时要把保存的状态清理掉；
-
如果您想了解低阶函数中的状态存取的更多细节，请参考《深入了解ProcessFunction的状态操作(Flink-1.10)》
-
分别开启本机的9998和9999端口，我这里是MacBook，执行nc -l 9998和nc -l 9999
-
启动Flink应用，如果您和我一样是Mac电脑，直接运行AddTwoSourcevalue.main方法即可（如果是windows电脑，我这没试过，不过做成jar在线部署也是可以的）；
-
在监听9998端口的控制台输入aaa,111，此时flink控制台输出如下，可见processElement1方法中，读取state2为空，表示
【一线大厂Java面试题解析+后端开发学习笔记+最新架构讲解视频+实战项目源码讲义】 浏览器打开：qq.cn.hn/FTf 免费领取
aaa在二号流还未出现过，此时的aaa是首次出现，应该放入state中保存：
22:35:12,135 INFO AddTwoSourcevalue - 处理元素1：(aaa,111)
22:35:12,136 INFO AddTwoSourcevalue - 2号流还未收到过[aaa]，把1号流收到的值[111]保存起来
- 在监听9999端口的控制台输入bbb,123，flink日志如下所示，表示bbb也是首次出现，把值保存在state中：
22:35:34,473 INFO AddTwoSourcevalue - 处理元素2：(bbb,123)
22:35:34,473 INFO AddTwoSourcevalue - 1号流还未收到过[bbb]，把2号流收到的值[123]保存起来
- 在监听9999端口的控制台输入aaa,222，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)
