4. Flink CEP 高级模式：组合模式、循环模式、条件模式、模式组、模式序列、模式跳过策略

在实时规则引擎中，简单的事件匹配往往无法满足复杂业务逻辑的需求。Flink CEP 提供了一套丰富的高级模式构造，允许开发者以声明式的方式定义事件之间的时序、数量、逻辑关系以及匹配策略。本章将深入剖析六种核心高级模式，并结合在线推理场景给出实战案例。

4.1 模式基础回顾与高级分类

Flink CEP 的模式（Pattern）分为两大类：单例模式（Singleton Pattern）和循环模式（Looping Pattern）。高级模式是在此基础上通过组合、条件、序列化等手段构建的复合逻辑单元。

4.2 组合模式：严格连续与非严格连续

组合模式定义了事件之间的时序关系。Flink CEP 提供两种核心组合方式：

• 严格连续（.next()）：要求事件 A 之后必须紧跟着事件 B，中间不能有任何其他事件。
• 非严格连续（.followedBy()）：允许事件 A 之后任意位置出现事件 B，中间可以插入其他事件。

// 严格连续：登录成功后立即支付失败
Pattern<Event, ?> strictPattern = Pattern
    .<Event>begin("login")
    .where(e -> e.getType().equals("LOGIN_SUCCESS"))
    .next("payment")
    .where(e -> e.getType().equals("PAYMENT_FAILURE"));

// 非严格连续：登录成功后，后续任意时刻发生支付失败
Pattern<Event, ?> loosePattern = Pattern
    .<Event>begin("login")
    .where(e -> e.getType().equals("LOGIN_SUCCESS"))
    .followedBy("payment")
    .where(e -> e.getType().equals("PAYMENT_FAILURE"));

4.3 循环模式：次数限定与贪婪匹配

循环模式用于匹配同一事件类型连续出现多次的场景。Flink CEP 提供三种次数限定方式：

贪婪匹配（.greedy()）是循环模式的重要补充。默认情况下，循环模式采用宽松匹配，即尽可能少地匹配事件；启用贪婪后，会尽可能多地匹配事件。

// 非贪婪：匹配 2 次后立即停止
Pattern<Event, ?> nonGreedy = Pattern
    .<Event>begin("errors")
    .where(e -> e.getType().equals("ERROR"))
    .times(2, 5);

// 贪婪：在满足条件的前提下，尽可能匹配更多次
Pattern<Event, ?> greedy = Pattern
    .<Event>begin("errors")
    .where(e -> e.getType().equals("ERROR"))
    .times(2, 5)
    .greedy();

4.4 条件模式：组合条件与迭代条件

条件模式通过 .where()、.or()、.until() 方法对事件属性进行过滤。Flink CEP 支持两种条件类型：

• 简单条件（Simple Condition）：基于当前事件的属性判断。
• 迭代条件（Iterative Condition）：基于之前已匹配事件的状态进行判断，使用 IterativeCondition。

// 简单条件：金额大于 10000
.where(e -> e.getAmount() > 10000)

// 迭代条件：当前事件金额大于之前所有事件金额之和
.where(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context<Event> ctx) throws Exception {
        double sum = ctx.getEventsForPattern("transactions")
                        .stream()
                        .mapToDouble(Event::getAmount)
                        .sum();
        return value.getAmount() > sum;
    }
})

// 组合条件：金额大于 10000 且类型为转账
.where(e -> e.getAmount() > 10000)
.where(e -> e.getType().equals("TRANSFER"))

4.5 模式组：嵌套子模式与逻辑分组

模式组允许在一个模式内部嵌套另一个完整的模式，用于表达子事件序列的逻辑。模式组通过 Pattern.begin(Pattern) 或 .next(Pattern) 创建。

// 定义子模式：连续 3 次登录失败
Pattern<Event, ?> loginFailurePattern = Pattern
    .<Event>begin("fail")
    .where(e -> e.getType().equals("LOGIN_FAIL"))
    .times(3);

// 主模式：登录失败子模式发生后，紧接着发生支付成功
Pattern<Event, ?> mainPattern = Pattern
    .<Event>begin(loginFailurePattern)
    .next("payment")
    .where(e -> e.getType().equals("PAYMENT_SUCCESS"));

模式组的主要应用场景：

• 多阶段风控：将“异常登录检测”作为子模式，嵌入到“交易风控”主模式中。
• 事件聚合：将多个细粒度事件聚合成一个逻辑单元，便于后续组合。
• 规则复用：将通用子模式定义为变量，在多个规则中重复使用。

4.6 模式序列：多阶段事件链路

模式序列是将多个模式按照业务逻辑顺序串联起来，形成一条完整的事件处理链路。Flink CEP 通过链式调用 .next()、.followedBy()、.followedByAny() 构建序列。

// 三阶段风控序列：登录 → 浏览 → 支付
Pattern<Event, ?> riskSequence = Pattern
    .<Event>begin("login")
    .where(e -> e.getType().equals("LOGIN"))
    .followedBy("browse")
    .where(e -> e.getType().equals("BROWSE"))
    .followedBy("payment")
    .where(e -> e.getType().equals("PAYMENT"))
    .within(Time.minutes(30));  // 整体超时 30 分钟

4.7 模式跳过策略：控制匹配重叠与优先级

模式跳过策略（After Match Skip Strategy）决定了当一个匹配完成后，下一个匹配从何处开始。这是解决重叠匹配和匹配优先级问题的关键机制。

Flink CEP 提供四种跳过策略：

// 使用 SKIP_TO_NEXT 策略，避免重复匹配
Pattern<Event, ?> pattern = Pattern
    .<Event>begin("a")
    .where(e -> e.getType().equals("A"))
    .next("b")
    .where(e -> e.getType().equals("B"))
    .within(Time.seconds(10));

PatternStream<Event> patternStream = CEP.pattern(
    inputStream,
    pattern,
    AfterMatchSkipStrategy.skipToNext()
);

4.8 综合实战：在线风控规则引擎中的高级模式应用

以下是一个完整的在线风控规则示例，综合运用了组合模式、循环模式、条件模式、模式组和跳过策略：

// 规则：30 分钟内，同一设备连续 3 次登录失败后，发生任意金额的支付，触发风控告警
Pattern<Event, ?> fraudPattern = Pattern
    .<Event>begin(
        // 模式组：连续 3 次登录失败
        Pattern.<Event>begin("login_fail")
            .where(e -> e.getType().equals("LOGIN_FAIL"))
            .times(3)
            .greedy()
    )
    .next("payment")
    .where(e -> e.getType().equals("PAYMENT"))
    .within(Time.minutes(30));

// 使用 SKIP_TO_NEXT 避免重复告警
PatternStream<Event> fraudStream = CEP.pattern(
    eventStream.keyBy(Event::getDeviceId),
    fraudPattern,
    AfterMatchSkipStrategy.skipToNext()
);

// 处理匹配结果
fraudStream.process(new PatternProcessFunction<Event, Alert>() {
    @Override
    public void processMatch(
            Map<String, List<Event>> match,
            Context ctx,
            Collector<Alert> out) throws Exception {
        List<Event> loginFails = match.get("login_fail");
        Event payment = match.get("payment").get(0);
        out.collect(new Alert(
            payment.getDeviceId(),
            "连续登录失败后支付",
            loginFails.size(),
            payment.getAmount()
        ));
    }
});

4.9 本章小结

Flink CEP 的高级模式为实时规则引擎提供了强大的事件匹配能力。通过组合模式控制事件时序，循环模式处理重复事件，条件模式实现精细化过滤，模式组支持嵌套逻辑，模式序列构建多阶段链路，跳过策略管理匹配重叠，开发者可以灵活应对各种复杂的在线推理场景。在实际应用中，建议根据业务需求选择合适的模式组合，并始终关注状态管理和性能优化。