java

6.2 Flink CEP：复杂事件处理

Flink CEP 高级模式：times/oneOrMore/greedy 量词、optional 可选模式、notNext/notFollowedBy 否定模式、迭代模式与循环条件、侧输出流处理匹配失败、以及 CEP 性能调优与状态清理策略。

Leeting Yan 2026-11-01 2 分钟阅读 933 字

量词

CEP 支持正则表达式风格的量词：

times：固定次数

Pattern.<LoginEvent>begin("fail")
    .where(evt -> evt.status().equals("FAIL"))
    .times(3)  // 正好 3 次
    .within(Duration.ofMinutes(5));

timesOrMore：最少 N 次

Pattern.<LoginEvent>begin("fail")
    .where(evt -> evt.status().equals("FAIL"))
    .timesOrMore(3)  // 3 次或更多
    .within(Duration.ofMinutes(5));

oneOrMore：1 次或更多

Pattern.<LoginEvent>begin("fail")
    .where(evt -> evt.status().equals("FAIL"))
    .oneOrMore()  // 等价于 timesOrMore(1)
    .within(Duration.ofMinutes(5));

greedy：贪婪匹配

Pattern.<OrderEvent>begin("orders")
    .where(evt -> evt.type().equals("ORDER"))
    .oneOrMore()
    .greedy()  // 尽可能多的匹配
    .next("end")
    .where(evt -> evt.type().equals("CHECKOUT"))
    .within(Duration.ofMinutes(30));

贪婪匹配优先选择最长的序列。

optional：可选模式

Pattern.<OrderEvent>begin("cart")
    .where(evt -> evt.type().equals("ADD_CART"))
    .next("coupon")
    .where(evt -> evt.type().equals("APPLY_COUPON"))
    .optional()  // 可能没有使用优惠券
    .next("checkout")
    .where(evt -> evt.type().equals("CHECKOUT"));

否定模式

notNext：后面不能紧跟

// 登录成功后 10 秒内不能有敏感操作
Pattern.<OrderEvent>begin("login")
    .where(evt -> evt.type().equals("LOGIN_SUCCESS"))
    .notNext("sensitive")
    .where(evt -> evt.type().equals("DELETE_ACCOUNT"))
    .within(Duration.ofSeconds(10));

notFollowedBy：后面不能有

// 大额转账后 1 小时内不能有退款
Pattern.<Transaction>begin("transfer")
    .where(evt -> evt.type().equals("LARGE_TRANSFER"))
    .notFollowedBy("refund")
    .where(evt -> evt.type().equals("REFUND"))
    .within(Duration.ofHours(1));

注意：notFollowedBy 模式不能作为 Pattern 的最后一个元素（因为没有后续事件来确认否定）。通常后面接一个虚拟的结束事件或 within。

迭代模式（循环条件）

Pattern.<LoginEvent>begin("fail")
    .where(evt -> evt.status().equals("FAIL"))
    .timesOrMore(3)
    // until 接受 IterativeCondition，方法签名是 filter(value, ctx) 两个参数；
    // 写成单参数的 evt -> ... 编译不过
    .until((evt, ctx) -> evt.status().equals("SUCCESS"))  // 直到成功为止
    .within(Duration.ofMinutes(10));

until 定义循环的终止条件。

侧输出流

CEP 的匹配结果通过 select 提取，也可以发送到侧输出：

OutputTag<Alert> lateTag = new OutputTag<Alert>("late-matches"){};

PatternStream<Event> patternStream = CEP.pattern(stream, pattern);

SingleOutputStreamOperator<Alert> result = patternStream.select(lateTag,
    new PatternTimeoutFunction<Event, Alert>() {
        @Override
        public Alert timeout(Map<String, List<Event>> partial, long timestamp) {
            return new Alert("TIMEOUT", timestamp);
        }
    },
    new PatternSelectFunction<Event, Alert>() {
        @Override
        public Alert select(Map<String, List<Event>> pattern) {
            return new Alert("MATCH");
        }
    });

// 获取超时事件
DataStream<Alert> timeouts = result.getSideOutput(lateTag);

状态清理

CEP 状态持续增长（每个未完成的模式都占用内存）。控制策略：

缩短 within 时间：模式超时后状态随事件时间一起清理，这是最有效的控制手段
保证 Watermark 能推进：CEP 按事件时间清理超时模式，某个分区长期空闲会拖住 Watermark，未完成的模式就一直留着；用 withIdleness 把空闲分区排除在 Watermark 计算之外
控制 key 基数：每个未完成模式按 key 保存，热 key 少、分布均匀时状态更可控

env.setStateBackend(new EmbeddedRocksDBStateBackend());

// 启用增量 Checkpoint
env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");

// 在 WatermarkStrategy 中设置 idle timeout
WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
    .withIdleness(Duration.ofMinutes(5));

eventd v3 的 CEP 规则

eventd v3 的风控 CEP：

// 检测：同一用户 5 分钟内连续失败 3 次后成功（可疑）
Pattern.<LoginEvent>begin("fails")
    .where(evt -> evt.status().equals("FAIL"))
    .timesOrMore(3)
    .greedy()
    .followedBy("success")
    .where(evt -> evt.status().equals("SUCCESS"))
    .within(Duration.ofMinutes(5));

// 检测：同一 IP 1 分钟内下单 >= 10 次（刷单）
// 注意 timesOrMore(10) 的语义是「至少 10 次」，不是「超过 10 次」
Pattern.<OrderEvent>begin("orders")
    .where(evt -> evt.type().equals("ORDER"))
    .timesOrMore(10)
    .within(Duration.ofMinutes(1));

性能调优

策略	效果
缩短 `within`	减少未完成模式的状态
使用 `greedy`	减少分支匹配数
keyBy 粒度细	每个 key 的状态更小
RocksDB 后端	大状态不耗内存
增量 Checkpoint	减少 Checkpoint 时间

小结

量词	API	含义
固定次数	`.times(n)`	正好 n 次
最少 N 次	`.timesOrMore(n)`	>= n 次
1 次或更多	`.oneOrMore()`	>= 1 次
贪婪	`.greedy()`	最大化匹配
可选	`.optional()`	0 或 1 次
否定紧邻	`.notNext()`	A 后面不能是 B
否定跟随	`.notFollowedBy()`	A 后面不能有 B

下一节 6.3 Flink Table API 与 SQL（概述）。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页

量词