温度上下限告警处理

agri-framework/src/main/java/com/agri/framework/interceptor/DeviceStatusHandler.java

@@ -16,6 +16,7 @@ import org.apache.commons.collections4.CollectionUtils;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.StringRedisTemplate;
@@ -23,10 +24,8 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 import java.math.BigDecimal;
-import java.util.ArrayList;
+import java.time.LocalDateTime;
-import java.util.HashSet;
+import java.util.*;
-import java.util.List;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -77,6 +76,9 @@ public class DeviceStatusHandler {
 
     private static final String AUTO_MODE = "auto_mode";
     private static final String AUTO_OFF = "autooff";
+
+    @Resource
+    private FrontendControlHandler frontendControlHandler;
     /**
      * 自动关任务管理器，调度/取消自动关任务
      */
@@ -169,17 +171,20 @@ public class DeviceStatusHandler {
             // 仅当up且所有key为数字时，才更新最新状态缓存
             for (String key : payloadObj.keySet()) {
                 BigDecimal value = payloadObj.getBigDecimal(key);
+                String valueIndex = key.substring(2);
+                SysAgriInfo sysAgriInfo = new SysAgriInfo();
+                BeanUtils.copyProperties(agriInfo, sysAgriInfo);
                 // 温度
                 if (key.startsWith("20")) {
                     // 温度大于温度上限
                     if (value.compareTo(tempUp) > 0 ) {
                         agriInfo.setTitle("温度异常");
-                        agriInfo.setMsg("温度"+key.substring(2)+ "异常！高于上限"+tempUp+"℃！");
+                        agriInfo.setMsg("温度"+valueIndex+ "异常！高于上限"+tempUp+"℃！");
                     }
                     // 温度小于温度下限
                     if (value.compareTo(tempLow) < 0) {
                         agriInfo.setTitle("温度异常");
-                        agriInfo.setMsg("温度"+key.substring(2)+ "异常！低于下限"+tempLow+"℃！");
+                        agriInfo.setMsg("温度"+valueIndex+ "异常！低于下限"+tempLow+"℃！");
                     }
                     msgList.add(agriInfo);
                 }
@@ -188,18 +193,24 @@ public class DeviceStatusHandler {
                     // 湿度大于湿度上限
                     if (value.compareTo(humiUp) > 0 ) {
                         agriInfo.setTitle("湿度异常");
-                        agriInfo.setMsg("湿度"+key.substring(2)+ "异常！高于上限"+humiUp+"RH%！");
+                        agriInfo.setMsg("湿度"+valueIndex+ "异常！高于上限"+humiUp+"RH%！");
                     }
                     // 湿度小于湿度下限
                     if (value.compareTo(humiLow) < 0) {
                         agriInfo.setTitle("湿度异常");
-                        agriInfo.setMsg("湿度"+key.substring(2)+ "异常！低于下限"+humiLow+"RH%！");
+                        agriInfo.setMsg("湿度"+valueIndex+ "异常！低于下限"+humiLow+"RH%！");
                     }
                     msgList.add(agriInfo);
                 }
             }
-
+            if (!msgList.isEmpty()) {
-            agriService.saveMessage(msgList);
+                List<SysMessage> messages = agriService.saveMessage(msgList);
+                try {
+                    frontendControlHandler.sendAlarmMessage(messages);
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }
         }
     }
 

agri-framework/src/main/java/com/agri/framework/interceptor/FrontendControlHandler.java

@@ -7,6 +7,7 @@ import com.agri.framework.manager.MqttAutoOffManager;
 import com.agri.system.domain.SysAgriInfo;
 import com.agri.system.domain.SysAgriLimit;
 import com.agri.system.domain.SysDevOperLog;
+import com.agri.system.domain.SysMessage;
 import com.agri.system.mapper.SysUserMapper;
 import com.agri.system.service.ISysAgriInfoService;
 import com.agri.system.service.ISysAgriLimitService;
@@ -16,6 +17,7 @@ import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONObject;
 import com.alibaba.fastjson2.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.ObjectUtils;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.slf4j.Logger;
@@ -28,7 +30,9 @@ import org.springframework.util.StringUtils;
 
 import javax.annotation.Resource;
 import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -71,6 +75,7 @@ public class FrontendControlHandler {
     private ISysAgriInfoService sysAgriInfoService;
     @Autowired
     private ISysDevOperLogService sysDevOperLogService;
+    private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
 
     @Autowired
     private ISysUserService sysUserService;
@@ -258,14 +263,16 @@ public class FrontendControlHandler {
         return Boolean.TRUE;
     }
 
-    public void sendAlarmMessage(String msg, String imei, String timeStr) throws Exception {
+    public void sendAlarmMessage(List<SysMessage> messages) throws Exception {
-
+        if (CollectionUtils.isEmpty(messages)) {
-        Map<String, Object> alarmMsg = new HashMap<>();
+            return;
-        alarmMsg.put("online", msg);
+        }
-        alarmMsg.put("time", timeStr);
+        for (SysMessage message : messages) {
-        alarmMsg.put("imei", imei);
+            Map<String, Object> alarmMsg = new HashMap<>();
-        String alarmMessage = objectMapper.writeValueAsString(alarmMsg);
+            alarmMsg.put("msg",message.getContent());
-        mqttMessageSender.publish("frontend/" + imei + "/alarm", alarmMessage);
+            alarmMsg.put("time", LocalDateTime.now().format(DATE_TIME_FORMATTER));
-
+            String alarmMessage = objectMapper.writeValueAsString(alarmMsg);
+            mqttMessageSender.publish("frontend/" + message.getImei() + "/alarm", alarmMessage);
+        }
     }
 }

agri-framework/src/main/java/com/agri/framework/manager/AgriStatusManager.java

@@ -132,10 +132,6 @@ public class AgriStatusManager {
                     String onlineMessage = objectMapper.writeValueAsString(onlineMsg);
                     mqttMessageSender.publish("device/" + imei + "/status", onlineMessage);
                 }
-                // 无论设备是否在线 只要离线就推送设备状态
-                if (!imeiMap.get("imeiOnline")) {
-                    frontendControlHandler.sendAlarmMessage("设备离线", imei, dateNow);
-                }
                 successCount++;
             } catch (Exception e) {
                 failCount++;

agri-quartz/src/main/java/com/agri/quartz/task/AgriStatusTask.java

@@ -1,7 +1,9 @@
 package com.agri.quartz.task;
 
+import com.agri.framework.interceptor.FrontendControlHandler;
 import com.agri.framework.manager.AgriStatusManager;
 import com.agri.system.domain.SysAgriInfo;
+import com.agri.system.domain.SysMessage;
 import com.agri.system.service.AgriService;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.agri.framework.config.MqttConfig;
@@ -52,6 +54,9 @@ public class AgriStatusTask {
     @Autowired
     private AgriService agriService;
 
+    @Resource
+    private FrontendControlHandler frontendControlHandler;
+
     /**
      * 定时任务：每10秒执行一次
      * 1. 扫描所有 sub: 开头的 key，提取 IMEI
@@ -84,12 +89,14 @@ public class AgriStatusTask {
             }
             Map<String, Map<String, Boolean>> statusMap
                     = agriStatusManager.batchCheckDeviceOnline(imeiList);
-            // 3. 批量查询设备在线状态（Redis Pipeline，一次网络往返）
+            // 3. 首页状态
             agriStatusManager.asyncBatchPushMqtt(statusMap);
-            // 4. 保存离线设备
+            // 4. 离线设备
             List<SysAgriInfo> offlineDevices = findOfflineDevices(agriInfos, statusMap);
             // 5. 保存消息中心
-            agriService.saveMessage(offlineDevices);
+            List<SysMessage> messages = agriService.saveMessage(offlineDevices);
+            // 6. 发送告警消息
+            frontendControlHandler.sendAlarmMessage(messages);
 
         } catch (Exception e) {
             log.error("设备在线状态推送任务异常", e);
@@ -107,9 +114,15 @@ public class AgriStatusTask {
             log.info("不存在任何imei");
             return new ArrayList<>();
         }
+        Map<String, Object> offlineMap = new HashMap<>();
+        for (Map.Entry<String, Map<String, Boolean>> map : statusMap.entrySet()) {
+            if (!map.getValue().get("imeiOnline")) {
+                offlineMap.put(map.getKey(), map.getValue());
+            }
+        }
         List<SysAgriInfo> offlineDevices = new ArrayList<>();
         for (SysAgriInfo agriInfo : agriInfos) {
-            if (!statusMap.containsKey(agriInfo.getImei())) {
+            if (!offlineMap.containsKey(agriInfo.getImei())) {
                 agriInfo.setTitle("设备离线告警");
                 agriInfo.setMsg("怀疑设备离线！请及时检查");
                 offlineDevices.add(agriInfo);

agri-quartz/src/main/java/com/agri/quartz/task/AgriTempTask.java

@@ -1,6 +1,7 @@
 package com.agri.quartz.task;
 
 import com.agri.framework.interceptor.FrontendControlHandler;
+import com.agri.system.domain.SysMessage;
 import com.agri.system.service.AgriService;
 import com.agri.system.domain.SysAgriInfo;
 import com.agri.system.service.ISysAgriInfoService;
@@ -57,8 +58,9 @@ public class AgriTempTask {
             }
             Map<String, Object> latestDataMap = queryLatestDtuData(imeiList);
             List<SysAgriInfo> offlineDevices = findOfflineDevices(agriInfos, latestDataMap);
-            pushOfflineAlarm(offlineDevices);
+            List<SysMessage> messages = agriService.saveMessage(offlineDevices);
-            agriService.saveMessage(offlineDevices);
+            // 推送离线告警
+            frontendControlHandler.sendAlarmMessage(messages);
         } catch (Exception e) {
             log.error("设备在线状态推送任务异常", e);
         } finally {
@@ -117,20 +119,4 @@ public class AgriTempTask {
         return offlineList;
     }
 
-    // 推送离线告警
-    private void pushOfflineAlarm(List<SysAgriInfo> offlineList) {
-        if (CollectionUtils.isEmpty(offlineList)) {
-            return;
-        }
-        String timeStr = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
-        for (SysAgriInfo agriInfo : offlineList) {
-            try {
-                frontendControlHandler.sendAlarmMessage("温度离线",agriInfo.getImei(), timeStr);
-            } catch (Exception e) {
-                log.error("发送设备离线告警失败, imei={}", agriInfo.getImei(), e);
-            }
-        }
-    }
-
-
 }

agri-system/src/main/java/com/agri/system/domain/SysMessage.java

@@ -25,6 +25,7 @@ public class SysMessage extends BaseEntity
     /** 消息主键ID */
     private Long id;
 
+    private String imei;
     /** 接收人：all=全体用户，其他=用户ID */
     @Excel(name = "接收人：all=全体用户，其他=用户ID")
     private Long receiver;

agri-system/src/main/java/com/agri/system/service/AgriService.java

@@ -1,13 +1,14 @@
 package com.agri.system.service;
 
 import com.agri.system.domain.SysAgriInfo;
+import com.agri.system.domain.SysMessage;
 
 import java.util.List;
 import java.util.Map;
 
 public interface AgriService {
 
-    void saveMessage(List<SysAgriInfo> offlineList);
+    List<SysMessage> saveMessage(List<SysAgriInfo> offlineList);
 
     List<String> queryAllGreenhouseImei(List<SysAgriInfo> agriInfos);
 

agri-system/src/main/java/com/agri/system/service/impl/AgriServiceImpl.java

@@ -27,10 +27,10 @@ public class AgriServiceImpl implements AgriService {
 
     // 保存离线设备消息
     @Override
-    public void saveMessage(List<SysAgriInfo> offlineList) {
+    public List<SysMessage> saveMessage(List<SysAgriInfo> offlineList) {
         // 离线是否为空
         if (CollectionUtils.isEmpty(offlineList)) {
-            return;
+            return Collections.emptyList();
         }
         List<Long> idList
                 = offlineList.stream().map(SysAgriInfo::getId).collect(Collectors.toList());
@@ -41,10 +41,11 @@ public class AgriServiceImpl implements AgriService {
         List<SysMessage> msgList = new ArrayList<>();
         for (SysAgriInfo agriInfo : offlineList) {
             SysMessage message = new SysMessage();
+            message.setImei(agriInfo.getImei());
             message.setTitle(agriInfo.getTitle());
             message.setMsgType("status");
             message.setReadStatus(0L);
-            message.setContent("大棚【" + agriInfo.getAgriName() + "】" + agriInfo.getMsg());
+            message.setContent("大棚【" + agriInfo.getAgriName() + " - " + agriInfo.getImei() + "】" + agriInfo.getMsg());
             message.setImgUrl("");
             message.setLinkUrl(UrlEncodeUtil.buildControlPageUrl(agriInfo, "/pages/home/control/index?agriInfo="));
             for (SysUserAgri userAgri : agriUser){
@@ -55,6 +56,7 @@ public class AgriServiceImpl implements AgriService {
         if (CollectionUtils.isNotEmpty(msgList)) {
             messageService.saveBatch(msgList);
         }
+        return msgList;
 
     }