使用[KafkaStreams流计算框架实时计算产生报警（升级报警）

使用KafkaStream（Apache Kafka）实时计算报警，官方文档非常完善。

对Kafka不太了解的，可以看下我的博客Kafka集群部署和调优实践_offsets.topic.replication.factor-CSDN博客

需求背景很简单，每秒钟采集一次设备数据，流计算框架需要对数据做处理，判断采集值超过100就产生报警，如果持续5分钟产生高报，持续10分钟产生高高报。流计算服务只负责产出报警到topic，下游服务负责监听topic后续处理。需要注意，当报警被处置后会向接收数据的主题发送处置信号，处置后需要重置这个设备的时间窗口，它对应的报警从新开始计算。每个设备在报警未被处置前只会升级报警，不会重复报警

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.Stores;

import java.time.Duration;
import java.util.Properties;

public class SensorAlarmApp {

    public static void main(String[] args) {
        // 配置 Kafka Streams
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "sensor-alarm-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Double().getClass());

        StreamsBuilder builder = new StreamsBuilder();

        // 从 sensor-readings 主题读取传感器数据
        KStream<String, Double> readings = builder.stream("sensor-readings");
        
        // 从 sensor-reset-topic 主题读取报警处置信号
        KStream<String, Double> resetStream = builder.stream("sensor-reset-topic");
        
        // 合并传感器数据流和重置信号流
        KStream<String, Double> filteredReadings = readings
            .merge(resetStream) // 合并数据流
            .filter((k, v) -> true); // 可以添加更多的过滤逻辑，如果需要的话

        // 使用 SessionWindows 来处理数据流的窗口
        SessionWindows sessionWindows = SessionWindows.with(Duration.ofMinutes(10)).grace(Duration.ofMinutes(5));

        // 将数据流转换为 KTable，并使用 SessionWindows 计算报警次数
        KTable<SessionWindowed<String>, Long> alarmCounts = filteredReadings
            .filter((k, v) -> v > 100) // 只处理值大于100的记录
            .groupBy((k, v) -> k) // 按照传感器ID分组
            .windowedBy(sessionWindows) // 使用 SessionWindows 窗口
            .count(Materialized.<String, Long, SessionStore<Bytes, byte[]>>as("alarm-count-store").withValueSerde(Serdes.Long()));

        // 创建一个状态存储，用于跟踪报警状态
        final String alarmStateStoreName = "alarm-state-store";
        final KeyValueStore<String, AlarmStatus> alarmStateStore = builder
            .store(
                Stores.keyValueStoreBuilder(
                    Stores.persistentKeyValueStore(alarmStateStoreName),
                    Serdes.String(),
                    AlarmStatus.serde()
                ).withCachingEnabled()
            );

        // 处理报警
        KTable<Windowed<String>, String> lowAlarms = alarmCounts
            .toStream()
            .filter((k, v) -> v == 1) // 第一次超过100
            .filter((k, v) -> shouldTriggerAlarm(k.key(), "low", alarmStateStore))
            .mapValues((k, v) -> updateAlarmStatus(k.key(), "low", "ALARM: Sensor value over 100", alarmStateStore));

        // 处理高报
        KTable<Windowed<String>, String> highAlarms = alarmCounts
            .toStream()
            .filter((k, v) -> k.window().end() - k.window().start() >= Duration.ofMinutes(5).toMillis()) // 窗口持续时间 >= 5分钟
            .filter((k, v) -> shouldTriggerAlarm(k.key(), "high", alarmStateStore))
            .mapValues((k, v) -> updateAlarmStatus(k.key(), "high", "HIGH ALARM: Sensor value over 100 for more than 5 minutes", alarmStateStore));

        // 处理高高报
        KTable<Windowed<String>, String> highHighAlarms = alarmCounts
            .toStream()
            .filter((k, v) -> k.window().end() - k.window().start() >= Duration.ofMinutes(10).toMillis()) // 窗口持续时间 >= 10分钟
            .filter((k, v) -> shouldTriggerAlarm(k.key(), "high-high", alarmStateStore))
            .mapValues((k, v) -> updateAlarmStatus(k.key(), "high-high", "HIGH HIGH ALARM: Sensor value over 100 for more than 10 minutes", alarmStateStore));

        // 处置报警
        filteredReadings.foreach((k, v) -> handleAlarmDisposal(k, v, alarmStateStore, filteredReadings));

        // 输出报警通知
        lowAlarms.toStream().to("low-alarm-notifications", Produced.with(Serdes.String(), Serdes.String()));
        highAlarms.toStream().to("high-alarm-notifications", Produced.with(Serdes.String(), Serdes.String()));
        highHighAlarms.toStream().to("high-high-alarm-notifications", Produced.with(Serdes.String(), Serdes.String()));

        // 启动 Kafka Streams 实例
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }

    // 更新报警状态的方法
    private static String updateAlarmStatus(String sensorId, String alarmType, String message, KeyValueStore<String, AlarmStatus> store) {
        AlarmStatus status = store.get(sensorId);
        if (status == null) {
            status = new AlarmStatus(); // 创建新的报警状态
        }
        status.setAlarmType(alarmType);
        status.setAlarmMessage(message);
        status.setLastUpdated(System.currentTimeMillis());
        store.put(sensorId, status); // 保存报警状态
        return message;
    }

    // 决定是否触发报警的方法
    private static boolean shouldTriggerAlarm(String sensorId, String alarmType, KeyValueStore<String, AlarmStatus> store) {
        AlarmStatus status = store.get(sensorId);
        if (status == null) {
            return true; // 初始状态，可以触发报警
        } else {
            if (alarmType.equals(status.getAlarmType())) {
                return false; // 报警类型相同，不触发
            }
            if ("low".equals(status.getAlarmType()) && "high".equals(alarmType)) {
                return true; // 升级到高报
            }
            if ("high".equals(status.getAlarmType()) && "high-high".equals(alarmType)) {
                return true; // 升级到高高报
            }
            return false; // 其他情况不触发
        }
    }

    // 处置报警的方法
    private static void handleAlarmDisposal(String sensorId, Double value, KeyValueStore<String, AlarmStatus> store, KStream<String, Double> readings) {
        if (value < 100) {
            store.remove(sensorId); // 清除报警状态
            // 发送设备的重置信号到 sensor-reset-topic
            readings.filter((k, v) -> k.equals(sensorId))
                    .to("sensor-reset-topic", Produced.with(Serdes.String(), Serdes.Double()));
        }
    }

    // 报警状态类
    static class AlarmStatus {
        private String alarmType;
        private String alarmMessage;
        private long lastUpdated;

        public String getAlarmType() {
            return alarmType;
        }

        public void setAlarmType(String alarmType) {
            this.alarmType = alarmType;
        }

        public String getAlarmMessage() {
            return alarmMessage;
        }

        public void setAlarmMessage(String alarmMessage) {
            this.alarmMessage = alarmMessage;
        }

        public long getLastUpdated() {
            return lastUpdated;
        }

        public void setLastUpdated(long lastUpdated) {
            this.lastUpdated = lastUpdated;
        }

        public static Serde<AlarmStatus> serde() {
            return Serdes.serdeFrom(new JsonSerializer<>(), new JsonDeserializer<>(AlarmStatus.class));
        }
    }
}