CoProcessFunction實戰三部曲之三：定時器和側輸出-有解無憂

歡迎訪問我的GitHub

https://github.com/zq2599/blog_demos

內容：所有原創文章分類匯總及配套原始碼，涉及Java、Docker、Kubernetes、DevOPS等；

系列文章鏈接

基本功能
狀態處理
定時器和側輸出

本篇概覽

本文是《CoProcessFunction實戰三部曲》的終篇，主要內容是在CoProcessFunction中使用定時器和側輸出，對上一篇的功能進行增強；
回顧上一篇的功能：一號流收到aaa后保存在狀態中，直到二號流收到aaa，把兩個aaa的值相加后輸出到下游；
上述功能有個問題：二號流如果一直收不到aaa，下游就一直沒有aaa的輸出，相當于進入一號流的aaa已經石沉大海了；
今天的實戰就是修復上述問題：aaa在一個流中出現后，10秒之內如果出現在另一個流中，就像以前那樣值相加，輸出到下游，如果10秒內沒有出現在另一個流，就流向側輸出，再將所有狀態清理干凈；

參考文章

理解狀態：《深入了解ProcessFunction的狀態操作(Flink-1.10)》
理解定時器：《理解ProcessFunction的Timer邏輯》

梳理流程

為了編碼的邏輯正確，咱們把正常和例外的流程先梳理清楚；
下圖是正常流程：aaa在一號流出現后，10秒內又在二號流出現了，于是相加并流向下游：
再來看例外的流程，如下圖，一號流在16:14:01收到aaa，但二號流一直沒有收到aaa，等到10秒后，也就是16:14:11，定時器被觸發，從狀態1得知10秒前一號流收到過aaa，于是將資料流向一號側輸出：
接下來編碼實作上面的功能；

原始碼下載

如果您不想寫代碼，整個系列的原始碼可在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)	[email protected]:zq2599/blog_demos.git	該專案原始碼的倉庫地址，ssh協議

這個git專案中有多個檔案夾，本章的應用在flinkstudy檔案夾下，如下圖紅框所示：
在這里插入圖片描述

CoProcessFunction的子類

前面的兩篇實戰中，CoProcessFunction的子類都寫成了匿名類，如下圖紅框：
本文中，CoProcessFunction子類會用到外部類的成員變數，因此不能再用匿名類了，新增CoProcessFunction的子類ExecuteWithTimeoutCoProcessFunction.java，稍后會說明幾個關鍵點：

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();
    }
}

關鍵點之一：新增狀態timerState，用于保存定時器的key；
關鍵點之二：CoProcessFunction的onTimer中拿不到當前key(KeyedProcessFunction可以，其OnTimerContext類提供了API)，因此新增狀態currentKeyState，這樣在onTimer中就知道當前key了；
關鍵點之三：processElement1中，處理aaa時，如果2號流還沒收到過aaa，就存入狀態，并啟動10秒定時器；
關鍵點之四：processElement2處理aaa時，發現1號流收到過aaa，就相加再輸出到下游，并且洗掉processElement1中創建的定時器，aaa相關的所有狀態也全部清理掉；
關鍵點之五：如果10秒內aaa在兩個流中都出現過，那么一定會流入下游并且定時器會被洗掉，因此，一旦onTimer被執行，意味著aaa只在一個流中出現過，而且已經過去10秒了，此時在onTimer中可以執行流向側輸出的操作；
以上就是雙流處理的邏輯和代碼，接下來撰寫AbstractCoProcessFunctionExecutor的子類；

業務執行類AddTwoSourceValueWithTimeout

負責執行整個功能的，是抽象類AbstractCoProcessFunctionExecutor的子類，如下，稍后會說明幾個關鍵點：

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 [email protected]
 * @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();
    }
}

關鍵點之一：增減成員變數source1SideOutput和source2SideOutput，用于側輸出；
關鍵點之二：重寫父類的buildStreamFromSocket方法，加了個時間戳分配器，這樣每個元素都帶有時間戳；
關鍵點之三：重寫父類的doSideOutput方法，這里面會把側輸出的資料列印出來；
以上就是所有代碼了，接下來開始驗證；

驗證（不超時的操作）

分別開啟本機的9998和9999埠，我這里是MacBook，執行nc -l 9998和nc -l 9999
啟動Flink應用，如果您和我一樣是Mac電腦，直接運行AddTwoSourceValueWithTimeout.main方法即可（如果是windows電腦，我這沒試過，不過做成jar在線部署也是可以的）；
在監聽9998埠的控制臺輸入aaa,1，此時flink控制臺輸出如下，可見processElement1方法中，讀取state2為空，表示aaa在2號流還未出現過，此時的aaa是首次出現，應該放入state中保存，并且創建了定時器：

18:18:10,472 INFO  AddTwoSourceValueWithTimeout  - 添加時間戳，值：(aaa,1)，時間戳：2020-11-12 06:18:10
18: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埠的控制臺輸入aaa,2，flink日志如下所示，可見相加后輸出到下游，并且定時器也洗掉了：

18:18:15,813 INFO  AddTwoSourceValueWithTimeout  - 添加時間戳，值：(aaa,2)，時間戳：2020-11-12 06:18:15
18: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埠的控制臺輸入aaa,1，然后等待十秒，flink控制臺輸出如下，可見定時器被觸發，并且aaa流向了1號流的側輸出：

18:23:37,393 INFO  AddTwoSourceValueWithTimeout - 添加時間戳，值：(aaa,1)，時間戳：2020-11-12 06:23:37
18: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

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/230972.html

標籤：Java

上一篇：Python No.11_深淺Copy

下一篇：LeetCode——905. 按奇偶排序陣列