modbus读取本地设备数据增加告警逻辑

ems-common/src/main/java/com/xzzn/common/constant/RedisKeyConstants.java

@@ -96,6 +96,10 @@ public class RedisKeyConstants
     public static final String TOPIC_FAILED_ALRAM_RECORD = "topic_failed_";
     /**  topic 内没有数据设备维度告警 */
     public static final String TOPIC_EMPTY_ALARM_RECORD = "topic_empty_";
+    /**  modbus读取 没有数据设备维度告警 */
+    public static final String MODBUS_EMPTY_ALARM_RECORD = "modbus_empty_";
+    /**  modbus读取 设备离线告警 */
+    public static final String MODBUS_OFFLINE_ALARM_RECORD = "modbus_offline_";
 
     /** 设备信息初始化 */
     public static final String INIT_DEVICE_INFO = "init_device_info";

ems-common/src/main/java/com/xzzn/common/core/modbus/ModbusProcessor.java

@@ -170,14 +170,20 @@ public class ModbusProcessor {
         }
     }
 
+    public ModbusMaster borrowMaster(DeviceConfig config) throws Exception {
+        ModbusMaster master = connectionManager.borrowMaster(config);
+        // 设置了Modbus通信的超时时间为3000毫秒（3秒）。当主设备与从设备通信时，若在3秒内未收到响应，则认为通信超时并抛出异常。这有助于避免长时间等待无响应的设备。
+        master.setTimeout(5000);
+        return master;
+    }
 
     public Map<String, Object> readDataFromDevice(DeviceConfig config, ModbusMaster master) {
         Map<String, Object> deviceData = new HashMap<>();
 //        ModbusMaster master = null; // 将master的声明提前
         try {
-            master = connectionManager.borrowMaster(config);
+//            master = connectionManager.borrowMaster(config);
 //            设置了Modbus通信的超时时间为3000毫秒（3秒）。当主设备与从设备通信时，若在3秒内未收到响应，则认为通信超时并抛出异常。这有助于避免长时间等待无响应的设备。
-            master.setTimeout(5000);
+//            master.setTimeout(5000);
             BatchResults<String> results = readTagValues(master, config.getSlaveId(), config.getTags());
             for (TagConfig tag : config.getTags()) {
                 if (Objects.equals(tag.getDataType(), "FOUR_BYTE_FLOAT_DBCA")){
@@ -199,7 +205,7 @@ public class ModbusProcessor {
 //                }
             }
         } catch (Exception e) {
-            logger.error("Failed to borrow connection or read from devices '{}'", config.getDeviceName(), e);
+            logger.error("Failed read from devices '{}'", config.getDeviceName(), e);
         }
         finally {
             // 关键：无论成功与否，都必须将连接归还到池中
@@ -207,8 +213,8 @@ public class ModbusProcessor {
                 connectionManager.returnMaster(config, master);
             }
         }
-        String deviceNumber = config.getDeviceNumber();
+//        String deviceNumber = config.getDeviceNumber();
-        redisCache.setCacheObject(deviceNumber, deviceData);
+//        redisCache.setCacheObject(deviceNumber, deviceData);
         return deviceData;
     }
 
@@ -341,7 +347,7 @@ public class ModbusProcessor {
                 }
             }
             return results;
-        }catch (Exception e){
+        } catch (Exception e){
             logger.error("Failed to read master '{}'", slaveId, e);
             throw new Exception(e);
         }

ems-quartz/src/main/java/com/xzzn/quartz/task/ModbusPoller.java

@@ -8,12 +8,18 @@ import com.xzzn.common.constant.RedisKeyConstants;
 import com.xzzn.common.core.modbus.ModbusProcessor;
 import com.xzzn.common.core.modbus.domain.DeviceConfig;
 import com.xzzn.common.core.redis.RedisCache;
+import com.xzzn.common.enums.DeviceRunningStatus;
 import com.xzzn.common.utils.DateUtils;
+import com.xzzn.ems.domain.EmsDevicesSetting;
+import com.xzzn.ems.mapper.EmsDevicesSettingMapper;
 import com.xzzn.ems.service.IEmsAlarmRecordsService;
 import com.xzzn.ems.service.impl.DeviceDataProcessServiceImpl;
 import com.xzzn.framework.manager.MqttLifecycleManager;
 import com.xzzn.framework.web.service.MqttPublisher;
 import com.xzzn.quartz.config.ScheduledTask;
+import com.xzzn.quartz.domain.SysJob;
+import com.xzzn.quartz.service.ISysJobService;
+import com.xzzn.quartz.util.CronUtils;
 
 import java.io.IOException;
 import java.nio.file.Files;
@@ -24,6 +30,8 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -44,14 +52,19 @@ public class ModbusPoller {
     private final MqttLifecycleManager mqttLifecycleManager;
     private final ScheduledTask scheduledTask;
     private final ObjectMapper objectMapper = new ObjectMapper();
+    private final Map<String, Integer> deviceFailureCounts = new ConcurrentHashMap<>();
 
     @Autowired
     private ModbusProcessor modbusProcessor;
     @Autowired
     private IEmsAlarmRecordsService iEmsAlarmRecordsService;
     @Autowired
+    private ISysJobService iSysJobService;
+    @Autowired
     private DeviceDataProcessServiceImpl deviceDataProcessServiceImpl;
     @Autowired
+    private EmsDevicesSettingMapper emsDevicesSettingMapper;
+    @Autowired
     private RedisCache redisCache;
     @Autowired
     private MqttPublisher mqttPublisher;
@@ -101,16 +114,17 @@ public class ModbusPoller {
             }
         }
 
+        int interval = getScheduledTaskInterval();
         // 为每个 host:port 启动一个任务
         for (Map.Entry<String, List<DeviceConfig>> entry : groupedConfigs.entrySet()) {
             String groupKey = entry.getKey();
             List<DeviceConfig> configs = entry.getValue();
             // 取其中一个配置的时间间隔作为该组任务的执行周期
-            long interval = configs.get(0).getTime();
+//            long interval = configs.get(0).getTime();
             scheduledTask.startTask(groupKey, () -> {
                 for (DeviceConfig config : configs) {
                     try {
-                        scheduledStart(config, null);
+                        scheduledStart(config);
                     } catch (Exception e) {
                         log.error("采集设备数据异常: {}", config.getDeviceName(), e);
                     }
@@ -119,10 +133,18 @@ public class ModbusPoller {
         }
     }
 
-    public void scheduledStart(DeviceConfig config, ModbusMaster master) {
+    public void scheduledStart(DeviceConfig config) {
-        try {
         if (config.isEnabled()) {
             log.info("Reading data from devices: {}", config.getDeviceName());
+            ModbusMaster master = null;
+            try {
+                master = modbusProcessor.borrowMaster(config);
+            } catch (Exception e) {
+                log.error("Failed to borrow connection '{}'", config.getDeviceName(), e);
+                // 处理设备连接失败的情况，更新设备状态为离线，添加报警记录
+                addDeviceOfflineRecord(siteId, config.getDeviceNumber());
+                return;
+            }
             Map<String, Object> data = modbusProcessor.readDataFromDevice(config, master);
             // 在这里处理采集到的数据
             config.getTags().forEach(tag -> {
@@ -154,17 +176,29 @@ public class ModbusPoller {
             //处理数据并发送MQTT消息、保存Redis数据和数据入库
             processingData(data, deviceNumber);
         }
-        } catch (Exception e) {
-            log.error("设备数据采集异常: {}", config.getDeviceName(), e);
-        }
     }
 
     private void processingData(Map<String, Object> data, String deviceNumber) {
         if (data == null || data.size() == 0) {
-            // 添加设备告警
+            // 增加失败计数
-            iEmsAlarmRecordsService.addEmptyDataAlarmRecord(siteId, deviceNumber);
+            int failureCount = deviceFailureCounts.getOrDefault(deviceNumber, 0) + 1;
+            deviceFailureCounts.put(deviceNumber, failureCount);
+
+            log.warn("设备 {} 数据读取失败，当前连续失败次数: {}", deviceNumber, failureCount);
+
+            // 连续6次失败触发报警
+            if (failureCount >= 6) {
+                addDeviceOfflineRecord(siteId, deviceNumber);
+                log.error("设备 {} 连续 {} 次未读取到数据，触发报警", deviceNumber, failureCount);
+            }
             return;
         }
+
+        // 数据读取成功，重置计数器
+        deviceFailureCounts.remove(deviceNumber);
+        updateDeviceStatus(siteId, deviceNumber, DeviceRunningStatus.RUNNING.getCode());
+
+        // 发送MQTT消息、保存Redis数据和数据入库
         Long timestamp = System.currentTimeMillis();
         JSONObject json = new JSONObject();
         json.put("Data", data);
@@ -201,4 +235,26 @@ public class ModbusPoller {
         deviceDataProcessServiceImpl.processingDeviceData(siteId, deviceNumber, JSON.toJSONString(data), DateUtils.convertUpdateTime(timestamp));
     }
 
+    //处理设备连接失败的情况，更新设备状态为离线，添加报警记录
+    private void addDeviceOfflineRecord(String siteId, String deviceNumber) {
+        updateDeviceStatus(siteId, deviceNumber, DeviceRunningStatus.OFFLINE.getCode());
+        iEmsAlarmRecordsService.addDeviceOfflineRecord(siteId, deviceNumber);
+    }
+
+    // 更新设备状态为在线或离线
+    private void updateDeviceStatus(String siteId, String deviceNumber, String deviceStatus) {
+        EmsDevicesSetting emsDevicesSetting = emsDevicesSettingMapper.getDeviceBySiteAndDeviceId(deviceNumber,  siteId);
+        if (emsDevicesSetting != null && !Objects.equals(emsDevicesSetting.getDeviceStatus(), deviceStatus)) {
+            emsDevicesSetting.setDeviceStatus(deviceStatus);
+            emsDevicesSettingMapper.updateEmsDevicesSetting(emsDevicesSetting);
+        }
+    }
+
+    private int getScheduledTaskInterval() {
+        SysJob query = new SysJob();
+        query.setInvokeTarget("modbusPoller.pollAllDevices");
+        List<SysJob> sysJobs = iSysJobService.selectJobList(query);
+        return Math.toIntExact(CronUtils.getNextExecutionIntervalMillis(sysJobs.get(0).getCronExpression()));
+    }
+
 }

ems-quartz/src/main/java/com/xzzn/quartz/util/CronUtils.java

@@ -60,4 +60,25 @@ public class CronUtils
             throw new IllegalArgumentException(e.getMessage());
         }
     }
+
+    /**
+     * 返回当前时间到下一次执行时间间隔的毫秒
+     */
+    public static long getNextExecutionIntervalMillis(String cronExpression)
+    {
+        try
+        {
+            CronExpression cron = new CronExpression(cronExpression);
+            Date now = new Date();
+            Date nextExecution = cron.getNextValidTimeAfter(now);
+            Date nextExecution2 = cron.getNextValidTimeAfter(nextExecution);
+
+            return nextExecution2.getTime() - nextExecution.getTime();
+        }
+        catch (ParseException e)
+        {
+            throw new IllegalArgumentException(e.getMessage());
+        }
+    }
+
 }

ems-system/src/main/java/com/xzzn/ems/service/IEmsAlarmRecordsService.java

@@ -1,10 +1,11 @@
 package com.xzzn.ems.service;
 
-import java.util.List;
 import com.xzzn.ems.domain.EmsAlarmRecords;
 import com.xzzn.ems.domain.vo.AlarmRecordListRequestVo;
 import com.xzzn.ems.domain.vo.AlarmRecordListResponseVo;
 
+import java.util.List;
+
 /**
  * 告警记录Service接口
  *
@@ -98,4 +99,6 @@ public interface IEmsAlarmRecordsService
 
     // 处理本地端同步的保护策略告警信息
     public void dealSyncData(String content, String operateType);
+
+    public void addDeviceOfflineRecord(String siteId, String deviceNumber);
 }

ems-system/src/main/java/com/xzzn/ems/service/impl/DeviceDataProcessServiceImpl.java

@@ -31,7 +31,25 @@ import com.xzzn.ems.domain.*;
 import com.xzzn.ems.domain.vo.EnergyPriceTimeRange;
 import com.xzzn.ems.domain.vo.EnergyPriceVo;
 import com.xzzn.ems.enums.DeviceMatchTable;
-import com.xzzn.ems.mapper.*;
+import com.xzzn.ems.mapper.EmsAmmeterDataMapper;
+import com.xzzn.ems.mapper.EmsBatteryClusterMapper;
+import com.xzzn.ems.mapper.EmsBatteryDataMapper;
+import com.xzzn.ems.mapper.EmsBatteryDataMinutesMapper;
+import com.xzzn.ems.mapper.EmsBatteryGroupMapper;
+import com.xzzn.ems.mapper.EmsBatteryStackMapper;
+import com.xzzn.ems.mapper.EmsClusterAlarmDataMapper;
+import com.xzzn.ems.mapper.EmsCoolingAlarmDataMapper;
+import com.xzzn.ems.mapper.EmsCoolingDataMapper;
+import com.xzzn.ems.mapper.EmsDailyChargeDataMapper;
+import com.xzzn.ems.mapper.EmsDailyEnergyDataMapper;
+import com.xzzn.ems.mapper.EmsDevicesSettingMapper;
+import com.xzzn.ems.mapper.EmsDhDataMapper;
+import com.xzzn.ems.mapper.EmsEmsDataMapper;
+import com.xzzn.ems.mapper.EmsPcsAlarmDataMapper;
+import com.xzzn.ems.mapper.EmsPcsBranchDataMapper;
+import com.xzzn.ems.mapper.EmsPcsDataMapper;
+import com.xzzn.ems.mapper.EmsStackAlarmDataMapper;
+import com.xzzn.ems.mapper.EmsXfDataMapper;
 import com.xzzn.ems.service.IDeviceDataProcessService;
 import com.xzzn.ems.service.IEmsAlarmRecordsService;
 import com.xzzn.ems.service.IEmsDeviceSettingService;
@@ -557,21 +575,31 @@ public class DeviceDataProcessServiceImpl extends AbstractBatteryDataProcessor i
         }
     }
 
+    /**
+     *
+     * @param pointMatchList 字段匹配规则列表
+     * @param obj 原始数据Map
+     * @param entity 目标Java对象
+     * @param pointEnumMatchMap 枚举匹配规则
+     */
     private void saveDeviceData(List<EmsPointMatch> pointMatchList, Map<String, Object> obj,
                                 Object entity, Map<String, List<EmsPointEnumMatch>> pointEnumMatchMap) {
         Map<String, String> pointMatchMap = pointMatchList.stream()
                 .collect(Collectors.toMap(
-                        data -> StringUtils.toCamelCase(data.getMatchField()),
+                        data -> StringUtils.toCamelCase(data.getMatchField()), // 源字段名转为驼峰
-                        EmsPointMatch::getDataPoint,
+                        EmsPointMatch::getDataPoint, // 获取目标点位名
-                        (existing, replacement) -> replacement));
+                        (existing, replacement) -> replacement)); // 处理重复键
         Field[] fields = entity.getClass().getDeclaredFields();
         for (Field field : fields) {
             String fieldName = field.getName();
             if (pointMatchMap.containsKey(fieldName)) {
-                field.setAccessible(true);
+                // 处理匹配的字段名，并设置值
+                field.setAccessible(true); // 允许访问私有字段
                 try {
+                    // 1. 从原始数据中获取匹配值
                     Object matchValue = obj.get(pointMatchMap.get(fieldName));
-                    //匹配枚举值转换
+
+                    // 2. 处理枚举值转换
                     List<EmsPointEnumMatch> pointEnumMatchList = pointEnumMatchMap.get(fieldName);
                     if (CollectionUtils.isNotEmpty(pointEnumMatchList)) {
                         Object finalMatchValue = matchValue;
@@ -581,6 +609,8 @@ public class DeviceDataProcessServiceImpl extends AbstractBatteryDataProcessor i
                             matchValue = enumMatch.get().getEnumCode();
                         }
                     }
+
+                    // 3. 类型转换
                     Class<?> fieldType = field.getType();
                     if (String.class.equals(fieldType)) {
                         matchValue = StringUtils.getString(matchValue);
@@ -591,6 +621,8 @@ public class DeviceDataProcessServiceImpl extends AbstractBatteryDataProcessor i
                     } else if (Integer.class.equals(fieldType)) {
                         matchValue = MapUtils.getInteger(obj, pointMatchMap.get(fieldName));
                     }
+
+                    // 4. 设置字段值
                     field.set(entity, matchValue);
                 } catch (IllegalAccessException e) {
                     log.warn("deviceDataProcess 设置字段值时出错", e);
@@ -1001,14 +1033,14 @@ public class DeviceDataProcessServiceImpl extends AbstractBatteryDataProcessor i
 
         if (!deviceId.contains(SiteDevice.METEGF.name()) && !deviceId.contains(SiteDevice.METE0.name())) {
             // 处理储能电表-METE每日充放电数据
-            EmsAmmeterData yestData = dealDailyChargeDate(siteId, deviceId, dataMete);
+            dealDailyChargeDate(siteId, deviceId, dataMete);
 
             // 处理储能电表-METE每日数据(尖、峰、平、谷差值)
-            if (SiteEnum.FX.getCode().equals(siteId)) {
+//            if (SiteEnum.FX.getCode().equals(siteId)) {
                 dealAmmeterDailyDate(siteId, dataMete, dataUpdateTime, lastAmmeterData);
-            } else if (SiteEnum.DDS.getCode().equals(siteId)) {
+//            } else if (SiteEnum.DDS.getCode().equals(siteId)) {
-                dealDailyEnergyData(siteId, dataMete, yestData);
+//                dealDailyEnergyData(siteId, dataMete, yestData);
-            }
+//            }
         }
     }
 

ems-system/src/main/java/com/xzzn/ems/service/impl/EmsAlarmRecordsServiceImpl.java

@@ -1,9 +1,5 @@
 package com.xzzn.ems.service.impl;
 
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
 import com.alibaba.fastjson2.JSON;
 import com.xzzn.common.constant.RedisKeyConstants;
 import com.xzzn.common.core.domain.entity.SysUser;
@@ -14,17 +10,26 @@ import com.xzzn.common.enums.TicketStatus;
 import com.xzzn.common.utils.DateUtils;
 import com.xzzn.common.utils.StringUtils;
 import com.xzzn.ems.domain.EmsAlarmMatchData;
+import com.xzzn.ems.domain.EmsAlarmRecords;
 import com.xzzn.ems.domain.EmsTicket;
 import com.xzzn.ems.domain.vo.AlarmRecordListRequestVo;
 import com.xzzn.ems.domain.vo.AlarmRecordListResponseVo;
 import com.xzzn.ems.mapper.EmsAlarmMatchDataMapper;
+import com.xzzn.ems.mapper.EmsAlarmRecordsMapper;
 import com.xzzn.ems.mapper.EmsTicketMapper;
+import com.xzzn.ems.service.IEmsAlarmRecordsService;
 import com.xzzn.system.mapper.SysUserMapper;
+
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
-import com.xzzn.ems.mapper.EmsAlarmRecordsMapper;
-import com.xzzn.ems.domain.EmsAlarmRecords;
-import com.xzzn.ems.service.IEmsAlarmRecordsService;
 
 /**
  * 告警记录Service业务层处理
@@ -308,5 +313,18 @@ public class EmsAlarmRecordsServiceImpl implements IEmsAlarmRecordsService
         }
     }
 
+    @Override
+    public void addDeviceOfflineRecord(String siteId, String deviceId) {
+        EmsAlarmRecords emsAlarmRecords = redisCache.getCacheObject(RedisKeyConstants.MODBUS_OFFLINE_ALARM_RECORD + siteId + "_" + deviceId);
+        if (emsAlarmRecords != null) {
+            return;
+        }
+        emsAlarmRecords =  createAlarmAtPcs(siteId, deviceId,"modbus连接设备失败", AlarmLevelStatus.EMERGENCY.getCode());
+        emsAlarmRecordsMapper.insertEmsAlarmRecords(emsAlarmRecords);
+
+        // 存redis-防止重复插入-有效期一天
+        redisCache.setCacheObject(RedisKeyConstants.MODBUS_OFFLINE_ALARM_RECORD + siteId + "_" + deviceId, emsAlarmRecords,1, TimeUnit.DAYS);
+    }
+
 
 }