From 1f7dc3452baf1d4b293112e9abc8b95199d1fa1b Mon Sep 17 00:00:00 2001 From: zch Date: Sat, 24 May 2025 23:30:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(ems):=20=E9=87=8D=E6=9E=84=E7=89=A9?= =?UTF-8?q?=E8=81=94=E7=BD=91=E6=95=B0=E6=8D=AE=E9=98=88=E5=80=BC=E5=91=8A?= =?UTF-8?q?=E8=AD=A6=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增精确的字段级阈值检查逻辑，支持多字段设备的独立规则 - 优化异常记录的生成和去重策略，确保数据准确性 - 重构原有告警任务，提高系统性能和可维护性 - 增加设备类型和监测字段的业务规则校验 --- .../mapper/EmsRecordAlarmDataMapper.java | 18 +- .../mapper/RecordIotenvInstantMapper.java | 10 + .../service/IEmsRecordAlarmDataService.java | 17 +- .../impl/EmsRecordAlarmDataServiceImpl.java | 446 +++++++++++------- .../impl/RecordIotenvInstantServiceImpl.java | 17 + .../ems/record/EmsRecordAlarmDataMapper.xml | 13 +- .../ems/record/RecordIotenvInstantMapper.xml | 12 +- 7 files changed, 355 insertions(+), 178 deletions(-) diff --git a/os-ems/src/main/java/com/os/ems/record/mapper/EmsRecordAlarmDataMapper.java b/os-ems/src/main/java/com/os/ems/record/mapper/EmsRecordAlarmDataMapper.java index f551968..6cc36c2 100644 --- a/os-ems/src/main/java/com/os/ems/record/mapper/EmsRecordAlarmDataMapper.java +++ b/os-ems/src/main/java/com/os/ems/record/mapper/EmsRecordAlarmDataMapper.java @@ -64,8 +64,22 @@ public interface EmsRecordAlarmDataMapper /** * 查询最近时间内的异常数据 * - * @param fiveMinutesAgo 开始时间 + * @param startTime 开始时间 * @return 异常数据列表 */ - public List selectRecentAlarmData(@Param("fiveMinutesAgo") Date fiveMinutesAgo); + public List selectRecentAlarmData(@Param("startTime") Date startTime); + + /** + * 检查特定条件的异常数据是否已存在（精确去重） + * + * @param monitorId 设备ID + * @param cause 异常原因 + * @param startTime 采集时间开始 + * @param endTime 采集时间结束 + * @return 数量 + */ + public Integer checkDuplicateAlarmData(@Param("monitorId") String monitorId, + @Param("cause") String cause, + @Param("startTime") Date startTime, + @Param("endTime") Date endTime); } diff --git a/os-ems/src/main/java/com/os/ems/record/mapper/RecordIotenvInstantMapper.java b/os-ems/src/main/java/com/os/ems/record/mapper/RecordIotenvInstantMapper.java index 74c3be3..7396918 100644 --- a/os-ems/src/main/java/com/os/ems/record/mapper/RecordIotenvInstantMapper.java +++ b/os-ems/src/main/java/com/os/ems/record/mapper/RecordIotenvInstantMapper.java @@ -124,4 +124,14 @@ public interface RecordIotenvInstantMapper */ public List selectLatestRecordsFromTable(@Param("tableName") String tableName, @Param("limit") int limit); + + /** + * 从指定表查询指定时间之后的记录 + * + * @param tableName 表名 + * @param startTime 开始时间 + * @return 记录列表 + */ + public List selectRecentRecordsFromTable(@Param("tableName") String tableName, + @Param("startTime") java.util.Date startTime); } diff --git a/os-ems/src/main/java/com/os/ems/record/service/IEmsRecordAlarmDataService.java b/os-ems/src/main/java/com/os/ems/record/service/IEmsRecordAlarmDataService.java index 8ca8de7..345cbf9 100644 --- a/os-ems/src/main/java/com/os/ems/record/service/IEmsRecordAlarmDataService.java +++ b/os-ems/src/main/java/com/os/ems/record/service/IEmsRecordAlarmDataService.java @@ -64,21 +64,10 @@ public interface IEmsRecordAlarmDataService * @param objIds * @return */ - public int handleExceptionsAlarmData(Long[] objIds); + int handleExceptionsAlarmData(Long[] objIds); /** - * 采集设备告警定时任务 + * 物联网数据阈值检查定时任务 */ - public void collectDeviceAlarmsTask(); - - /** - * 超过电阈值定时任务 - */ - public void exceedDnbThresholdAlarmsTask(); - - /** - * 小时耗量告警定时任务 - */ - public void hourlyConsumptionAlarmsTask(); - + void checkIotenvThresholdAlarms(); } diff --git a/os-ems/src/main/java/com/os/ems/record/service/impl/EmsRecordAlarmDataServiceImpl.java b/os-ems/src/main/java/com/os/ems/record/service/impl/EmsRecordAlarmDataServiceImpl.java index 5d2962f..cb4afd3 100644 --- a/os-ems/src/main/java/com/os/ems/record/service/impl/EmsRecordAlarmDataServiceImpl.java +++ b/os-ems/src/main/java/com/os/ems/record/service/impl/EmsRecordAlarmDataServiceImpl.java @@ -1,25 +1,22 @@ package com.os.ems.record.service.impl; import java.math.BigDecimal; -import java.util.Date; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.text.SimpleDateFormat; +import java.util.*; import java.util.stream.Collectors; import com.os.common.utils.DateUtils; import com.os.common.utils.StringUtils; -import com.os.ems.base.domain.EmsBaseCollectDeviceInfo; -import com.os.ems.base.domain.EmsBaseMonitorThreshold; -import com.os.ems.base.mapper.EmsBaseCollectDeviceInfoMapper; -import com.os.ems.base.mapper.EmsBaseMonitorThresholdMapper; +import com.os.ems.base.domain.EmsBaseEnergyType; +import com.os.ems.base.domain.EmsBaseMonitorInfo; +import com.os.ems.base.mapper.EmsBaseEnergyTypeMapper; +import com.os.ems.base.mapper.EmsBaseMonitorInfoMapper; import com.os.ems.record.domain.EmsRecordAlarmRule; -import com.os.ems.record.domain.EmsRecordDnbInstant; +import com.os.ems.record.domain.RecordIotenvInstant; import com.os.ems.record.mapper.EmsRecordAlarmRuleMapper; -import com.os.ems.record.mapper.EmsRecordDnbInstantMapper; -import com.os.ems.report.domain.EmsReportPointDnb; -import com.os.ems.report.mapper.EmsReportPointDnbMapper; +import com.os.ems.record.mapper.RecordIotenvInstantMapper; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import com.os.ems.record.mapper.EmsRecordAlarmDataMapper; import com.os.ems.record.domain.EmsRecordAlarmData; @@ -42,16 +39,38 @@ public class EmsRecordAlarmDataServiceImpl implements IEmsRecordAlarmDataService private EmsRecordAlarmRuleMapper emsRecordAlarmRuleMapper; @Autowired - private EmsBaseCollectDeviceInfoMapper emsBaseCollectDeviceInfoMapper; + private RecordIotenvInstantMapper recordIotenvInstantMapper; @Autowired - private EmsRecordDnbInstantMapper emsRecordDnbInstantMapper; + private EmsBaseMonitorInfoMapper emsBaseMonitorInfoMapper; @Autowired - private EmsBaseMonitorThresholdMapper emsBaseMonitorThresholdMapper; + private EmsBaseEnergyTypeMapper emsBaseEnergyTypeMapper; - @Autowired - private EmsReportPointDnbMapper emsReportPointDnbMapper; + /** + * 监测字段字典映射 + * 将前端字典值映射到数据库字段名和中文描述 + */ + private static final Map MONITOR_FIELD_TO_DB_FIELD = new HashMap<>(); + private static final Map MONITOR_FIELD_TO_DESC = new HashMap<>(); + + static { + // 字典值 -> 数据库字段名映射 + MONITOR_FIELD_TO_DB_FIELD.put(0L, "temperature"); // 温度 + MONITOR_FIELD_TO_DB_FIELD.put(1L, "humidity"); // 湿度 + MONITOR_FIELD_TO_DB_FIELD.put(2L, "vibration_speed"); // 振动-速度(mm/s) + MONITOR_FIELD_TO_DB_FIELD.put(3L, "vibration_displacement"); // 振动-位移(um) + MONITOR_FIELD_TO_DB_FIELD.put(4L, "vibration_acceleration"); // 振动-加速度(g) + MONITOR_FIELD_TO_DB_FIELD.put(5L, "vibration_temp"); // 振动-温度(℃) + + // 字典值 -> 中文描述映射（用于异常记录的cause字段） + MONITOR_FIELD_TO_DESC.put(0L, "温度"); + MONITOR_FIELD_TO_DESC.put(1L, "湿度"); + MONITOR_FIELD_TO_DESC.put(2L, "振动-速度(mm/s)"); + MONITOR_FIELD_TO_DESC.put(3L, "振动-位移(um)"); + MONITOR_FIELD_TO_DESC.put(4L, "振动-加速度(g)"); + MONITOR_FIELD_TO_DESC.put(5L, "振动-温度(℃)"); + } /** * 查询异常数据记录 @@ -141,172 +160,279 @@ public class EmsRecordAlarmDataServiceImpl implements IEmsRecordAlarmDataService } /** - * 采集设备告警定时任务 + * 物联网数据阈值检查定时任务 + * 每分钟执行一次，检查当天分表中的最新数据是否超过阈值 + * + * 重构说明： + * 1. 改为基于monitor_field字段的精确匹配，而非按设备类型检查所有字段 + * 2. 支持多字段设备的独立阈值规则（如振动设备的4个字段可设置不同阈值） + * 3. 优化异常记录的cause字段，使用字典描述而非数据库字段名 + * 4. 修复数据关联：monitor_id存储的是monitor_code值，与采集数据分表保持一致 + * 5. 完全重构去重逻辑：确保绝对不会产生重复数据 */ - public void collectDeviceAlarmsTask() { - long minuteValue = 1000 * 60; - Date date = new Date(); - EmsRecordAlarmRule alarmRule = new EmsRecordAlarmRule(); - alarmRule.setTriggerRule(1L); - List alarmRules = emsRecordAlarmRuleMapper.selectEmsRecordAlarmRuleList(alarmRule); - if (alarmRules.size() > 0) { - minuteValue = alarmRules.get(0).getDeviceOfflineTime() * minuteValue; - } else { - return; - } + @Scheduled(cron = "0 */1 * * * ?") // 每分钟执行一次 + @Override + public void checkIotenvThresholdAlarms() { + try { + // 1. 获取当天的分表名 + SimpleDateFormat tableFormat = new SimpleDateFormat("yyyyMMdd"); + String tableName = "record_iotenv_instant_" + tableFormat.format(new Date()); + + // 检查表是否存在 + if (!checkTableExists(tableName)) { + System.out.println("当天分表不存在: " + tableName); + return; + } + + // 2. 查询所有启用的告警规则（trigger_rule=0 超过阈值） + EmsRecordAlarmRule queryRule = new EmsRecordAlarmRule(); + queryRule.setTriggerRule(0L); // 0表示超过阈值 + List alarmRules = emsRecordAlarmRuleMapper.selectEmsRecordAlarmRuleList(queryRule); + + if (alarmRules.isEmpty()) { + System.out.println("没有启用的阈值告警规则"); + return; + } + + // 3. 按设备编码分组告警规则，每个设备可能有多个字段的规则 + // 注意：monitor_id字段存储的实际是monitor_code值，与采集数据分表的monitorId保持一致 + Map> rulesByMonitor = alarmRules.stream() + .filter(rule -> StringUtils.isNotEmpty(rule.getMonitorId())) + .collect(Collectors.groupingBy(EmsRecordAlarmRule::getMonitorId)); + + // 4. 只查询最近2分钟的数据，确保每分钟执行时不重复处理 + Date twoMinutesAgo = new Date(System.currentTimeMillis() - 2 * 60 * 1000); + List recentRecords = recordIotenvInstantMapper.selectRecentRecordsFromTable(tableName, twoMinutesAgo); + + if (recentRecords.isEmpty()) { + System.out.println("当天分表最近2分钟无新数据: " + tableName); + return; + } - EmsBaseCollectDeviceInfo collectDeviceInfo = new EmsBaseCollectDeviceInfo(); - collectDeviceInfo.setIsFlag("1"); - List deviceInfoList = emsBaseCollectDeviceInfoMapper.selectEmsBaseCollectDeviceInfoList(collectDeviceInfo); - - if (deviceInfoList.size() > 0) { - EmsRecordAlarmData recordAlarmData = new EmsRecordAlarmData(); - recordAlarmData.setAlarmType(1L); - recordAlarmData.setAlarmStatus(1L); - List alarmDataList = emsRecordAlarmDataMapper.selectEmsRecordAlarmDataList(recordAlarmData); - List collectDeviceIdList = alarmDataList.stream().map(EmsRecordAlarmData::getCollectDeviceId).collect(Collectors.toList()); - for (EmsBaseCollectDeviceInfo deviceInfo : deviceInfoList) { - if (collectDeviceIdList.contains(deviceInfo.getCollectDeviceId())) { + // 5. 获取所有相关设备信息（用于验证设备类型，确保规则合理性） + // 这里需要按monitor_code查询，因为rulesByMonitor的key是monitor_code + Set monitorCodes = rulesByMonitor.keySet(); + List monitorInfos = emsBaseMonitorInfoMapper.selectMonitorInfoByCodes(new ArrayList<>(monitorCodes)); + Map monitorInfoMap = monitorInfos.stream() + .collect(Collectors.toMap(EmsBaseMonitorInfo::getMonitorCode, m -> m)); + + // 6. 处理每条记录，基于精确的字段规则进行阈值检查 + List candidateAlarms = new ArrayList<>(); + for (RecordIotenvInstant record : recentRecords) { + String monitorId = record.getMonitorId(); // 采集数据中的monitorId实际是monitor_code + List rules = rulesByMonitor.get(monitorId); + + if (rules == null || rules.isEmpty()) { continue; } - if ((deviceInfo.getUpdateTime().getTime() + minuteValue) < date.getTime()) { - EmsRecordAlarmData alarmData = new EmsRecordAlarmData(); - alarmData.setCollectDeviceId(deviceInfo.getCollectDeviceId()); - alarmData.setCollectTime(new Date()); - alarmData.setAlarmType(1L); - alarmData.setAlarmStatus(1L); - alarmData.setAlarmData("设备离线已超过" + alarmRules.get(0).getDeviceOfflineTime() + "分钟"); - this.insertEmsRecordAlarmData(alarmData); + + EmsBaseMonitorInfo monitorInfo = monitorInfoMap.get(monitorId); + if (monitorInfo == null) { + continue; + } + + // 7. 遍历该设备的所有规则，每个规则对应一个具体的监测字段 + for (EmsRecordAlarmRule rule : rules) { + // 检查规则的基本有效性 + if (rule.getTriggerValue() == null || rule.getMonitorField() == null) { + continue; + } + + // 验证监测字段是否符合设备类型（可选的业务校验） + if (!isValidMonitorFieldForDeviceType(rule.getMonitorField(), monitorInfo.getMonitorType())) { + System.out.println("警告：设备类型与监测字段不匹配 - 设备:" + monitorId + + ", 设备类型:" + monitorInfo.getMonitorType() + + ", 监测字段:" + rule.getMonitorField()); + continue; + } + + // 8. 根据规则的monitor_field精确检查对应的数据字段 + BigDecimal threshold = rule.getTriggerValue(); + BigDecimal actualValue = getFieldValue(record, rule.getMonitorField()); + + // 9. 阈值比较：实际值 > 阈值时触发异常 + if (actualValue != null && actualValue.compareTo(threshold) > 0) { + EmsRecordAlarmData alarmData = createAlarmData(record, rule, actualValue.toString()); + candidateAlarms.add(alarmData); + + System.out.println("检测到阈值超标 - 设备:" + monitorId + + ", 字段:" + MONITOR_FIELD_TO_DESC.get(rule.getMonitorField()) + + ", 实际值:" + actualValue + + ", 阈值:" + threshold + + ", 采集时间:" + record.getCollectTime()); + } } } + + // 10. 使用严格的去重逻辑插入异常数据 + if (!candidateAlarms.isEmpty()) { + int insertedCount = insertAlarmDataWithStrictDuplicateCheck(candidateAlarms); + System.out.println("严格去重后实际插入异常数据: " + insertedCount + " 条"); + } else { + System.out.println("未检测到异常数据"); + } + + } catch (Exception e) { + System.err.println("阈值检查定时任务执行异常: " + e.getMessage()); + e.printStackTrace(); } } /** - * 超过电阈值定时任务 + * 根据监测字段获取记录中对应的数据值 + * 这是关键的字段映射方法，将字典值映射到具体的数据库字段 + * + * @param record 物联网数据记录 + * @param monitorField 监测字段字典值（0温度，1湿度，2振动-速度等） + * @return 对应字段的数据值，如果字段不存在返回null */ - public void exceedDnbThresholdAlarmsTask() { - EmsBaseMonitorThreshold monitorThreshold = new EmsBaseMonitorThreshold(); - monitorThreshold.setMonitorType(2L); - List thresholdList = emsBaseMonitorThresholdMapper.selectEmsBaseMonitorThresholdList(monitorThreshold); - - EmsRecordDnbInstant dnbInstant = new EmsRecordDnbInstant(); - List dnbInstantList = emsRecordDnbInstantMapper.selectEmsRecordDnbInstantList(dnbInstant); - Map thresholdMap = compareThresholdAndRecord(thresholdList, dnbInstantList); - //防止多次存入异常数据 - EmsRecordAlarmData recordAlarmData = new EmsRecordAlarmData(); - recordAlarmData.setAlarmType(0L); - recordAlarmData.setAlarmStatus(1L); - List alarmDataList = emsRecordAlarmDataMapper.selectEmsRecordAlarmDataList(recordAlarmData); - List monitorIdList = alarmDataList.stream().map(EmsRecordAlarmData::getMonitorId).collect(Collectors.toList()); - for (String monitorId : thresholdMap.keySet()) { - if (monitorIdList.contains(monitorId)) { - continue; - } - EmsRecordAlarmData alarmData = new EmsRecordAlarmData(); - alarmData.setMonitorId(monitorId); - alarmData.setCollectTime(new Date()); - alarmData.setAlarmType(0L); - alarmData.setAlarmStatus(1L); - alarmData.setAlarmData(thresholdMap.get(monitorId)); - this.insertEmsRecordAlarmData(alarmData); + private BigDecimal getFieldValue(RecordIotenvInstant record, Long monitorField) { + switch (monitorField.intValue()) { + case 0: // 温度 + return record.getTemperature(); + case 1: // 湿度 + return record.getHumidity(); + case 2: // 振动-速度(mm/s) + return record.getVibrationSpeed(); + case 3: // 振动-位移(um) + return record.getVibrationDisplacement(); + case 4: // 振动-加速度(g) + return record.getVibrationAcceleration(); + case 5: // 振动-温度(℃) + return record.getVibrationTemp(); + default: + System.out.println("未知的监测字段: " + monitorField); + return null; } - } /** - * 超过电阈值 对比记录 - * - * @param baseDnbThresholds - * @param recordDnbInstants - * @return + * 验证监测字段是否符合设备类型 + * 业务规则校验，确保规则配置的合理性 + * + * @param monitorField 监测字段 + * @param monitorType 设备类型 + * @return 是否匹配 */ - public Map compareThresholdAndRecord(List baseDnbThresholds, List recordDnbInstants) { - Map resultMap = new HashMap<>(); - for (EmsRecordDnbInstant recordDnbInstant : recordDnbInstants) { - String monitorCode = recordDnbInstant.getMonitorCode(); - EmsBaseMonitorThreshold baseDnbThreshold = null; - for (EmsBaseMonitorThreshold threshold : baseDnbThresholds) { - if (threshold.getMonitorCode().equals(monitorCode)) { - baseDnbThreshold = threshold; - break; - } - } - if (baseDnbThreshold != null) { - String reason = compare(recordDnbInstant, baseDnbThreshold); - if (reason != null) { - resultMap.put(monitorCode, reason); - } - } + private boolean isValidMonitorFieldForDeviceType(Long monitorField, Long monitorType) { + if (monitorField == null || monitorType == null) { + return false; } - return resultMap; - } - - private String compare(EmsRecordDnbInstant recordDnbInstant, EmsBaseMonitorThreshold baseDnbThreshold) { - if (StringUtils.isNull(recordDnbInstant.getIA())) { - return null; + + // 温湿度设备(6)应该只能选择温度(0)或湿度(1) + if (monitorType == 6L) { + return monitorField == 0L || monitorField == 1L; } - BigDecimal zero = new BigDecimal("0.00"); - if (recordDnbInstant.getVA().compareTo(zero) == 0 || recordDnbInstant.getVB().compareTo(zero) == 0 - || recordDnbInstant.getVC().compareTo(zero) == 0) { - return "缺项报警"; + + // 振动设备(10)应该只能选择振动相关字段(2,3,4,5) + if (monitorType == 10L) { + return monitorField >= 2L && monitorField <= 5L; } - if (recordDnbInstant.getIA().compareTo(baseDnbThreshold.getiAMax()) > 0) { - return "超过A相电流最大值"; + + // 纯温度设备(5)只能选择温度(0) + if (monitorType == 5L) { + return monitorField == 0L; } - if (recordDnbInstant.getIB().compareTo(baseDnbThreshold.getiBMax()) > 0) { - return "超过B相电流最大值"; - } - if (recordDnbInstant.getIC().compareTo(baseDnbThreshold.getiCMax()) > 0) { - return "超过C相电流最大值"; - } - if (recordDnbInstant.getVA().compareTo(baseDnbThreshold.getvAMax()) > 0) { - return "超过A相电压最大值"; - } - if (recordDnbInstant.getVB().compareTo(baseDnbThreshold.getvBMax()) > 0) { - return "超过B相电压最大值"; - } - if (recordDnbInstant.getVC().compareTo(baseDnbThreshold.getvCMax()) > 0) { - return "超过C相电压最大值"; - } - if (recordDnbInstant.getIA().compareTo(baseDnbThreshold.getiAMin()) < 0) { - return "小于A相电流最小值"; - } - if (recordDnbInstant.getIB().compareTo(baseDnbThreshold.getiBMin()) < 0) { - return "小于B相电流最小值"; - } - if (recordDnbInstant.getIC().compareTo(baseDnbThreshold.getiCMin()) < 0) { - return "小于C相电流最小值"; - } - if (recordDnbInstant.getVA().compareTo(baseDnbThreshold.getvAMin()) < 0) { - return "小于A相电压最小值"; - } - if (recordDnbInstant.getVB().compareTo(baseDnbThreshold.getvBMin()) < 0) { - return "小于B相电压最小值"; - } - if (recordDnbInstant.getVC().compareTo(baseDnbThreshold.getvCMin()) < 0) { - return "小于C相电压最小值"; - } - return null; + + // 其他设备类型暂不支持字段级规则 + return false; } /** - * 小时耗量告警定时任务 + * 创建异常数据对象 + * 使用字典描述作为cause，提高可读性 + * + * @param record 物联网数据记录 + * @param rule 告警规则 + * @param actualValue 实际数值 + * @return 异常数据对象 */ - public void hourlyConsumptionAlarmsTask() { - EmsBaseMonitorThreshold monitorThreshold = new EmsBaseMonitorThreshold(); - monitorThreshold.setMonitorType(2L); - List thresholdList = emsBaseMonitorThresholdMapper.selectEmsBaseMonitorThresholdList(monitorThreshold); - for (EmsBaseMonitorThreshold threshold : thresholdList) { - BigDecimal consumption = threshold.getHourConsumption(); - if (consumption.equals(new BigDecimal(0))){ - continue; - } - EmsReportPointDnb reportPointDnb = new EmsReportPointDnb(); - reportPointDnb.setMonitorCode(threshold.getMonitorCode()); - List dnbList = emsReportPointDnbMapper.selectEmsReportPointDnbList(reportPointDnb); - - } + private EmsRecordAlarmData createAlarmData(RecordIotenvInstant record, EmsRecordAlarmRule rule, String actualValue) { + EmsRecordAlarmData alarmData = new EmsRecordAlarmData(); + alarmData.setMonitorId(record.getMonitorId()); + alarmData.setCollectTime(record.getCollectTime()); + alarmData.setAlarmType(0L); // 0表示超过阈值 + alarmData.setAlarmStatus(1L); // 1表示未处理 + alarmData.setAlarmData(actualValue); // 记录实际超标数值 + + // 使用字典描述作为异常原因，更加友好 + String fieldDesc = MONITOR_FIELD_TO_DESC.get(rule.getMonitorField()); + alarmData.setCause(fieldDesc != null ? fieldDesc : "未知字段"); + + alarmData.setCreateTime(new Date()); + return alarmData; + } + + /** + * 使用严格的去重逻辑插入异常数据 + * 防重复策略：同一设备的同一字段在30分钟内只记录一次异常 + * 每条数据插入前都会检查数据库是否已存在相同记录 + * + * @param alarmDataList 待插入的异常数据列表 + * @return 实际插入的异常数据数量 + */ + private int insertAlarmDataWithStrictDuplicateCheck(List alarmDataList) { + int insertedCount = 0; + + // 30分钟去重窗口（前后各15分钟） + long fifteenMinutesInMs = 15 * 60 * 1000; + + for (EmsRecordAlarmData alarmData : alarmDataList) { + try { + // 为每条异常数据检查是否已存在重复记录 + Date collectTime = alarmData.getCollectTime(); + if (collectTime == null) { + continue; // 跳过没有采集时间的数据 + } + + // 构建30分钟的时间窗口 + Date startTime = new Date(collectTime.getTime() - fifteenMinutesInMs); + Date endTime = new Date(collectTime.getTime() + fifteenMinutesInMs); + + // 查询数据库是否已存在相同的异常记录 + Integer existingCount = emsRecordAlarmDataMapper.checkDuplicateAlarmData( + alarmData.getMonitorId(), + alarmData.getCause(), + startTime, + endTime + ); + + // 如果不存在重复记录，则插入 + if (existingCount == null || existingCount == 0) { + int result = insertEmsRecordAlarmData(alarmData); + if (result > 0) { + insertedCount++; + System.out.println("成功插入异常数据 - 设备:" + alarmData.getMonitorId() + + ", 原因:" + alarmData.getCause() + + ", 采集时间:" + alarmData.getCollectTime() + + ", 数值:" + alarmData.getAlarmData()); + } + } else { + System.out.println("跳过重复异常数据 - 设备:" + alarmData.getMonitorId() + + ", 原因:" + alarmData.getCause() + + ", 采集时间:" + alarmData.getCollectTime() + + ", 30分钟内已存在 " + existingCount + " 条相同记录"); + } + + } catch (Exception e) { + System.err.println("插入异常数据时发生错误: " + e.getMessage()); + e.printStackTrace(); + } + } + + return insertedCount; + } + + /** + * 检查数据库表是否存在 + * + * @param tableName 表名 + * @return 是否存在 + */ + private boolean checkTableExists(String tableName) { + Map params = new HashMap<>(); + params.put("tableName", tableName); + Integer count = recordIotenvInstantMapper.checkTableExists(params); + return count != null && count > 0; } - } diff --git a/os-ems/src/main/java/com/os/ems/record/service/impl/RecordIotenvInstantServiceImpl.java b/os-ems/src/main/java/com/os/ems/record/service/impl/RecordIotenvInstantServiceImpl.java index 2d51ffc..a923874 100644 --- a/os-ems/src/main/java/com/os/ems/record/service/impl/RecordIotenvInstantServiceImpl.java +++ b/os-ems/src/main/java/com/os/ems/record/service/impl/RecordIotenvInstantServiceImpl.java @@ -12,6 +12,8 @@ import com.os.common.constant.HttpStatus; import com.os.common.core.page.TableDataInfo; import com.os.common.exception.ServiceException; import com.os.common.utils.bean.BeanUtils; +import com.os.ems.base.domain.EmsBaseMonitorInfo; +import com.os.ems.base.mapper.EmsBaseMonitorInfoMapper; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import com.os.ems.record.mapper.RecordIotenvInstantMapper; @@ -30,6 +32,9 @@ public class RecordIotenvInstantServiceImpl implements IRecordIotenvInstantServi @Autowired private RecordIotenvInstantMapper recordIotenvInstantMapper; + @Autowired + private EmsBaseMonitorInfoMapper emsBaseMonitorInfoMapper; + /** * 查询物联网数据 * @@ -203,6 +208,18 @@ public class RecordIotenvInstantServiceImpl implements IRecordIotenvInstantServi int offset = (pageNum - 1) * pageSize; // 执行分页查询 List list = recordIotenvInstantMapper.selectFromTablesWithPage(tableNames, recordIotenvInstant, offset, pageSize); + //获取设备列表，根据monitorId查询设备名称 +// EmsBaseMonitorInfo baseMonitorInfo = new EmsBaseMonitorInfo(); +// List baseMonitorInfos = emsBaseMonitorInfoMapper.selectEmsBaseMonitorInfoList(baseMonitorInfo); +// // 将baseMonitorInfos中的monitorName字段添加到list中 +// for (RecordIotenvInstant iotenvInstant : list) { +// for (EmsBaseMonitorInfo monitorInfo : baseMonitorInfos) { +// if (iotenvInstant.getMonitorId().equals(monitorInfo.getMonitorCode())){ +// iotenvInstant.setMonitorName(monitorInfo.getMonitorName()); +// } +// } +// } + TableDataInfo rspData = new TableDataInfo(); rspData.setCode(HttpStatus.SUCCESS); diff --git a/os-ems/src/main/resources/mapper/ems/record/EmsRecordAlarmDataMapper.xml b/os-ems/src/main/resources/mapper/ems/record/EmsRecordAlarmDataMapper.xml index d767e14..d7e4e83 100644 --- a/os-ems/src/main/resources/mapper/ems/record/EmsRecordAlarmDataMapper.xml +++ b/os-ems/src/main/resources/mapper/ems/record/EmsRecordAlarmDataMapper.xml @@ -160,6 +160,17 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" FROM ems_record_alarm_data rad WHERE rad.alarm_status = 1 AND rad.alarm_type = 0 - AND rad.create_time >= #{fiveMinutesAgo} + AND rad.create_time >= #{startTime} + + + + diff --git a/os-ems/src/main/resources/mapper/ems/record/RecordIotenvInstantMapper.xml b/os-ems/src/main/resources/mapper/ems/record/RecordIotenvInstantMapper.xml index c4ee1d5..d1aa592 100644 --- a/os-ems/src/main/resources/mapper/ems/record/RecordIotenvInstantMapper.xml +++ b/os-ems/src/main/resources/mapper/ems/record/RecordIotenvInstantMapper.xml @@ -219,7 +219,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" t.collectTime, t.recodeTime, ebmi.monitor_name FROM ${tableName} t - left join ems_base_monitor_info ebmi on t.monitorId = ebmi.obj_id + left join ems_base_monitor_info ebmi on t.monitorId = ebmi.monitor_code and monitorId = #{recordIotenvInstant.monitorId} and temperature = #{recordIotenvInstant.temperature} @@ -306,4 +306,14 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" LIMIT #{limit} + + + \ No newline at end of file