Apache Flink中CoFlatMapFunction未按事件时间顺序处理事件

你遇到的这个问题其实是Flink中ConnectedStream + CoFlatMapFunction的常见认知误区——CoFlatMapFunction本身并不会基于事件时间对两个输入流的元素做全局排序，它只是被动处理每个流到达的元素，完全按照元素的实际到达顺序来调用flatMap1或flatMap2。

在你的测试代码里，偶数源是一次性把所有偶数（0,2,4...18）全部发送完毕，之后奇数源才开始发送所有奇数（1,3,5...19），所以元素的到达顺序就是先全偶数、再全奇数，自然会先打印所有偶数，再打印奇数，这和你设置的事件时间戳没有直接关系——因为Flink不会为了事件时间顺序，主动暂停一个流的处理去等待另一个流的元素。

如何实现按事件时间顺序输出？ #

如果想要严格按照事件时间戳的顺序输出两个流的元素，你需要将两个流合并后，通过状态和定时器来实现全局排序，具体步骤如下：

1. 合并两个输入流 #

首先用union将偶数流和奇数流合并成一个统一的数据流：

DataStream<Integer> combinedStream = evenStream.union(oddStream);

2. 统一分配时间戳与水位线 #

虽然你的源已经在发送水位线，但合并后的流最好统一处理时间戳和水位线（这里用单调递增的水位线策略，匹配你示例中时间戳递增的场景）：

DataStream<Integer> timedStream = combinedStream
    .assignTimestampsAndWatermarks(WatermarkStrategy.<Integer>forMonotonousTimestamps()
        .withTimestampAssigner((element, recordTimestamp) -> element));

3. 用ProcessFunction实现按事件时间排序 #

通过ProcessFunction维护一个有序集合（比如TreeSet），并利用定时器在水位线推进时，输出所有时间戳小于等于当前水位线的元素，保证输出顺序符合事件时间：

timedStream.process(new ProcessFunction<Integer, Integer>() {
    // 维护一个有序集合，存储待输出的元素
    private transient ValueState<TreeSet<Integer>> sortedElements;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<TreeSet<Integer>> descriptor = new ValueStateDescriptor<>(
            "sortedElements",
            TypeInformation.of(new TypeHint<TreeSet<Integer>>() {})
        );
        sortedElements = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
        TreeSet<Integer> set = sortedElements.value();
        if (set == null) {
            set = new TreeSet<>();
        }
        // 将元素加入有序集合
        set.add(value);
        sortedElements.update(set);
        // 注册定时器，当水位线推进到当前元素的时间戳时触发
        ctx.timerService().registerEventTimeTimer(value);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
        TreeSet<Integer> set = sortedElements.value();
        if (set != null && !set.isEmpty()) {
            // 输出所有时间戳小于等于当前水位线的元素
            while (!set.isEmpty() && set.first() <= timestamp) {
                Integer element = set.pollFirst();
                out.collect(element);
                System.out.println(element);
            }
            sortedElements.update(set);
        }
    }
})
.setParallelism(1); // 全局排序需要单并行度，否则无法保证全局顺序

补充说明 #

• 如果你的实际场景中，两个源的元素是交错到达的（比如0,1,2,3依次发送），那么CoFlatMapFunction会按元素到达顺序处理，这时候输出会符合预期；但如果源的发送顺序和事件时间顺序不一致，就必须用上述的排序逻辑。
• 全局排序需要单并行度，这会影响作业的吞吐量，如果你的数据量很大，需要权衡排序需求和性能。

内容的提问来源于stack exchange，提问作者Sebastian