4. Flink CEP 快速入门：环境搭建与第一个 Demo

本章是 Flink CEP 实战的起点。我们将从零搭建开发环境，选择兼容的 Flink 版本，配置 Maven 依赖，并编写第一个 CEP 程序——检测连续登录失败事件。通过逐行解析代码，理解 CEP 的核心概念：事件流、Pattern 定义、状态转移与匹配输出。

4.1 环境搭建：Maven 依赖与 Flink 版本选择

Flink CEP 是 Flink 的官方库，与 Flink 核心版本严格绑定。版本不匹配会导致运行时 NoSuchMethodError 或 ClassNotFoundException。推荐使用 Flink 1.17.x 或 1.18.x，这两个版本对 CEP 库支持最稳定，且与后续章节的 PSI 监控模块兼容。

以下为完整的 Maven 依赖配置（pom.xml 核心片段）：

<properties>
    <flink.version>1.17.2</flink.version>
    <java.version>11</java.version>
    <scala.binary.version>2.12</scala.binary.version>
</properties>

<dependencies>
    <!-- Flink 核心客户端 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <!-- Flink CEP 库 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-cep</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <!-- 日志桥接 -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.36</version>
    </dependency>
</dependencies>

4.2 第一个 CEP Demo：检测连续登录失败事件

业务场景：在 5 分钟内，同一用户连续登录失败 3 次及以上，则触发风控告警。我们将用 CEP 的 Pattern<T> 定义该规则，并在 DataStream 上应用。

4.2.1 事件模型定义

// LoginEvent.java
public class LoginEvent {
    private String userId;
    private String ip;
    private String status;   // "fail" 或 "success"
    private Long timestamp;

    public LoginEvent() {}

    public LoginEvent(String userId, String ip, String status, Long timestamp) {
        this.userId = userId;
        this.ip = ip;
        this.status = status;
        this.timestamp = timestamp;
    }

    // getter / setter / toString 省略
}

4.2.2 Pattern 定义：连续三次失败

// 核心 Pattern 定义
Pattern<LoginEvent, LoginEvent> pattern = Pattern
    .<LoginEvent>begin("first")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent event) {
            return "fail".equals(event.getStatus());
        }
    })
    .next("second")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent event) {
            return "fail".equals(event.getStatus());
        }
    })
    .next("third")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent event) {
            return "fail".equals(event.getStatus());
        }
    })
    .within(Time.minutes(5));  // 时间窗口约束

4.2.3 在 DataStream 上应用 Pattern

DataStream<LoginEvent> inputStream = env
    .fromElements(
        new LoginEvent("u001", "192.168.1.1", "fail", 1000L),
        new LoginEvent("u001", "192.168.1.2", "fail", 2000L),
        new LoginEvent("u001", "192.168.1.3", "fail", 3000L),
        new LoginEvent("u002", "10.0.0.1", "fail", 4000L),
        new LoginEvent("u002", "10.0.0.2", "success", 5000L)
    )
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, ts) -> event.getTimestamp())
    );

// 将 Pattern 应用到流上
PatternStream<LoginEvent> patternStream = CEP.pattern(
    inputStream.keyBy(LoginEvent::getUserId),
    pattern
);

// 输出匹配结果
DataStream<String> alertStream = patternStream.process(
    new PatternProcessFunction<LoginEvent, String>() {
        @Override
        public void processMatch(
                Map<String, List<LoginEvent>> match,
                Context ctx,
                Collector<String> out) {
            LoginEvent first = match.get("first").get(0);
            LoginEvent second = match.get("second").get(0);
            LoginEvent third = match.get("third").get(0);
            out.collect("告警：用户 " + first.getUserId()
                + " 在 5 分钟内连续 3 次登录失败，IP 分别为 "
                + first.getIp() + " -> " + second.getIp() + " -> " + third.getIp());
        }
    }
);

alertStream.print();  // 控制台输出告警

4.3 代码逐行解析：核心机制拆解

我们将上述 Demo 拆解为 5 个关键步骤，对应 CEP 的处理流水线：

1. 事件源与时间语义： assignTimestampsAndWatermarks 为事件分配时间戳，CEP 依赖事件时间判断 within 窗口。
2. KeyBy 分组： keyBy(LoginEvent::getUserId) 将不同用户的登录事件隔离，CEP 在每个分组内独立匹配。
3. Pattern 链式定义： begin("first").next("second").next("third") 定义了严格连续（strict contiguity）的事件序列。每个节点通过 where 设置过滤条件。
4. 时间约束： .within(Time.minutes(5)) 限定整个序列必须在 5 分钟内完成，超时则丢弃部分匹配。
5. 匹配输出： PatternProcessFunction 的 processMatch 方法接收一个 Map<String, List<LoginEvent>>，key 为 Pattern 中定义的名称（first/second/third），value 为匹配到的事件列表。此处每个名称只匹配一个事件，所以取 get(0)。

4.3.1 连续性与量词对比

4.4 CEP 运行时架构图

下图展示了 CEP 在 Flink 算子内部的运行流程：从 DataStream 输入，经过 KeyBy 分组、NFA（非确定性有限自动机）状态转移，到匹配结果输出。

4.5 常见踩坑与排查

• 事件时间未设置： 未调用 assignTimestampsAndWatermarks 时，within 默认使用 Processing Time，导致结果与预期不符。务必为事件时间模式指定时间戳提取器。
• KeyBy 遗漏： 不进行 KeyBy 则所有用户的事件混在一起，可能将不同用户的失败事件拼接成一次告警。风控场景必须按用户分组。
• Pattern 连续性误用： 使用 followedBy 代替 next 会导致中间插入 success 事件也能匹配，不符合“连续失败”语义。
• Watermark 延迟过大： 如果 Watermark 滞后太多，CEP 会延迟输出结果，甚至因超时丢弃匹配。建议根据业务容忍度设置 forBoundedOutOfOrderness 参数。