重构

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

@@ -2,31 +2,25 @@ package com.xzzn.quartz.task;
 
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONObject;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.serotonin.modbus4j.ModbusMaster;
 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.modbus.domain.TagConfig;
 import com.xzzn.common.core.redis.RedisCache;
 import com.xzzn.common.enums.DeviceRunningStatus;
-import com.xzzn.common.utils.DateUtils;
+import com.xzzn.common.utils.StringUtils;
 import com.xzzn.ems.domain.EmsDevicesSetting;
+import com.xzzn.ems.domain.EmsPointConfig;
 import com.xzzn.ems.mapper.EmsDevicesSettingMapper;
+import com.xzzn.ems.mapper.EmsPointConfigMapper;
 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;
-import java.nio.file.Path;
-import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -54,10 +48,8 @@ import org.springframework.util.CollectionUtils;
 @Component("modbusPoller")
 public class ModbusPoller {
     private static final Logger log = LoggerFactory.getLogger(ModbusPoller.class);
+    private static final int SITE_DEVICE_OFFLINE_THRESHOLD = 6;
 
-    private final MqttLifecycleManager mqttLifecycleManager;
-    private final ScheduledTask scheduledTask;
-    private final ObjectMapper objectMapper = new ObjectMapper();
     private final Map<String, Integer> deviceFailureCounts = new ConcurrentHashMap<>();
     private final AtomicBoolean polling = new AtomicBoolean(false);
 
@@ -69,77 +61,46 @@ public class ModbusPoller {
     @Autowired
     private IEmsAlarmRecordsService iEmsAlarmRecordsService;
 
-    @Autowired
-    private ISysJobService iSysJobService;
     @Autowired
     private DeviceDataProcessServiceImpl deviceDataProcessServiceImpl;
     @Autowired
     private EmsDevicesSettingMapper emsDevicesSettingMapper;
     @Autowired
+    private EmsPointConfigMapper emsPointConfigMapper;
+    @Autowired
     private RedisCache redisCache;
     @Autowired
     private MqttPublisher mqttPublisher;
 
     @Value("${mqtt.topic}")
     private String topic;
-    @Value("${mqtt.siteId}")
-    private String siteId;
-
-    @Autowired
-    public ModbusPoller(MqttLifecycleManager mqttLifecycleManager, ScheduledTask scheduledTask) {
-        this.mqttLifecycleManager = mqttLifecycleManager;
-        this.scheduledTask = scheduledTask;
-    }
 
     public void pollAllDevices() {
         if (!polling.compareAndSet(false, true)) {
             log.warn("上一次轮询尚未完成，本次轮询跳过");
             return;
         }
-        Path devicesDir = Paths.get(System.getProperty("user.dir"), "devices");
-        if (!Files.exists(devicesDir)) {
-            log.error("Devices目录不存在: {}", devicesDir);
+        List<PollingTask> pollingTasks = buildPollingTasks();
+        if (CollectionUtils.isEmpty(pollingTasks)) {
+            log.warn("未查询到可用的Modbus采集点位配置，跳过本轮轮询");
             polling.set(false);
             return;
         }
 
-        List<Path> jsonFiles = null;
-        try {
-            jsonFiles = Files.list(devicesDir)
-                    .filter(path -> path.toString().endsWith(".json"))
-                    .collect(Collectors.toList());
-        } catch (IOException e) {
-            log.error("modbusPoller.loadConfigs 获取设备配置文件失败: {}", devicesDir, e);
-            polling.set(false);
-            return;
-        }
+        // 按主机IP分组，同一网关串行访问，避免连接抖动
+        Map<String, List<PollingTask>> groupedByHost = pollingTasks.stream()
+                .collect(Collectors.groupingBy(
+                        task -> task.getDeviceConfig().getHost(),
+                        HashMap::new,
+                        Collectors.toList()));
 
-        // 按主机IP分组（同一网关的不同端口也归为一组，避免并发访问导致Connection Reset）
-        Map<String, List<DeviceConfig>> groupedByHost = new HashMap<>();
-        for (Path filePath : jsonFiles) {
-            DeviceConfig config = null;
-            try {
-                config = objectMapper.readValue(filePath.toFile(), DeviceConfig.class);
-            } catch (IOException e) {
-                log.error("modbusPoller.loadConfigs 解析设备配置文件失败: {}", filePath, e);
-                continue;
-            }
-            if (config.isEnabled()) {
-                // 只按主机IP分组，确保同一网关的所有端口串行访问
-                String hostKey = config.getHost();
-                groupedByHost.computeIfAbsent(hostKey, k -> new ArrayList<>()).add(config);
-            }
-        }
-
-        // 使用单线程 executor 串行执行所有主机的 Modbus 操作
-        // 将所有主机的设备按顺序串行处理，避免任何并发访问
         Future<?> future = modbusExecutor.submit(() -> {
-            for (Map.Entry<String, List<DeviceConfig>> entry : groupedByHost.entrySet()) {
+            for (Map.Entry<String, List<PollingTask>> entry : groupedByHost.entrySet()) {
                 String hostKey = entry.getKey();
-                List<DeviceConfig> configs = entry.getValue();
-                for (DeviceConfig config : configs) {
+                List<PollingTask> tasks = entry.getValue();
+                for (PollingTask task : tasks) {
                     try {
-                        scheduledStart(config);
+                        scheduledStart(task.getSiteId(), task.getDeviceConfig());
                         // 每次读取后等待200ms，给Modbus网关足够的处理时间
                         Thread.sleep(200);
                     } catch (InterruptedException ie) {
@@ -147,7 +108,8 @@ public class ModbusPoller {
                         log.warn("Modbus轮询被中断");
                         return;
                     } catch (Exception e) {
-                        log.error("采集设备数据异常: {}", config.getDeviceName(), e);
+                        log.error("采集设备数据异常: siteId={}, deviceId={}",
+                                task.getSiteId(), task.getDeviceConfig().getDeviceNumber(), e);
                     }
                 }
                 log.info("采集设备数据{}轮询任务执行完成", hostKey);
@@ -165,7 +127,139 @@ public class ModbusPoller {
         }
     }
 
-    public void scheduledStart(DeviceConfig config) {
+    private List<PollingTask> buildPollingTasks() {
+        List<EmsPointConfig> pointConfigs = emsPointConfigMapper.selectModbusCollectPointConfigs(null);
+        if (CollectionUtils.isEmpty(pointConfigs)) {
+            return Collections.emptyList();
+        }
+
+        List<EmsDevicesSetting> allDevices = emsDevicesSettingMapper.selectEmsDevicesSettingList(null);
+        Map<String, EmsDevicesSetting> deviceMap = allDevices.stream()
+                .filter(Objects::nonNull)
+                .filter(device -> StringUtils.isNoneBlank(device.getSiteId(), device.getDeviceId()))
+                .collect(Collectors.toMap(
+                        this::buildSiteDeviceKey,
+                        device -> device,
+                        (left, right) -> left));
+
+        Map<String, List<EmsPointConfig>> pointsByDevice = pointConfigs.stream()
+                .filter(Objects::nonNull)
+                .filter(point -> StringUtils.isNoneBlank(point.getSiteId(), point.getDeviceId()))
+                .collect(Collectors.groupingBy(
+                        point -> point.getSiteId() + "_" + point.getDeviceId(),
+                        HashMap::new,
+                        Collectors.toList()));
+
+        List<PollingTask> tasks = new ArrayList<>();
+        for (Map.Entry<String, List<EmsPointConfig>> entry : pointsByDevice.entrySet()) {
+            String siteDeviceKey = entry.getKey();
+            EmsDevicesSetting device = deviceMap.get(siteDeviceKey);
+            if (device == null) {
+                log.warn("未找到设备连接配置，跳过采集: key={}", siteDeviceKey);
+                continue;
+            }
+            DeviceConfig deviceConfig = buildDeviceConfig(device, entry.getValue());
+            if (deviceConfig == null) {
+                continue;
+            }
+            tasks.add(new PollingTask(device.getSiteId(), deviceConfig));
+        }
+        return tasks;
+    }
+
+    private DeviceConfig buildDeviceConfig(EmsDevicesSetting device, List<EmsPointConfig> pointConfigs) {
+        if (device == null || CollectionUtils.isEmpty(pointConfigs)) {
+            return null;
+        }
+        if (StringUtils.isBlank(device.getIpAddress()) || device.getIpPort() == null || device.getSlaveId() == null) {
+            log.warn("设备连接参数不完整，跳过采集: siteId={}, deviceId={}", device.getSiteId(), device.getDeviceId());
+            return null;
+        }
+        List<TagConfig> tags = pointConfigs.stream()
+                .sorted(Comparator.comparing(point -> point.getModbusReadOrder() == null ? 0 : point.getModbusReadOrder()))
+                .map(this::toTagConfig)
+                .filter(Objects::nonNull)
+                .collect(Collectors.toList());
+        if (CollectionUtils.isEmpty(tags)) {
+            log.warn("设备无有效Modbus点位配置，跳过采集: siteId={}, deviceId={}", device.getSiteId(), device.getDeviceId());
+            return null;
+        }
+
+        DeviceConfig deviceConfig = new DeviceConfig();
+        deviceConfig.setEnabled(true);
+        deviceConfig.setDeviceName(device.getDeviceName());
+        deviceConfig.setDeviceNumber(device.getDeviceId());
+        deviceConfig.setHost(device.getIpAddress());
+        deviceConfig.setPort(device.getIpPort().intValue());
+        deviceConfig.setSlaveId(device.getSlaveId().intValue());
+        deviceConfig.setTags(tags);
+        return deviceConfig;
+    }
+
+    private TagConfig toTagConfig(EmsPointConfig pointConfig) {
+        if (pointConfig == null) {
+            return null;
+        }
+        if (StringUtils.isBlank(pointConfig.getDataKey())) {
+            return null;
+        }
+        String address = normalizeAddress(pointConfig.getRegisterAddress(), pointConfig.getModbusRegisterType());
+        if (StringUtils.isBlank(address)) {
+            return null;
+        }
+        if (StringUtils.isBlank(pointConfig.getModbusDataType())) {
+            return null;
+        }
+
+        TagConfig tag = new TagConfig();
+        tag.setKey(pointConfig.getDataKey().trim());
+        tag.setAddress(address);
+        tag.setDataType(pointConfig.getModbusDataType().trim());
+        tag.setA(pointConfig.getDataA() == null ? 0F : pointConfig.getDataA().floatValue());
+        tag.setK(pointConfig.getDataK() == null ? 1F : pointConfig.getDataK().floatValue());
+        tag.setB(pointConfig.getDataB() == null ? 0F : pointConfig.getDataB().floatValue());
+        tag.setBit(pointConfig.getDataBit());
+        return tag;
+    }
+
+    private String normalizeAddress(String registerAddress, String registerType) {
+        if (StringUtils.isBlank(registerAddress)) {
+            return null;
+        }
+        String normalizedAddress = registerAddress.trim();
+        if (!normalizedAddress.chars().allMatch(Character::isDigit)) {
+            log.warn("寄存器地址必须为数字，当前值: {}", normalizedAddress);
+            return null;
+        }
+        if (normalizedAddress.length() > 1) {
+            char first = normalizedAddress.charAt(0);
+            if (first >= '0' && first <= '4') {
+                return normalizedAddress;
+            }
+        }
+        return getRegisterPrefix(registerType) + normalizedAddress;
+    }
+
+    private String getRegisterPrefix(String registerType) {
+        String normalized = StringUtils.defaultString(registerType).trim().toUpperCase();
+        switch (normalized) {
+            case "COIL":
+                return "0";
+            case "DISCRETE_INPUT":
+                return "1";
+            case "INPUT_REGISTER":
+                return "3";
+            case "HOLDING_REGISTER":
+            default:
+                return "4";
+        }
+    }
+
+    private String buildSiteDeviceKey(EmsDevicesSetting device) {
+        return device.getSiteId() + "_" + device.getDeviceId();
+    }
+
+    public void scheduledStart(String siteId, DeviceConfig config) {
         if (config.isEnabled()) {
             log.info("Reading data from devices: {}", config.getDeviceName());
             
@@ -205,7 +299,7 @@ public class ModbusPoller {
             log.info("Data from {}: {}", config.getDeviceName(), data);
             String deviceNumber = config.getDeviceNumber();
             //处理数据并发送MQTT消息、保存Redis数据和数据入库
-            processingData(data, deviceNumber);
+            processingData(siteId, data, deviceNumber);
         }
     }
 
@@ -254,24 +348,25 @@ public class ModbusPoller {
         return data;
     }
 
-    private void processingData(Map<String, Object> data, String deviceNumber) {
+    private void processingData(String siteId, Map<String, Object> data, String deviceNumber) {
+        String siteDeviceKey = siteId + "_" + deviceNumber;
         if (CollectionUtils.isEmpty(data)) {
             // 增加失败计数
-            int failureCount = deviceFailureCounts.getOrDefault(deviceNumber, 0) + 1;
-            deviceFailureCounts.put(deviceNumber, failureCount);
+            int failureCount = deviceFailureCounts.getOrDefault(siteDeviceKey, 0) + 1;
+            deviceFailureCounts.put(siteDeviceKey, failureCount);
 
-            log.warn("设备 {} 数据读取失败，当前连续失败次数: {}", deviceNumber, failureCount);
+            log.warn("设备 {} 数据读取失败，当前连续失败次数: {}", siteDeviceKey, failureCount);
 
             // 连续6次失败触发报警
-            if (failureCount >= 6) {
+            if (failureCount >= SITE_DEVICE_OFFLINE_THRESHOLD) {
                 addDeviceOfflineRecord(siteId, deviceNumber);
-                log.error("设备 {} 连续 {} 次未读取到数据，触发报警", deviceNumber, failureCount);
+                log.error("设备 {} 连续 {} 次未读取到数据，触发报警", siteDeviceKey, failureCount);
             }
             return;
         }
 
         // 数据读取成功，重置计数器
-        deviceFailureCounts.remove(deviceNumber);
+        deviceFailureCounts.remove(siteDeviceKey);
         // 读取到数据后告警自恢复
         deleteDeviceOfflineRecord(siteId, deviceNumber);
 
@@ -281,9 +376,42 @@ public class ModbusPoller {
         json.put("Data", data);
         json.put("timestamp", timestamp);
         json.put("Device", deviceNumber);
-        sendMqttMsg(json);
-        saveRedisData(json, deviceNumber);
-        saveDataToDatabase(data, deviceNumber, timestamp);
+        if (shouldSendMqttOnChange(siteId, deviceNumber, data)) {
+            sendMqttMsg(json);
+        } else {
+            sendMqttHeartbeat(deviceNumber, timestamp);
+            log.info("设备 {} 数据无变化，已发送心跳MQTT", siteDeviceKey);
+        }
+        saveRedisData(siteId, json, deviceNumber);
+        saveDataToDatabase(siteId, data, deviceNumber, timestamp);
+    }
+
+    /**
+     * 逢变上送：仅当Data发生变化时才发送MQTT
+     */
+    private boolean shouldSendMqttOnChange(String siteId, String deviceNumber, Map<String, Object> currentData) {
+        JSONObject lastPayload = redisCache.getCacheObject(RedisKeyConstants.ORIGINAL_MQTT_DATA + siteId + "_" + deviceNumber);
+        if (lastPayload == null) {
+            return true;
+        }
+
+        Object lastDataObj = lastPayload.get("Data");
+        if (lastDataObj == null) {
+            return true;
+        }
+
+        JSONObject lastData = JSON.parseObject(JSON.toJSONString(lastDataObj));
+        JSONObject currentDataJson = JSON.parseObject(JSON.toJSONString(currentData));
+        return !Objects.equals(lastData, currentDataJson);
+    }
+
+    private void sendMqttHeartbeat(String deviceNumber, Long timestamp) {
+        JSONObject heartbeat = new JSONObject();
+        heartbeat.put("Device", deviceNumber);
+        heartbeat.put("timestamp", timestamp);
+        heartbeat.put("Heartbeat", 1);
+        heartbeat.put("Data", new JSONObject());
+        sendMqttMsg(heartbeat);
     }
 
     public void sendMqttMsg(JSONObject json) {
@@ -296,7 +424,7 @@ public class ModbusPoller {
     }
 
 
-    public void saveRedisData(JSONObject obj, String deviceNumber) {
+    public void saveRedisData(String siteId, JSONObject obj, String deviceNumber) {
         try {
             // 存放mqtt原始每个设备最晚一次数据，便于后面点位获取数据
             redisCache.setCacheObject(RedisKeyConstants.ORIGINAL_MQTT_DATA + siteId + "_" + deviceNumber, obj);
@@ -308,8 +436,12 @@ public class ModbusPoller {
         }
     }
 
-    private void saveDataToDatabase(Map<String, Object> data, String deviceNumber, Long timestamp) {
-        deviceDataProcessServiceImpl.processingDeviceData(siteId, deviceNumber, JSON.toJSONString(data), DateUtils.convertUpdateTime(timestamp));
+    private void saveDataToDatabase(String siteId, Map<String, Object> data, String deviceNumber, Long timestamp) {
+        JSONObject payload = new JSONObject();
+        payload.put("Device", deviceNumber);
+        payload.put("Data", JSON.toJSONString(data));
+        payload.put("timestamp", timestamp);
+        deviceDataProcessServiceImpl.handleDeviceData(Collections.singletonList(payload).toString(), siteId);
     }
 
     //处理设备连接失败的情况，更新设备状态为离线，添加报警记录
@@ -333,11 +465,22 @@ public class ModbusPoller {
         }
     }
 
-    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()));
+    private static class PollingTask {
+        private final String siteId;
+        private final DeviceConfig deviceConfig;
+
+        private PollingTask(String siteId, DeviceConfig deviceConfig) {
+            this.siteId = siteId;
+            this.deviceConfig = deviceConfig;
+        }
+
+        public String getSiteId() {
+            return siteId;
+        }
+
+        public DeviceConfig getDeviceConfig() {
+            return deviceConfig;
+        }
     }
 
 }

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

@@ -7,8 +7,11 @@ import com.xzzn.common.core.modbus.ModbusProcessor;
 import com.xzzn.common.core.modbus.domain.DeviceConfig;
 import com.xzzn.common.core.modbus.domain.WriteTagConfig;
 import com.xzzn.common.core.redis.RedisCache;
+import com.xzzn.common.enums.BusinessStatus;
+import com.xzzn.common.enums.BusinessType;
 import com.xzzn.common.enums.ChargeStatus;
 import com.xzzn.common.enums.DeviceCategory;
+import com.xzzn.common.enums.OperatorType;
 import com.xzzn.common.enums.SiteDevice;
 import com.xzzn.common.enums.SocLimit;
 import com.xzzn.common.enums.WorkStatus;
@@ -33,6 +36,8 @@ import com.xzzn.ems.mapper.EmsStrategyRuntimeConfigMapper;
 import com.xzzn.ems.mapper.EmsStrategyRunningMapper;
 import com.xzzn.ems.mapper.EmsStrategyTempMapper;
 import com.xzzn.ems.mapper.EmsStrategyTimeConfigMapper;
+import com.xzzn.system.domain.SysOperLog;
+import com.xzzn.system.service.ISysOperLogService;
 
 import java.math.BigDecimal;
 import java.math.RoundingMode;
@@ -115,6 +120,8 @@ public class StrategyPoller {
     private RedisCache redisCache;
     @Autowired
     private ModbusProcessor modbusProcessor;
+    @Autowired
+    private ISysOperLogService operLogService;
 
     @Resource(name = "modbusExecutor")
     private ExecutorService modbusExecutor;
@@ -511,8 +518,10 @@ public class StrategyPoller {
             boolean result = modbusProcessor.writeDataToDeviceWithRetry(deviceConfig);
             if (!result) {
                 logger.info("当前站点: {}, PCS设备: {} modbus控制设备{}指令发送失败", siteId, deviceId, chargeStatus.getInfo());
+                recordDeviceOperationLog(siteId, deviceId, "写功率", chargeDischargePower, false, chargeStatus.getInfo() + "功率下发失败");
                 continue;
             } else {
+                recordDeviceOperationLog(siteId, deviceId, "写功率", chargeDischargePower, true, null);
                 if (ChargeStatus.STANDBY.equals(chargeStatus)) {
                     // 待机，先写功率值，再关机
                     if (!switchDevice(pcsDevice, pcsSetting, WorkStatus.STOP)) {
@@ -541,10 +550,42 @@ public class StrategyPoller {
         if (!result) {
             pcsDevice.setWorkStatus(originalWorkStatus);
             logger.info("当前站点: {}, PCS设备: {} modbus控制设备{}指令发送失败", siteId, deviceId, workStatus.getInfo());
+            recordDeviceOperationLog(siteId, deviceId, "开关机", workStatus.getInfo(), false, workStatus.getInfo() + "指令发送失败");
+        } else {
+            recordDeviceOperationLog(siteId, deviceId, "开关机", workStatus.getInfo(), true, null);
         }
         return result;
     }
 
+    private void recordDeviceOperationLog(String siteId, String deviceId, String action, Object param, boolean success, String errorMsg) {
+        try {
+            SysOperLog operLog = new SysOperLog();
+            operLog.setTitle("策略设备控制-" + action);
+            operLog.setBusinessType(BusinessType.UPDATE.ordinal());
+            operLog.setMethod(this.getClass().getName() + "." + action);
+            operLog.setRequestMethod("SCHEDULE");
+            operLog.setOperatorType(OperatorType.OTHER.ordinal());
+            operLog.setOperName("system");
+            operLog.setOperIp("127.0.0.1");
+            operLog.setOperUrl("/quartz/strategyPoller");
+            operLog.setOperTime(DateUtils.getNowDate());
+            Map<String, Object> operParam = new HashMap<>();
+            operParam.put("siteId", siteId);
+            operParam.put("deviceId", deviceId);
+            operParam.put("action", action);
+            operParam.put("param", param);
+            operLog.setOperParam(StringUtils.substring(JSON.toJSONString(operParam), 0, 2000));
+            operLog.setJsonResult(StringUtils.substring(JSON.toJSONString(Collections.singletonMap("success", success)), 0, 2000));
+            operLog.setStatus(success ? BusinessStatus.SUCCESS.ordinal() : BusinessStatus.FAIL.ordinal());
+            if (!success) {
+                operLog.setErrorMsg(StringUtils.substring(errorMsg, 0, 2000));
+            }
+            operLogService.insertOperlog(operLog);
+        } catch (Exception e) {
+            logger.error("记录sys_oper_log失败, siteId={}, deviceId={}, action={}", siteId, deviceId, action, e);
+        }
+    }
+
     private ProtectionConstraintVo getProtectionConstraint(String siteId) {
         ProtectionConstraintVo constraint = redisCache.getCacheObject(RedisKeyConstants.PROTECTION_CONSTRAINT + siteId);
         if (constraint == null) {