Flink CEP 完整实战：订单创建支付匹配 + 超时未支付侧输出踩坑全解

　Palpitate.

234人浏览 · 2026-06-11 20:48:17

　Palpitate. · 2026-06-11 20:48:17 发布

第一段代码

package com.day08;
import com.alibaba.fastjson.JSON;
import lombok.SneakyThrows;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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.windowing.time.Time;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import java.text.ParseException;
import java.time.Duration;
import java.util.*;
public class demo02Cep {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("hadoop11:9092")
                .setTopics("topic3")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema()).build();
        DataStreamSource<String> ds1 = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");
        SingleOutputStreamOperator<payEvent> map = ds1.map(new MapFunction<String, payEvent>() {
            @Override
            public payEvent map(String s) throws Exception {
                payEvent payEvent = JSON.parseObject(s, payEvent.class);
                return payEvent;
            }
        });
        SingleOutputStreamOperator<payEvent> payEventSingleOutputStreamOperator = map.assignTimestampsAndWatermarks(WatermarkStrategy.<payEvent>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(new SerializableTimestampAssigner<payEvent>() {
            @SneakyThrows
            @Override
            public long extractTimestamp(payEvent payEvent, long l) {
                Date date = DateUtils.parseDate(payEvent.getTs(), "yyyy-MM-dd HH:mm:ss");
                return date.getTime();
            }
        }));

        KeyedStream<payEvent, Integer> payEventIntegerKeyedStream = payEventSingleOutputStreamOperator.keyBy(new KeySelector<payEvent, Integer>() {
            @Override
            public Integer getKey(payEvent payEvent) throws Exception {
                return payEvent.getUserId();
            }
        });

        Pattern<payEvent, payEvent> where = Pattern.<payEvent>begin("first", AfterMatchSkipStrategy.skipPastLastEvent()).where(new SimpleCondition<payEvent>() {
            @Override
            public boolean filter(payEvent payEvent) throws Exception {
                return payEvent.getType().equals("create");
            }
        }).next("second").where(new SimpleCondition<payEvent>() {
            @Override
            public boolean filter(payEvent payEvent) throws Exception {
                return payEvent.getType().equals("pay");
            }
        });
        PatternStream<payEvent> pattern2 = CEP.pattern(payEventIntegerKeyedStream, where);
        SingleOutputStreamOperator<Set<Integer>> select = pattern2.select(new PatternSelectFunction<payEvent, Set<Integer>>() {
            @Override
            public Set<Integer> select(Map<String, List<payEvent>> map) throws Exception {
                HashSet<Integer> set = new HashSet<>();
                Collection<List<payEvent>> values = map.values();
                for (List<payEvent> value : values) {
                    for (payEvent payEvent : value) {
                        set.add(payEvent.getUserId());
                    }

                }
                return set;
            }
        });
        select.print();
        env.execute();

    }
}

踩坑一：

发送了

{"userId":"4","type":"create","ts":"2024-07-18 11:14:10"}
{"userId":"4","type":"pay","ts":"2024-07-18 11:14:11"}

控制台却无任何输出

继续发送

{"userId":"4","type":"pay","ts":"2024-07-18 11:15:11"}

控制台却有输出

原因：

水印配置：forBoundedOutOfOrderness(Duration.ZERO)，水位 = 整条流全局最大事件时间

第一条消息：userId4 create 11:14:10 全局最大时间 = 11:14:10，理论水位 = 11:14:10；但 Flink 水印是周期性推送（默认 200ms），不会收到一条立刻广播到 CEP 算子。 CEP 存入 create 等待 pay。
第二条消息：userId4 pay 11:14:11 全局最大时间刷新 11:14:11，模式 create.next (pay) 完整匹配成功，匹配结果先被 CEP 缓存压着，不打印；此时水印还没推送到 11:14:11，水位没超过匹配结束时间 11:14:11，结果阻塞。
第三条消息：userId4 pay 11:15:11 全局最大时间直接拉高到 11:15:11，触发水印更新推送，水位同步涨到 11:15:11；水位 11:15:11 > 匹配片段结束时间 11:14:11，阻塞的匹配结果瞬间释放，控制台打印输出。

第二段代码：

package com.day08;


import com.alibaba.fastjson.JSON;
import lombok.SneakyThrows;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.*;

public class demo02CeDao{
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("hadoop11:9092")
                .setTopics("topic3")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema()).build();

        DataStreamSource<String> ds1 = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");
        SingleOutputStreamOperator<payEvent> map = ds1.map(new MapFunction<String, payEvent>() {
            @Override
            public payEvent map(String s) throws Exception {
                return JSON.parseObject(s, payEvent.class);
            }
        });
        SingleOutputStreamOperator<payEvent> payEventSingleOutputStreamOperator = map.assignTimestampsAndWatermarks(WatermarkStrategy.<payEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<payEvent>() {
            @SneakyThrows
            @Override
            public long extractTimestamp(payEvent payEvent, long l) {
                Date date = DateUtils.parseDate(payEvent.getTs(), "yyyy-MM-dd HH:mm:ss");
                return date.getTime();
            }
        }));
        KeyedStream<payEvent, Integer> payEventIntegerKeyedStream = payEventSingleOutputStreamOperator.keyBy(new KeySelector<payEvent, Integer>() {
            @Override
            public Integer getKey(payEvent payEvent) throws Exception {
                return payEvent.getUserId();
            }
        });

        Pattern<payEvent, payEvent> within = Pattern.<payEvent>begin("first", AfterMatchSkipStrategy.skipPastLastEvent()).where(new SimpleCondition<payEvent>() {
            @Override
            public boolean filter(payEvent payEvent) throws Exception {
                return payEvent.getType().equals("create");
            }
        }).next("second").where(new SimpleCondition<payEvent>() {
            @Override
            public boolean filter(payEvent payEvent) throws Exception {
                return payEvent.getType().equals("pay");
            }
        }).within(Time.minutes(10));
        OutputTag<Integer> outputTag = new OutputTag<Integer>("x1") {};
        PatternStream<payEvent> pattern = CEP.pattern(payEventIntegerKeyedStream, within);
        SingleOutputStreamOperator<Integer> select = pattern.select(outputTag, new PatternTimeoutFunction<payEvent, Integer>() {
            @Override
            public Integer timeout(Map<String, List<payEvent>> map, long l) throws Exception {
                // 严格匹配first，不能写错
                List<payEvent> firstList = map.get("first");
                // 双层判空
                if (firstList != null && !firstList.isEmpty()) {
                    return firstList.get(0).getUserId();
                }
                // 兜底返回，防止null
                return -1;

            }
        }, new PatternSelectFunction<payEvent, Integer>() {
            @Override
            public Integer select(Map<String, List<payEvent>> map) throws Exception {
                return map.get("frist").get(0).getUserId();
            }
        });
        select.print("正常的数据：");
        SideOutputDataStream<Integer> sideOutput = select.getSideOutput(outputTag);
        sideOutput.print("超时的数据");
        env.execute();

    }
}

踩坑二：

这段代码是正常输出1001的

{"userId":1001,"type":"create","ts":"2028-08-11 14:00:00"}
{"userId":1001,"type":"pay","ts":"2028-08-11 14:09:01"}
{"userId":1001,"type":"pay","ts":"2028-08-11 14:10:06"}

但下面这段代码却输出的是超时1001，但我明明设置了5秒的延迟


{"userId":1001,"type":"create","ts":"2028-09-11 14:00:00"}
{"userId":1001,"type":"pay","ts":"2028-09-11 14:10:01"}
{"userId":1001,"type":"pay","ts":"2028-09-11 14:10:06"}

原因：

Watermark 5 秒（forBoundedOutOfOrderness (Duration.ofSeconds (5))）它只负责过滤极度迟到的单条消息，规则：
一条消息的事件时间＜当前全局水印时间 −5 秒 → 这条消息直接被丢弃，连 CEP 算子都进不去
它完全不修改、不延长 CEP within 的 10 分钟匹配生命周期。
within (Time.minutes (10)) 是 CEP 匹配链的生命周期死线
从first(create)的事件时间开始，固定 10 分钟事件时间有效期：
有效期区间：[createTs , createTs + 10min]
在区间内收到 pay → 匹配成功走主流
超过区间之后才来 pay → 匹配链已经过期关闭，pay 无法配对；等到水印越过区间终点，这条 create 直接判定超时进侧输出
水印 5 秒不会拓宽这个 10 分钟区间，区间永远卡死 10 分钟。
如果你想把匹配有效期拉长到 10 分 05 秒该怎么做

只能直接修改 within 时长，水印改不了匹配窗口：
```
// 想要10分05秒有效期，直接写10分5秒
.within(Time.minutes(10).plus(Time.seconds(5)));
```

亚马逊云科技技术品牌专区

更多推荐

基于STM32的全屋环境智能监测与预警终端设计

亚马逊云科技技术品牌专区

Keras：面向人类的深度学习库

Keras3支持多后端（TensorFlow、PyTorch、JAX），通过设置环境变量切换，兼容指定版本。pip安装后可用Sequential API快速构建CNN模型，如MNIST分类示例，实

亚马逊云科技技术品牌专区

AI行业日报｜2026-09-25：5个热点事件

本期行业动态聚焦于端侧算力重构、自主智能体工作流、开发工具经济模型、网络隔离架构设计与公共部门合规审查。以下结合一线技术场景进行结构化梳理，供 AI 研究员与软件工程团队参考。 PrismML 正将其轻量级语言模型适配至搭载高通处理器的智能眼镜终端 1 。该平台的核心战略指向开源权重模型的本地化运行，旨在最大化挖掘穿戴设备现有算力的利用效率 1 。对于嵌入式与边缘计算开发者而言，此类微型模型向低功