3.1 什么是复杂事件处理（CEP）

复杂事件处理（Complex Event Processing, CEP）是一种在连续数据流中检测符合特定规则的事件组合的技术。其核心思想是：从多个简单事件中识别出有意义的复杂事件模式。

在实时风控场景中，单次登录失败可能只是用户输错密码，但“5分钟内连续登录失败3次+异地IP+更换设备”这一事件组合，就极大概率是暴力破解攻击。CEP正是用来描述和检测这类组合规则的计算范式。

3.2 CEP 与普通流处理的区别

普通流处理（如简单的 map/filter/reduce）与 CEP 在抽象层次、关注点和表达能力上有本质差异。下表从多个维度进行对比：

3.3 Flink CEP 核心组件

Flink 提供了完整的 CEP 库，其核心由三个组件构成：Pattern（模式定义）、PatternStream（模式流）、Select/FlatSelect（结果提取）。三者形成一条清晰的“定义→匹配→输出”流水线。

3.3.1 Pattern：模式定义

Pattern 是 CEP 的“规则描述语言”，通过链式 API 定义事件序列的约束条件。每个 Pattern 节点包含：

• 事件类型：通过 .where() 指定过滤条件
• 量词：.oneOrMore()、.times(n)、.optional()
• 连续性策略：.next()（严格连续）、.followedBy()（宽松连续）、.notNext()（否定）
• 时间约束：.within(Time.minutes(5)) 限定匹配窗口

// 示例：定义"5分钟内连续登录失败3次"的模式
Pattern<LoginEvent, LoginEvent> failPattern = Pattern
    .<LoginEvent>begin("first")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent event) {
            return event.getStatus().equals("FAIL");
        }
    })
    .times(3)
    .consecutive()
    .within(Time.minutes(5));

3.3.2 PatternStream：模式流

将定义好的 Pattern 应用到数据流上，即得到 PatternStream。它是 CEP 匹配引擎的运行时视图，内部维护了 NFA（非确定性有限自动机）状态机，负责在事件到达时进行状态转移和匹配判定。

// 将 Pattern 应用到 DataStream 上
DataStream<LoginEvent> inputStream = ...;
PatternStream<LoginEvent> patternStream = CEP.pattern(
    inputStream,
    failPattern
);

PatternStream 本身不包含事件数据，它只是一个“已绑定模式的数据流句柄”，后续通过 select/flatSelect 从中提取匹配结果。

3.3.3 Select / FlatSelect：结果提取

匹配完成后，需要从 PatternStream 中提取出匹配的事件序列。Flink 提供两种方式：

// 使用 select 提取匹配结果
patternStream.select(new PatternSelectFunction<LoginEvent, Alert>() {
    @Override
    public Alert select(Map<String, List<LoginEvent>> pattern) throws Exception {
        List<LoginEvent> first = pattern.get("first");
        // 构建告警对象
        return new Alert(first.get(0).getUserId(), "暴力破解", System.currentTimeMillis());
    }
});

3.4 核心组件协作流程

下图展示了 Flink CEP 三个核心组件在运行时的协作关系：

3.5 三组件关系总结

下表从职责、输入输出、生命周期三个维度总结三个核心组件：