歡迎訪問我的GitHub
https://github.com/zq2599/blog_demos
內容：所有原創文章分類匯總及配套源碼，涉及Java、Docker、Kubernetes、DevOPS等；
本篇概覽
本文是《CoProcessFunction實戰三部曲》的第二篇，咱們要實戰的是雙流連接場景下，處理一號流中的數據時，還要結合該key在二號流中的情況；最簡單的例子：aaa在一號流中的value和二號流的value相加，再輸出到下遊，如下圖所示，一號流中的value存入state，在二號流中取出並相加，將結果輸出給下遊：
本篇的內容就是編碼實現上圖的功能；參考文章
理解狀態：《深入了解ProcessFunction的狀態操作(Flink-1.10)》
源碼下載
如果您不想寫代碼，整個系列的源碼可在GitHub下載到，地址和連結信息如下表所示(https://github.com/zq2599/blog_demos)：
這個git項目中有多個文件夾，本章的應用在flinkstudy文件夾下，如下圖紅框所示：
編碼
字符串轉Tuple2的Map函數，以及抽象類AbstractCoProcessFunctionExecutor都和上一篇《CoProcessFunction實戰三部曲之一：基本功能》一模一樣；新增AbstractCoProcessFunctionExecutor的子類AddTwoSourceValue.java，源碼如下，稍後會說明幾個關鍵點：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(); }}關鍵點之一：對於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為空，表示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)至此，雙流場景下的狀態互通實踐咱們已經完成了，接下來的文章，會加上定時器和旁路輸出，將雙流場景的數據處理考慮得更加全面；你不孤單，欣宸原創一路相伴
Java系列Spring系列Docker系列kubernetes系列資料庫+中間件系列DevOps系列歡迎關注公眾號：程式設計師欣宸
微信搜索「程式設計師欣宸」，我是欣宸，期待與您一同暢遊Java世界...https://github.com/zq2599/blog_demos