4. 预警规则模型：基于阈值规则、基于滑动窗口规则、基于CEP复杂事件规则

预警规则模型是实时预警系统的核心决策引擎，它定义了“什么样的数据模式会触发告警”。在Flink生态中，根据业务场景的复杂程度，通常将预警规则分为三个层次：阈值规则（单点判断）、滑动窗口规则（聚合趋势判断）和CEP复杂事件规则（时序模式匹配）。本章将逐一剖析这三种模型的结构、适用场景及评估要点。

4.1 基于阈值规则

阈值规则是最基础的预警模型，它针对单条数据流中的某个指标字段，与预设的固定或动态阈值进行比较。当指标超过（或低于）阈值时，立即触发告警。

4.1.1 规则结构

4.1.2 Flink实现片段

// 基于阈值规则的Flink DataStream处理示例
DataStream<MetricEvent> metricStream = ...;

DataStream<AlertEvent> thresholdAlerts = metricStream
    .filter(event -> {
        // 规则：CPU使用率 > 90%
        if ("cpu_usage".equals(event.getMetricName())) {
            return event.getValue() > 90.0;
        }
        return false;
    })
    .map(event -> new AlertEvent(
        event.getMetricName(),
        "THRESHOLD_VIOLATION",
        String.format("CPU usage %.2f exceeds threshold 90.0", event.getValue()),
        System.currentTimeMillis()
    ));

4.1.3 评估要点

• 误报率：阈值设置过窄容易导致频繁误报，需结合业务容忍度调整。
• 漏报率：阈值设置过宽可能遗漏真实异常，建议使用动态阈值（如基于历史百分位）。
• 冷却机制：必须引入冷却窗口，避免在持续超阈值期间产生告警风暴。

4.2 基于滑动窗口规则

滑动窗口规则不再关注单条数据，而是在一个持续移动的时间窗口内对数据进行聚合计算（如平均值、最大值、计数等），当聚合结果超过阈值时触发告警。这种模型能够有效过滤瞬时抖动，捕捉持续异常趋势。

4.2.1 窗口类型对比

4.2.2 滑动窗口预警实现

// 滑动窗口规则：过去5分钟内平均响应时间 > 2000ms 则告警
DataStream<MetricEvent> metricStream = ...;

DataStream<AlertEvent> windowAlerts = metricStream
    .keyBy(MetricEvent::getServiceId)
    .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
    .aggregate(new AvgResponseTimeAggregate())
    .filter(avgRt -> avgRt > 2000.0)
    .map(avgRt -> new AlertEvent(
        "response_time",
        "WINDOW_AVG_VIOLATION",
        String.format("Average response time %.0f ms exceeds 2000 ms", avgRt),
        System.currentTimeMillis()
    ));

4.2.3 评估要点

• 窗口大小与步长：窗口越大，告警越稳定但延迟越高；步长越小，告警越灵敏但计算开销越大。
• 水位线配置：使用事件时间时必须合理设置水位线延迟，避免数据乱序导致窗口计算不准确。
• 状态大小：滑动窗口需要维护窗口内的所有数据，需评估状态存储压力，必要时使用增量聚合函数。

4.3 基于CEP复杂事件规则

CEP（Complex Event Processing）规则用于检测多个事件之间的时序关系、逻辑组合和因果依赖。它能够表达“在10分钟内，连续3次登录失败，且IP地址不同，则触发暴力破解告警”这类复杂模式。

4.3.1 CEP模式定义

4.3.2 CEP预警规则实现

// CEP规则：10分钟内，同一用户连续3次登录失败，且IP不同，触发暴力破解告警
Pattern<LoginEvent, ?> bruteForcePattern = Pattern
    .<LoginEvent>begin("first")
        .where(new SimpleCondition<LoginEvent>() {
            @Override
            public boolean filter(LoginEvent event) {
                return "FAILED".equals(event.getStatus());
            }
        })
    .next("second")
        .where(new SimpleCondition<LoginEvent>() {
            @Override
            public boolean filter(LoginEvent event) {
                return "FAILED".equals(event.getStatus());
            }
        })
        .where(new IterativeCondition<LoginEvent>() {
            @Override
            public boolean filter(LoginEvent event, Context<LoginEvent> ctx) {
                LoginEvent first = ctx.getEventsForPattern("first").iterator().next();
                return !event.getIp().equals(first.getIp());
            }
        })
    .next("third")
        .where(new SimpleCondition<LoginEvent>() {
            @Override
            public boolean filter(LoginEvent event) {
                return "FAILED".equals(event.getStatus());
            }
        })
        .where(new IterativeCondition<LoginEvent>() {
            @Override
            public boolean filter(LoginEvent event, Context<LoginEvent> ctx) {
                LoginEvent first = ctx.getEventsForPattern("first").iterator().next();
                LoginEvent second = ctx.getEventsForPattern("second").iterator().next();
                return !event.getIp().equals(first.getIp()) && !event.getIp().equals(second.getIp());
            }
        })
    .within(Time.minutes(10));

// 将模式应用到数据流
PatternStream<LoginEvent> patternStream = CEP.pattern(
    loginEventStream.keyBy(LoginEvent::getUserId),
    bruteForcePattern
);

// 输出告警
DataStream<AlertEvent> cepAlerts = patternStream
    .select((Map<String, List<LoginEvent>> pattern) -> {
        LoginEvent first = pattern.get("first").get(0);
        LoginEvent second = pattern.get("second").get(0);
        LoginEvent third = pattern.get("third").get(0);
        return new AlertEvent(
            "login_brute_force",
            "CEP_BRUTE_FORCE",
            String.format("User %s: 3 failed logins from different IPs in 10 min: %s, %s, %s",
                first.getUserId(), first.getIp(), second.getIp(), third.getIp()),
            System.currentTimeMillis()
        );
    });

4.3.3 评估要点

• 模式复杂度：CEP规则的状态数随模式长度指数增长，复杂模式可能导致状态膨胀和背压。
• 超时处理：必须为模式设置 .within() 时间约束，并定义超时后的处理逻辑（如丢弃或部分匹配告警）。
• 事件时间语义：CEP强烈依赖事件时间，水位线设置不当会导致模式匹配延迟或遗漏。
• 测试覆盖：需要构造正例（触发告警）和反例（不触发告警）的测试数据，验证模式逻辑的正确性。

4.4 三种规则模型对比总结

在实际预警系统中，这三种规则模型通常组合使用：阈值规则作为第一道防线快速响应，滑动窗口规则用于检测持续异常趋势，CEP规则处理需要多事件关联的复杂场景。评估时需根据业务对延迟、准确率和复杂度的要求，选择合适的规则模型或混合策略。