CoProcessFunction实战三部曲之三：定时器和侧输出

朴灿烈づ我的快乐病毒、 2022-12-20 12:45 124阅读 0赞

### 欢迎访问我的GitHub ###

> 这里分类和汇总了欣宸的全部原创(含配套源码)：[https://github.com/zq2599/blog\_demos][https_github.com_zq2599_blog_demos]

### 系列文章链接 ###

1.  [基本功能][Link 1]
2.  [状态处理][Link 2]
3.  [定时器和侧输出][Link 3]

### 本篇概览 ###

*  本文是《CoProcessFunction实战三部曲》的终篇，主要内容是在CoProcessFunction中使用定时器和侧输出，对上一篇的功能进行增强；
 *  回顾上一篇的功能：一号流收到aaa后保存在状态中，直到二号流收到aaa，把两个aaa的值相加后输出到下游；
 *  上述功能有个问题：二号流如果一直收不到aaa，下游就一直没有aaa的输出，相当于进入一号流的aaa已经石沉大海了；
 *  今天的实战就是修复上述问题：aaa在一个流中出现后，10秒之内如果出现在另一个流中，就像以前那样值相加，输出到下游，如果10秒内没有出现在另一个流，就流向侧输出，再将所有状态清理干净；

### 参考文章 ###

1.  理解状态：[《深入了解ProcessFunction的状态操作(Flink-1.10)》][ProcessFunction_Flink-1.10]
2.  理解定时器：[《理解ProcessFunction的Timer逻辑》][ProcessFunction_Timer]

### 梳理流程 ###

*  为了编码的逻辑正确，咱们把正常和异常的流程先梳理清楚；
 *  下图是正常流程：aaa在一号流出现后，10秒内又在二号流出现了，于是相加并流向下游：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_left]
 *  再来看异常的流程，如下图，一号流在16:14:01收到aaa，但二号流一直没有收到aaa，等到10秒后，也就是16:14:11，定时器被触发，从状态1得知10秒前一号流收到过aaa，于是将数据流向一号侧输出：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_left 1]
 *  接下来编码实现上面的功能；

### 源码下载 ###

如果您不想写代码，整个系列的源码可在GitHub下载到，地址和链接信息如下表所示(https://github.com/zq2599/blog\_demos)：

<table> 
 <thead> 
  <tr> 
   <th align="left">名称</th> 
   <th align="left">链接</th> 
   <th align="left">备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">项目主页</td> 
   <td align="left">https://github.com/zq2599/blog_demos</td> 
   <td align="left">该项目在GitHub上的主页</td> 
  </tr> 
  <tr> 
   <td align="left">git仓库地址(https)</td> 
   <td align="left">https://github.com/zq2599/blog_demos.git</td> 
   <td align="left">该项目源码的仓库地址，https协议</td> 
  </tr> 
  <tr> 
   <td align="left">git仓库地址(ssh)</td> 
   <td align="left">git@github.com:zq2599/blog_demos.git</td> 
   <td align="left">该项目源码的仓库地址，ssh协议</td> 
  </tr> 
 </tbody> 
</table>

这个git项目中有多个文件夹，本章的应用在flinkstudy文件夹下，如下图红框所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70]

### CoProcessFunction的子类 ###

1.  前面的两篇实战中，CoProcessFunction的子类都写成了匿名类，如下图红框：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center]
2.  本文中，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();
        }
    }

1.  关键点之一：新增状态timerState，用于保存定时器的key；
2.  关键点之二：CoProcessFunction的onTimer中拿不到当前key(KeyedProcessFunction可以，其OnTimerContext类提供了API)，因此新增状态currentKeyState，这样在onTimer中就知道当前key了；
3.  关键点之三：processElement1中，处理aaa时， 如果2号流还没收到过aaa，就存入状态，并启动10秒定时器；
4.  关键点之四：processElement2处理aaa时，发现1号流收到过aaa，就相加再输出到下游，并且删除processElement1中创建的定时器，aaa相关的所有状态也全部清理掉；
5.  关键点之五：如果10秒内aaa在两个流中都出现过，那么一定会流入下游并且定时器会被删除，因此，一旦onTimer被执行，意味着aaa只在一个流中出现过，而且已经过去10秒了，此时在onTimer中可以执行流向侧输出的操作；
6.  以上就是双流处理的逻辑和代码，接下来编写AbstractCoProcessFunctionExecutor的子类；

### 业务执行类AddTwoSourceValueWithTimeout ###

1.  负责执行整个功能的，是抽象类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 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();
        }
    }

1.  关键点之一：增减成员变量source1SideOutput和source2SideOutput，用于侧输出；
2.  关键点之二：重写父类的buildStreamFromSocket方法，加了个时间戳分配器，这样每个元素都带有时间戳；
3.  关键点之三：重写父类的doSideOutput方法，这里面会把侧输出的数据打印出来；
4.  以上就是所有代码了，接下来开始验证；

### 验证（不超时的操作） ###

1.  分别开启本机的9998和9999端口，我这里是MacBook，执行nc -l 9998和nc -l 9999
2.  启动Flink应用，如果您和我一样是Mac电脑，直接运行AddTwoSourceValueWithTimeout.main方法即可（如果是windows电脑，我这没试过，不过做成jar在线部署也是可以的）；
3.  在监听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号流接收数据

1.  尽快在监听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]

### 验证（超时的操作） ###

1.  前面试过了正常流程，再来试试超时流程是否符合预期；
2.  在监听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世界…

[https_github.com_zq2599_blog_demos]: https://github.com/zq2599/blog_demos
[Link 1]: https://xinchen.blog.csdn.net/article/details/109624375
[Link 2]: https://xinchen.blog.csdn.net/article/details/109629119
[Link 3]: https://xinchen.blog.csdn.net/article/details/109645214
[ProcessFunction_Flink-1.10]: https://xinchen.blog.csdn.net/article/details/106040312
[ProcessFunction_Timer]: https://xinchen.blog.csdn.net/article/details/109564999
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_left]: /images/20221120/236c86e3404b41d888bddae058f9ed73.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_left 1]: /images/20221120/54ad869a0bf14f3297d00dad98a16fe6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70]: /images/20221120/0160af69de7a4f43afbc1263f3fa5bd6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/8d65e8f66c914beeb044d478019452ba.png