flink count+time trigger

Flink 中的 Count+Time 触发器是一种组合触发器，它在满足两个条件之一时触发操作。其中一个条件是元素数量达到指定的阈值，另一个条件是时间达到指定的时间间隔。

在 Flink 中，可以使用 `countTrigger()` 和 `timeTrigger()` 方法来创建 Count+Time 触发器。下面是一个示例代码：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.triggers.PurgingTrigger;
import org.apache.flink.streaming.api.windowing.triggers.TimeTrigger;

public class CountTimeTriggerExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        DataStream<Tuple2<String, Integer>> stream = env.addSource(new SourceFunction<Tuple2<String, Integer>>() {
            private volatile boolean running = true;

            @Override
            public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
                while (running) {
                    ctx.collect(Tuple2.of("key", 1));
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        });

        stream.map(new MapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
                return value;
            }
        })
        .keyBy(0)
        .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
        .trigger(PurgingTrigger.of(CountTrigger.of(3)).or(TimeTrigger.create()))
        .sum(1)
        .print();

        env.execute();
    }
}

在上面的示例代码中，我们首先创建一个包含无限元素的源数据流 `stream`，然后使用 `window` 方法定义一个时间窗口，并使用 `trigger` 方法设置 Count+Time 触发器。在这个例子中，触发器的条件是元素数量达到 3 或者时间间隔达到 5 秒。

最后，我们对窗口内的元素进行求和，并打印结果。执行代码后，每当满足触发器的条件时，就会触发一次计算和打印操作。