2. Flink CEP 基础：CEP 概念、事件模型、模式定义、模式序列、模式组

2.1 CEP 核心概念与事件模型

复杂事件处理（Complex Event Processing, CEP）是在无界事件流中，通过定义事件之间的时序、逻辑与依赖关系，检测出符合特定规则的事件组合。Flink CEP 是 Flink 提供的专用库，用于在 DataStream 上执行模式匹配，是风控实时规则引擎的核心技术之一。

2.1.1 CEP 基本术语

2.1.2 事件模型与时间语义

Flink CEP 的事件模型基于 DataStream，每个事件需实现 equals() 和 hashCode() 方法。事件通常包含以下字段：

• 事件 ID：全局唯一标识
• 事件类型：如 LOGIN、TRANSFER、PASSWORD_CHANGE
• 业务属性：用户 ID、金额、IP、设备指纹等
• 时间戳：事件发生时间（Event Time）或处理时间（Processing Time）

// 事件 POJO 示例
public class RiskEvent {
    private String eventId;
    private String userId;
    private String eventType;
    private double amount;
    private long timestamp;

    // 必须实现 equals 和 hashCode，用于 CEP 状态管理
    @Override
    public boolean equals(Object o) { ... }
    @Override
    public int hashCode() { ... }
}

2.2 模式定义（Pattern Definition）

Flink CEP 使用 Pattern<T, F> 类定义匹配规则。模式定义包含三个核心维度：条件、量词、时间约束。

2.2.1 条件定义

2.2.2 量词与时间约束

// 模式定义完整示例：5 分钟内同一用户连续 3 次登录失败
Pattern<RiskEvent, RiskEvent> loginFailPattern = Pattern
    .<RiskEvent>begin("first")
    .where(evt -> "LOGIN_FAIL".equals(evt.getEventType()))
    .times(3)                          // 量词：连续出现 3 次
    .within(Time.minutes(5));          // 时间约束：5 分钟内

2.3 模式序列（Pattern Sequence）

多个模式通过 .next()、.followedBy()、.followedByAny() 连接形成模式序列，控制事件之间的连续性策略。

2.3.1 连续性策略对比

// 模式序列示例：登录失败 → 登录成功 → 大额转账（宽松连续）
Pattern<RiskEvent, ?> riskSequence = Pattern
    .<RiskEvent>begin("fail")
    .where(evt -> "LOGIN_FAIL".equals(evt.getEventType()))
    .followedBy("success")
    .where(evt -> "LOGIN_SUCCESS".equals(evt.getEventType()))
    .followedBy("transfer")
    .where(evt -> "TRANSFER".equals(evt.getEventType()) && evt.getAmount() > 50000)
    .within(Time.hours(1));

2.4 模式组（Pattern Group）

模式组允许将多个模式序列组合为一个复合模式，用于表达“且”或“或”的复杂逻辑关系。Flink CEP 通过 Pattern.begin() 嵌套调用实现模式组。

2.4.1 模式组定义方式

// 模式组示例：检测 "登录失败 3 次" 且 "5 分钟内尝试 2 个不同设备"
Pattern<RiskEvent, ?> deviceGroup = Pattern
    .<RiskEvent>begin("device_check")
    .where(evt -> "LOGIN_FAIL".equals(evt.getEventType()))
    .times(3)
    .within(Time.minutes(5));

// 将 deviceGroup 作为子模式嵌入主序列
Pattern<RiskEvent, ?> mainPattern = Pattern
    .<RiskEvent>begin("group_start")
    .where(evt -> "LOGIN_FAIL".equals(evt.getEventType()))
    .followedBy(deviceGroup)
    .within(Time.minutes(10));

2.4.2 模式组在风控中的典型用法

2.5 CEP 模式匹配流程总览

下图展示了 Flink CEP 从事件输入到模式匹配输出的完整流程，包含模式定义、序列组合、状态机执行与匹配输出四个核心阶段。

2.6 风控 CEP 最佳实践

// 超时处理示例
PatternStream<RiskEvent> patternStream = CEP.pattern(inputStream, pattern);
patternStream.select(
    new PatternTimeoutFunction<RiskEvent, Alert>() {
        @Override
        public Alert timeout(Map<String, List<RiskEvent>> pattern, long timeoutTimestamp) {
            // 超时部分匹配：记录日志并清理状态
            return new Alert("PARTIAL_MATCH_TIMEOUT", pattern);
        }
    },
    new PatternSelectFunction<RiskEvent, Alert>() {
        @Override
        public Alert select(Map<String, List<RiskEvent>> pattern) {
            // 完整匹配：生成风控告警
            return new Alert("FULL_MATCH", pattern);
        }
    }
);