fix: 缓存队列

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/core/event/SensorAlarmEvent.java

@@ -0,0 +1,22 @@
+package vip.xiaonuo.coldchain.core.event;
+
+import org.springframework.context.ApplicationEvent;
+
+import java.time.Clock;
+
+/**
+ * @author jackzhou
+ * @version 1.0
+ * @project jfcloud-coldchain
+ * @description
+ * @date 2024/11/25 16:55:56
+ */
+public class SensorAlarmEvent extends ApplicationEvent {
+    public SensorAlarmEvent(Object source) {
+        super(source);
+    }
+
+    public SensorAlarmEvent(Object source, Clock clock) {
+        super(source, clock);
+    }
+}

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/core/event/SensorAlarmEventListener.java

@@ -0,0 +1,26 @@
+package vip.xiaonuo.coldchain.core.event;
+
+/**
+ * @author jackzhou
+ * @version 1.0
+ * @project jfcloud-coldchain
+ * @description
+ * @date 2024/11/12 15:31:43
+ */
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@RequiredArgsConstructor
+@Component
+public class SensorAlarmEventListener {
+
+    @Async
+    @EventListener
+    public void handleSensorAlarmEvent(SensorAlarmEvent event) {
+    }
+}

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/core/event/SensorDataEventListener.java

@@ -1,36 +1,77 @@
-package vip.xiaonuo.coldchain.core.event;
-
-/**
- * @author jackzhou
- * @version 1.0
- * @project jfcloud-coldchain
- * @description
- * @date 2024/11/12 15:31:43
- */
-
-import com.github.jfcloud.influxdb.service.JfcloudInfluxDBService;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.context.event.EventListener;
-import org.springframework.scheduling.annotation.Async;
-import org.springframework.stereotype.Component;
-import vip.xiaonuo.coldchain.core.bean.influxdb.SensorData;
-
-@Slf4j
-@RequiredArgsConstructor
-@Component
-public class SensorDataEventListener {
-    private final JfcloudInfluxDBService jfcloudInfluxDBService;
-
-    @Async
-    @EventListener
-    public void handleSensorDataEvent(SensorDataEvent event) {
-        SensorData sensorData = event.getSensorData();
-        try {
-            jfcloudInfluxDBService.writePojo(sensorData);
+//package vip.xiaonuo.coldchain.core.event;
+//
+///**
+// * @author jackzhou
+// * @version 1.0
+// * @project jfcloud-coldchain
+// * @description 监听传感器数据事件并异步写入 InfluxDB
+// * @date 2024/11/12 15:31:43
+// */
+//
+//import com.github.jfcloud.influxdb.service.JfcloudInfluxDBService;
+//import lombok.RequiredArgsConstructor;
+//import lombok.extern.slf4j.Slf4j;
+//import org.springframework.context.ApplicationEventPublisher;
+//import org.springframework.context.event.EventListener;
+//import org.springframework.scheduling.annotation.Async;
+//import org.springframework.stereotype.Component;
+//import vip.xiaonuo.coldchain.core.bean.influxdb.SensorData;
+//
+//@Slf4j
+//@RequiredArgsConstructor
+//@Component
+//public class SensorDataEventListener {
+//    private final JfcloudInfluxDBService jfcloudInfluxDBService;
+//    private final ApplicationEventPublisher eventPublisher;
+//
+//    @Async
+//    @EventListener
+//    public void handleSensorDataEvent(SensorDataEvent event) {
+//        SensorData sensorData = event.getSensorData();
+//        try {
+//            log.info("开始写入数据到 InfluxDB: {}", sensorData);
+//            jfcloudInfluxDBService.writePojo(sensorData);
 //            log.info("成功写入数据到 InfluxDB: {}", sensorData);
-        } catch (Exception e) {
-            log.error("写入数据到 InfluxDB 时出错: {}", e.getMessage());
-        }
-    }
-}
+//
+//            // 数据写入成功后检查是否需要报警
+//            checkForAlarm(sensorData);
+//
+//        } catch (Exception e) {
+//            log.error("写入数据到 InfluxDB 时出错: {}", e.getMessage(), e);
+//            // 可以根据需要选择是否抛出新的异常或进行其他处理
+//        }
+//    }
+//
+//
+//    // 检查是否需要触发报警
+//    private void checkForAlarm(SensorData sensorData) {
+//        // 设置阈值（可以根据实际情况调整这些阈值）
+//        float temperatureThreshold = 50.0f;  // 温度阈值
+//        float humidityThreshold = 90.0f;     // 湿度阈值
+//        float co2Threshold = 1000.0f;        // 二氧化碳阈值
+//        StringBuilder alarmMessage = new StringBuilder();
+//
+//        // 检查温度是否超标
+//        if (sensorData.getTemperature() != null && sensorData.getTemperature() > temperatureThreshold) {
+//            alarmMessage.append("温度超标: ").append(sensorData.getTemperature()).append("°C; ");
+//        }
+//
+//        // 检查湿度是否超标
+//        if (sensorData.getHumidity() != null && sensorData.getHumidity() > humidityThreshold) {
+//            alarmMessage.append("湿度超标: ").append(sensorData.getHumidity()).append("%; ");
+//        }
+//
+//        // 检查二氧化碳是否超标
+//        if (sensorData.getCo2() != null && sensorData.getCo2() > co2Threshold) {
+//            alarmMessage.append("二氧化碳超标: ").append(sensorData.getCo2()).append("ppm; ");
+//        }
+//        // 如果发现超标数据，发布报警事件
+//        if (!alarmMessage.isEmpty()) {
+//            // 构造报警事件
+//            SensorAlarmEvent alarmEvent = new SensorAlarmEvent(this);
+//            log.warn("数据异常，发布报警: {}", alarmMessage.toString());
+//            // 发布报警事件
+//            eventPublisher.publishEvent(alarmEvent);
+//        }
+//    }
+//}

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/core/event/SensorDataEventListener2.java

@@ -0,0 +1,134 @@
+package vip.xiaonuo.coldchain.core.event;
+
+import com.github.jfcloud.influxdb.service.JfcloudInfluxDBService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import vip.xiaonuo.coldchain.core.bean.influxdb.SensorData;
+import vip.xiaonuo.coldchain.core.event.alarm.SensorThreshold;
+
+import java.util.LinkedList;
+import java.util.Queue;
+
+/**
+ * 事件监听器：处理 SensorData 事件，缓存数据并进行定时写入数据库，同时检查是否需要报警
+ */
+@Slf4j
+@RequiredArgsConstructor
+@Component
+public class SensorDataEventListener2 {
+    private final JfcloudInfluxDBService jfcloudInfluxDBService;
+    private final ApplicationEventPublisher eventPublisher;
+    // 使用队列缓存 SensorData
+    private final Queue<SensorData> sensorDataQueue = new LinkedList<>();
+    // 队列最大容量和写入数据库的时间间隔
+    private static final int MAX_QUEUE_SIZE = 100;
+    // 每10秒检查一次
+    private static final long CHECK_INTERVAL_MS = 10000L;
+
+    /**
+     * 监听 SensorDataEvent 事件，收到传感器数据后进行缓存和处理
+     *
+     * @param event 传感器数据事件
+     */
+    @EventListener
+    public void handleSensorDataEvent(SensorDataEvent event) {
+        SensorData sensorData = event.getSensorData();
+
+        // 将数据放入队列
+        addSensorData(sensorData);
+
+        // 在数据放入队列后进行报警检查
+        checkForAlarm(sensorData);
+    }
+
+    /**
+     * 将新的 SensorData 加入队列，并根据队列大小判断是否需要写入数据库
+     *
+     * @param sensorData 传感器数据
+     */
+    public void addSensorData(SensorData sensorData) {
+        synchronized (sensorDataQueue) {
+            sensorDataQueue.offer(sensorData);  // 将数据放入队列
+        }
+
+        // 如果队列达到最大容量，批量写入数据库
+        if (sensorDataQueue.size() >= MAX_QUEUE_SIZE) {
+            writeDataToDatabase();
+        }
+    }
+
+    /**
+     * 定时检查队列并写入数据库
+     * 每隔 CHECK_INTERVAL_MS 时间检查一次
+     */
+    @Scheduled(fixedRate = CHECK_INTERVAL_MS)
+    public void checkAndWriteData() {
+        synchronized (sensorDataQueue) {
+            if (!sensorDataQueue.isEmpty()) {
+                writeDataToDatabase();
+            }
+        }
+    }
+
+    /**
+     * 将队列中的数据批量写入数据库，并清空队列
+     */
+    private void writeDataToDatabase() {
+        synchronized (sensorDataQueue) {
+            if (!sensorDataQueue.isEmpty()) {
+                try {
+                    log.info("开始批量写入数据到 InfluxDB，队列大小: {}", sensorDataQueue.size());
+                    // 批量写入数据
+                    for (SensorData sensorData : sensorDataQueue) {
+                        jfcloudInfluxDBService.writePojo(sensorData);
+                    }
+                    // 写入完成后清空队列
+                    sensorDataQueue.clear();
+                    log.info("成功批量写入数据到 InfluxDB，队列已清空");
+                } catch (Exception e) {
+                    log.error("写入数据到 InfluxDB 时出错: {}", e.getMessage(), e);
+                }
+            }
+        }
+    }
+
+    /**
+     * 检查传感器数据是否满足报警条件
+     * 触发报警事件
+     *
+     * @param sensorData 传感器数据
+     */
+    private void checkForAlarm(SensorData sensorData, SensorThreshold threshold) {
+        // 检查温度是否超标
+        if (sensorData.getTemperature() != null) {
+            if (sensorData.getTemperature() < threshold.getTemperatureMin() || sensorData.getTemperature() > threshold.getTemperatureMax()) {
+                publishAlarm("温度超标", sensorData.getTemperature(), "°C");
+            }
+        }
+
+        // 检查湿度是否超标
+        if (sensorData.getHumidity() != null) {
+            if (sensorData.getHumidity() < threshold.getHumidityMin() || sensorData.getHumidity() > threshold.getHumidityMax()) {
+                publishAlarm("湿度超标", sensorData.getHumidity(), "%");
+            }
+        }
+
+        // 检查二氧化碳是否超标
+        if (sensorData.getCo2() != null) {
+            if (sensorData.getCo2() < threshold.getCo2Min() || sensorData.getCo2() > threshold.getCo2Max()) {
+                publishAlarm("二氧化碳超标", sensorData.getCo2(), "ppm");
+            }
+        }
+    }
+
+    private void publishAlarm(String alarmType, Float value, String unit) {
+        String alarmMessage = String.format("%s: %.2f %s", alarmType, value, unit);
+        log.warn("数据异常，发布报警: {}", alarmMessage);
+        // 发布报警事件（假设 AlarmEvent 已经定义）
+        eventPublisher.publishEvent(new SensorAlarmEvent(this/*, alarmMessage*/));
+    }
+}

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/core/event/alarm/SensorThreshold.java

@@ -0,0 +1,20 @@
+package vip.xiaonuo.coldchain.core.event.alarm;
+
+import lombok.Data;
+
+/**
+ * @author jackzhou
+ * @version 1.0
+ * @project jfcloud-coldchain
+ * @description
+ * @date 2024/11/25 17:08:04
+ */
+@Data
+public class SensorThreshold {
+    private float temperatureMin;
+    private float temperatureMax;
+    private float humidityMin;
+    private float humidityMax;
+    private float co2Min;
+    private float co2Max;
+}

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/core/event/alarm/SensorThresholdService.java

@@ -0,0 +1,35 @@
+package vip.xiaonuo.coldchain.core.event.alarm;
+
+/**
+ * @author jackzhou
+ * @version 1.0
+ * @project jfcloud-coldchain
+ * @description
+ * @date 2024/11/25 17:13:50
+ */
+package vip.xiaonuo.coldchain.core.service.impl;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import vip.xiaonuo.coldchain.core.bean.SensorThreshold;
+import vip.xiaonuo.coldchain.core.service.ThresholdService;
+import vip.xiaonuo.coldchain.core.repository.SensorThresholdRepository;
+
+/**
+ * 从数据库获取传感器阈值的实现类
+ */
+@Service
+public class SensorThresholdService implements ThresholdService {
+//    @Override
+//    public SensorThreshold getThresholdBySensorId(Long sensorId) {
+////        // 查询数据库获取对应传感器的阈值
+////        return sensorThresholdRepository.findBySensorId(sensorId)
+////                .orElseThrow(() -> new IllegalArgumentException("传感器阈值未找到，sensorId: " + sensorId));
+//        return null;
+//    }
+
+    @Override
+    public SensorThreshold getThresholdBySensorId(String sensorCode, Integer sensorNo) {
+        return null;
+    }
+}

snowy-plugin/snowy-plugin-coldchain/src/main/java/vip/xiaonuo/coldchain/core/event/alarm/ThresholdService.java

@@ -0,0 +1,23 @@
+package vip.xiaonuo.coldchain.core.event.alarm;
+
+/**
+ * @author jackzhou
+ * @version 1.0
+ * @project jfcloud-coldchain
+ * @description
+ * @date 2024/11/25 17:08:36
+ */
+
+/**
+ * 获取传感器阈值范围的服务接口
+ */
+public interface ThresholdService {
+
+    /**
+     * 根据传感器 ID 获取对应的阈值范围
+     * @param sensorCode 传感器 ID
+     * @param sensorNo 传感器 路数
+     * @return 传感器的阈值范围
+     */
+    SensorThreshold getThresholdBySensorId(String sensorCode,Integer sensorNo);
+}

snowy-web-app/src/main/java/vip/xiaonuo/Application.java

@@ -20,6 +20,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.core.env.Environment;
 import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
 
@@ -33,6 +34,7 @@ import org.springframework.web.bind.annotation.RestController;
 @RestController
 @SpringBootApplication(scanBasePackages = {"vip.xiaonuo","com.github.jfcloud"})
 @EnableAsync
+@EnableScheduling
 public class Application {
 
     /* 解决druid 日志报错：discard long time none received connection:xxx */