Flink-1.9流计算开发：十二、apply函数

public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 添加自定义数据源,每秒发出一笔订单信息{商品名称,商品数量}
        DataStreamSource<Tuple2<String, Integer>> orderSource = env
                .addSource(new SourceFunction<Tuple2<String, Integer>>() {
                    private volatile boolean isRunning = true;
                    private final Random random = new Random();

                    @Override
                    public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
                        while (isRunning) {
                            TimeUnit.SECONDS.sleep(1);
                            ctx.collect(Tuple2.of(TYPE[random.nextInt(TYPE.length)], 1));
                        }
                    }
                    @Override
                    public void cancel() {
                        isRunning = false;
                    }

                }, "order-info");
        orderSource.
                //  未使用keyby分区流的时间窗口
                timeWindowAll(Time.seconds(10))
                // apply 函数是对窗口内的数据做处理的核心方法。这是对10s窗口中的所有元素做过滤，只输出商品名称为苹果的订单
                .apply(new AllWindowFunction<Tuple2<String, Integer>, String, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<Tuple2<String, Integer>> values,
                            Collector<String> out) throws Exception {
                        values.forEach(v -> {
                            if(v.f0.contentEquals("苹果")) {
                                out.collect(v.f0 + ":" + v.f1);
                            }
                        });
                    }
                }).print();

        env.execute("Flink Streaming Java API Skeleton");
    }

2> 苹果:1
3> 苹果:1
4> 苹果:1
1> 苹果:1
2> 苹果:1
3> 苹果:1
4> 苹果:1

https://github.com/chaoxxx/learn-flink-stream-api/blob/master/src/main/java/fun/cosmozhu/session12/StreamTest.java