java

10.2 实时告警规则引擎

eventd v3 实时告警规则引擎实现：基于 Flink KeyedProcessFunction 的阈值判断、滑动窗口计数与告警抑制（去抖），涵盖告警级别分级、规则热更新、Redis 持久化与 SSE 推送，并讨论规则误报、重复告警和状态膨胀的取舍。

Leeting Yan 5 分钟阅读 2082 字

为什么需要告警引擎

实时面板展示数据，但没有人 24 小时盯着面板。告警引擎自动检测异常并通知运维人员。

eventd v3 的告警规则：

规则条件级别
错误率过高错误率 > 5% 持续 2 分钟P1（紧急）
延迟过高P99 延迟 > 3sP2（高）
QPS 突降QPS < 预期值的 50%P2（高）
登录失败同一用户 5 分钟内失败 3 次P3（中）
大额异常单笔订单 > 10 万元P3（中）

告警引擎架构

Flink DataStream 指标流
        ↓
    AlertRuleFunction (KeyedProcessFunction)
        ├── 阈值判断
        ├── 滑动窗口计数
        ├── 告警抑制（去抖）
        └── 告警级别计算
        ↓
    告警事件写入 Redis
        ↓
    eventd-panel SSE 推送
        ↓
    运维人员收到通知

规则模型

public record AlertRule(
    String ruleId,           // 规则唯一标识
    String metricKey,        // 监控的指标：qps/error-rate/latency-p99
    RuleCondition condition, // 触发条件
    Duration duration,       // 持续时间（滑动窗口）
    AlertLevel level,        // P1/P2/P3
    String message           // 告警消息模板
) {}

public record RuleCondition(
    String operator,  // ">", "<", "==", "between"
    double threshold, // 阈值；between 时是下界
    double upper      // 仅 between 使用，上界（闭区间）
) {
    public RuleCondition(String operator, double threshold) {
        this(operator, threshold, threshold);
    }

    public RuleCondition {
        if ("between".equals(operator) && upper < threshold) {
            throw new IllegalArgumentException("between 的上界不能小于下界");
        }
    }
}

public enum AlertLevel { P1, P2, P3 }

// 告警流消费的是「指标名 + 当前值 + 时间」这个标量视图。
// 9.3 的 MetricRecord 是 (metric, dimension, dimensionValue, windowStart,
// windowEnd, payload)，字段完全不同，不要混用同一个类名；
// 从窗口结果到 AlertMetric 的转换由上游 map 完成
public record AlertMetric(
    String key,        // 指标名 + 维度，如 "error-rate:global"
    double value,      // 该窗口算出的值
    long timestamp     // 对应窗口的结束时间
) {}

AlertRuleFunction

public class AlertRuleFunction
    extends KeyedProcessFunction<String, AlertMetric, AlertEvent> {

    private final AlertRule rule;
    private transient ValueState<Boolean> alertActive;
    private transient ValueState<Long> alertStartTime;
    private transient ValueState<Integer> consecutiveViolations;

    public AlertRuleFunction(AlertRule rule) {
        this.rule = rule;
    }

    @Override
    public void open(OpenContext parameters) throws Exception {
        alertActive = getRuntimeContext().getState(
            new ValueStateDescriptor<>("alertActive", Types.BOOLEAN));
        alertStartTime = getRuntimeContext().getState(
            new ValueStateDescriptor<>("alertStartTime", Types.LONG));
        consecutiveViolations = getRuntimeContext().getState(
            new ValueStateDescriptor<>("consecutiveViolations", Types.INT));
    }

    @Override
    public void processElement(AlertMetric metric, Context ctx, Collector<AlertEvent> out)
            throws Exception {
        boolean violated = checkCondition(metric, rule.condition());
        Integer violations = consecutiveViolations.value();
        if (violations == null) violations = 0;

        if (violated) {
            // 第一次违反时记录起点，而不是到触发条件时才记
            if (violations == 0) {
                alertStartTime.update(metric.timestamp());
            }
            int next = violations + 1;
            consecutiveViolations.update(next);

            // 需要连续 N 次违反才触发告警。
            // 这里假设上游每分钟产出一条 AlertMetric（窗口是 1 分钟），
            // 所以 N = 持续时间 / 60；换成更快的采样率，这个换算就不成立，
            // 必须改成按时间比较 alertStartTime，而不是数条数
            int requiredCount = Math.max(1, (int) (rule.duration().getSeconds() / 60));
            if (next >= requiredCount && !Boolean.TRUE.equals(alertActive.value())) {
                alertActive.update(true);

                out.collect(new AlertEvent(
                    rule.ruleId(),
                    rule.level(),
                    rule.message().replace("${value}", String.valueOf(metric.value())),
                    alertStartTime.value(),
                    metric.key()
                ));
            }
        } else {
            // 恢复正常
            consecutiveViolations.update(0);
            if (alertActive.value() != null) {
                alertActive.clear();
                alertStartTime.clear();
                // 发送恢复通知
                out.collect(new AlertEvent(
                    rule.ruleId(),
                    // 恢复通知沿用原规则的级别，不能固定成 P3：
                    // P1 的恢复如果按 P3 路由，会被当成低优先级消息
                    rule.level(),
                    "✅ 恢复正常：" + rule.metricKey(),
                    metric.timestamp(),
                    metric.key()
                ));
            }
        }
    }

    private boolean checkCondition(AlertMetric metric, RuleCondition condition) {
        double value = metric.value();
        return switch (condition.operator()) {
            case ">" -> value > condition.threshold();
            case "<" -> value < condition.threshold();
            case "==" -> Math.abs(value - condition.threshold()) < 0.0001;
            case "between" -> value >= condition.threshold() && value <= condition.upper();
            default -> false;
        };
    }
}

public record AlertEvent(
    String ruleId,
    AlertLevel level,
    String message,
    long timestamp,
    String metricKey
) {}

CEP 告警模式

特定场景用 CEP 更直观：

// 规则：同一用户 5 分钟内登录失败 3 次 → P3 告警
// LoginEvent 沿用 6.1 的示例 record（访问器 userId()/status()）
Pattern<LoginEvent, ?> loginFailPattern = Pattern.<LoginEvent>begin("fail1")
    .where(e -> "FAIL".equals(e.status()))
    .next("fail2")
    .where(e -> "FAIL".equals(e.status()))
    .next("fail3")
    .where(e -> "FAIL".equals(e.status()))
    .within(Duration.ofMinutes(5));

DataStream<AlertEvent> loginAlerts = CEP.pattern(
        loginStream.keyBy(LoginEvent::userId),
        loginFailPattern
    )
    .process(new PatternProcessFunction<LoginEvent, AlertEvent>() {
        @Override
        public void processMatch(Map<String, List<LoginEvent>> match, Context ctx,
                                 Collector<AlertEvent> out) {
            LoginEvent first = match.get("fail1").get(0);
            out.collect(new AlertEvent(
                "RULE_LOGIN_FAIL",
                AlertLevel.P3,
                "用户 " + first.userId() + " 5 分钟内登录失败 3 次",
                ctx.timestamp(),
                first.userId()
            ));
        }
    });

CEP 的 within() 走的是事件时间：上游必须先用 assignTimestampsAndWatermarks 分配时间戳，否则模式会按到达顺序的默认时间戳匹配，迟到事件也会被漏掉。另外这条告警的时间戳取 ctx.timestamp()（匹配时刻的事件时间），不要用 System.currentTimeMillis()——后者会让重放结果不可复现。

告警抑制（去抖）

eventd v3 的告警抑制策略：

级别最小间隔升级策略
P15 分钟连续 3 次未恢复 → 升级为电话通知（级别仍是 P1，只是换通道）
P215 分钟持续 30 分钟 → 升级 P1
P31 小时不升级
public class AlertSuppressor
    extends KeyedProcessFunction<String, AlertEvent, AlertEvent> {

    private static final Duration P1_COOLDOWN = Duration.ofMinutes(5);
    private static final Duration P2_COOLDOWN = Duration.ofMinutes(15);
    private static final Duration P3_COOLDOWN = Duration.ofHours(1);

    private transient ValueState<Long> lastAlertTime;

    @Override
    public void open(OpenContext parameters) throws Exception {
        lastAlertTime = getRuntimeContext().getState(
            new ValueStateDescriptor<>("lastAlert", Types.LONG));
    }

    @Override
    public void processElement(AlertEvent alert, Context ctx, Collector<AlertEvent> out)
            throws Exception {
        long cooldown = getCooldown(alert.level());
        long now = ctx.timerService().currentProcessingTime();
        Long last = lastAlertTime.value();

        if (last == null || now - last >= cooldown.toMillis()) {
            lastAlertTime.update(now);
            out.collect(alert);
        }
        // 否则丢弃（抑制）
    }

    private long getCooldown(AlertLevel level) {
        return switch (level) {
            case P1 -> P1_COOLDOWN.toMillis();
            case P2 -> P2_COOLDOWN.toMillis();
            case P3 -> P3_COOLDOWN.toMillis();
        };
    }
}

Redis 存储与推送

// open/close 来自 RichFunction，必须继承 RichSinkFunction；
// 只 implements SinkFunction 时 open 无处可覆盖
public class RedisAlertSink extends RichSinkFunction<AlertEvent> {
    private transient RedisClient client;
    private transient RedisCommands<String, String> sync;

    @Override
    public void open(OpenContext openContext) {
        client = RedisClient.create("redis://localhost:6379");
        sync = client.connect().sync();
    }

    @Override
    public void close() {
        if (client != null) {
            client.shutdown();
        }
    }

    @Override
    public void invoke(AlertEvent alert, Context context) {
        // 1. 写入活跃告警列表（TTL 1 小时）
        String activeKey = "eventd:alerts:active";
        String alertJson = JsonMapper.toJson(alert);
        sync.lpush(activeKey, alertJson);
        sync.expire(activeKey, 3600);

        // 2. 写入级别索引
        String levelKey = "eventd:alerts:" + alert.level().name().toLowerCase();
        sync.lpush(levelKey, alertJson);
        sync.expire(levelKey, 3600);

        // 3. 写入时间序列
        String tsKey = "eventd:alerts:ts";
        sync.zadd(tsKey, alert.timestamp(), alertJson);
        sync.expire(tsKey, 86400);

        // 4. 发布到 channel（SSE 用）
        sync.publish("eventd:alerts", alertJson);
    }
}

这段写入在故障重放下会重复：LPUSH 每次都会新增一个元素，Checkpoint 回滚后同一条告警会进列表两次。要幂等，把告警 id 或 (ruleId, windowStart) 作为去重键先 SADD，返回 0 就跳过后续写入；或者直接用 ZADD key score member，member 用告警 id，天然覆盖。另外 localhost:6379 同样只适用于本机调试，容器内要用 compose 的服务名（见 12.2 部署与运维 ）。

面板告警展示

@RestController
public class AlertController {
    private final ReactiveRedisTemplate<String, String> redisTemplate;

    @GetMapping(value = "/api/stream/alerts", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> streamAlerts() {
        // listenTo 内部负责 SUBSCRIBE 与取消时退订，直接给出消息流
        return redisTemplate
            .listenTo(ChannelTopic.of("eventd:alerts"))
            .map(msg -> ServerSentEvent.builder(msg.getMessage())
                .event("alert")
                .build());
    }

    @GetMapping("/api/alerts/active")
    public Flux<AlertEvent> getActiveAlerts() {
        return redisTemplate.opsForList()
            .range("eventd:alerts:active", 0, 99)
            .map(json -> JsonMapper.parse(json, AlertEvent.class));
    }
}

如果直接拿 Lettuce 写：connectPubSub().reactive().subscribe(...) 返回的是 Mono<Void>，它只发起订阅、不携带消息；消息流要从 observeChannels(...) 拿，且仍要再调一次 subscribe(...) 才真正生效，退订与连接释放也得自己管。Spring Data Redis 的 listenTo(Topic...) 把这些步骤封在一起，面板这类场景优先用它。

小结

组件职责
AlertRuleFunction阈值判断、滑动窗口计数
CEP Pattern复杂事件模式触发
AlertSuppressor告警抑制、去抖
RedisAlertSink存储、索引、Pub/Sub
AlertControllerSSE 推送

下一节 10.3 实时面板数据流 ：SSE 实时推送与背压适配。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南