fix(mes): 过滤异常未来遥测时间

liu

2 天以前 f57a9de2dbe69e7f867b39c25abdab58b401a2df

 yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/service/dv/telemetry/MesDvTelemetryServiceImpl.java

@@ -6,6 +6,7 @@
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.common.util.http.HttpUtils;
import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
import cn.iocoder.yudao.module.mes.controller.admin.dv.telemetry.vo.MesDvTelemetryCheckSummaryRespVO;
import cn.iocoder.yudao.module.mes.controller.admin.dv.telemetry.vo.MesDvTelemetryDeviceRespVO;
import cn.iocoder.yudao.module.mes.controller.admin.dv.telemetry.vo.MesDvTelemetryLatestReqVO;
import cn.iocoder.yudao.module.mes.controller.admin.dv.telemetry.vo.MesDvTelemetryPageReqVO;
@@ -17,6 +18,8 @@
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.LinkedHashMap;
@@ -35,6 +38,8 @@
@Slf4j
public class MesDvTelemetryServiceImpl implements MesDvTelemetryService {

    static final long MAX_FUTURE_SKEW_MINUTES = 5;

    @Resource
    private MesDvTelemetryMapper telemetryMapper;
    @Resource
@@ -50,12 +55,52 @@
        // 取最近一批数据，按设备+信号去重保留最新，用于实时监控视图
        List<MesDvTelemetryDO> list = telemetryMapper.selectLatest(reqVO, 5000);
        Map<String, MesDvTelemetryDO> latestMap = new LinkedHashMap<>();
        LocalDateTime queryTime = LocalDateTime.now();
        for (MesDvTelemetryDO row : list) {
            // 双重防护：即使数据库中已有未来时间的脏数据，也不能作为实时数据返回
            if (!isAcceptableTelemetryTime(row.getTelemetryDataTime(), queryTime)) {
                continue;
            }
            String key = StrUtil.blankToDefault(row.getTbDeviceId(), "")
                    + ":" + StrUtil.blankToDefault(row.getParamKeyName(), "");
            latestMap.putIfAbsent(key, row);
        }
        return new ArrayList<>(latestMap.values());
    }

    @Override
    public List<MesDvTelemetryCheckSummaryRespVO> getCheckSummary(MesDvTelemetryLatestReqVO reqVO) {
        List<MesDvTelemetryDO> list = getLatestTelemetry(reqVO);
        Map<String, MesDvTelemetryCheckSummaryRespVO> deviceMap = new LinkedHashMap<>();
        for (MesDvTelemetryDO row : list) {
            String key = StrUtil.blankToDefault(row.getTbDeviceId(), "");
            MesDvTelemetryCheckSummaryRespVO vo = deviceMap.get(key);
            if (vo == null) {
                vo = new MesDvTelemetryCheckSummaryRespVO();
                vo.setTbDeviceId(row.getTbDeviceId());
                vo.setDeviceName(row.getDeviceName());
                vo.setTotalCount(0);
                vo.setNormalCount(0);
                vo.setAnomalyCount(0);
                deviceMap.put(key, vo);
            }
            vo.setTotalCount(vo.getTotalCount() + 1);
            if (Boolean.TRUE.equals(row.getWhetherAnomaly())) {
                vo.setAnomalyCount(vo.getAnomalyCount() + 1);
            } else {
                vo.setNormalCount(vo.getNormalCount() + 1);
            }
        }
        List<MesDvTelemetryCheckSummaryRespVO> result = new ArrayList<>(deviceMap.values());
        for (MesDvTelemetryCheckSummaryRespVO vo : result) {
            if (vo.getTotalCount() != null && vo.getTotalCount() > 0) {
                BigDecimal rate = BigDecimal.valueOf(vo.getAnomalyCount() * 100.0 / vo.getTotalCount());
                vo.setAnomalyRate(rate.setScale(1, RoundingMode.HALF_UP));
            } else {
                vo.setAnomalyRate(BigDecimal.ZERO);
            }
        }
        return result;
    }

    @Override
@@ -111,10 +156,17 @@
                continue;
            }
            MesDvTelemetryDO data = convert((Map<?, ?>) rowObj);
            if (!isAcceptableTelemetryTime(data.getTelemetryDataTime(), pullTime)) {
                log.warn("[pullTelemetry][忽略异常遥测时间，deviceId({})，telemetryDataTime({})，pullTime({})]",
                        data.getTbDeviceId(), data.getTelemetryDataTime(), pullTime);
                continue;
            }
            data.setPullTime(pullTime);
            saveList.add(data);
        }
        telemetryMapper.insertBatch(saveList);
        if (CollUtil.isNotEmpty(saveList)) {
            telemetryMapper.insertBatch(saveList);
        }
        // 6. 返回统计
        result.setRecordCount(saveList.size());
        result.setDeviceCount((int) saveList.stream()
@@ -166,4 +218,13 @@
        }
    }

    /**
     * 遥测时间必须存在，且不能明显晚于本次拉取/查询时间。
     * 保留 5 分钟容差，兼容设备时钟与应用服务器之间的轻微偏差。
     */
    static boolean isAcceptableTelemetryTime(LocalDateTime telemetryTime, LocalDateTime referenceTime) {
        return telemetryTime != null && referenceTime != null
                && !telemetryTime.isAfter(referenceTime.plusMinutes(MAX_FUTURE_SKEW_MINUTES));
    }

}