使用[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));
}
}
}