From 14930678e6d70317d1e56acc1ff326010a6ec388 Mon Sep 17 00:00:00 2001 From: tangwei Date: Fri, 10 Jul 2026 13:54:04 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BC=98=E5=8C=96=E6=B0=B4=E6=B8=A9?= =?UTF-8?q?=E7=AB=99=E9=94=9A=E7=82=B9=E6=95=B0=E6=8D=AE=E8=8E=B7=E5=8F=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/yfd/platform/config/JobRunner.java | 13 +- .../service/impl/SdWtMonitorServiceImpl.java | 19 ++ .../qgc_job/service/IDwJobService.java | 10 + .../service/impl/DwJobServiceImpl.java | 291 ++++++++++++++++++ .../yfd/platform/qgc_job/vo/WtLastDataVo.java | 59 ++++ .../platform/qgc_job/vo/WtvtPointDataVo.java | 37 +++ .../yfd/platform/qgc_job/vo/WtvtPointVo.java | 37 +++ 7 files changed, 463 insertions(+), 3 deletions(-) create mode 100644 backend/src/main/java/com/yfd/platform/qgc_job/service/IDwJobService.java create mode 100644 backend/src/main/java/com/yfd/platform/qgc_job/service/impl/DwJobServiceImpl.java create mode 100644 backend/src/main/java/com/yfd/platform/qgc_job/vo/WtLastDataVo.java create mode 100644 backend/src/main/java/com/yfd/platform/qgc_job/vo/WtvtPointDataVo.java create mode 100644 backend/src/main/java/com/yfd/platform/qgc_job/vo/WtvtPointVo.java diff --git a/backend/src/main/java/com/yfd/platform/config/JobRunner.java b/backend/src/main/java/com/yfd/platform/config/JobRunner.java index 6154f2d6..ebabd5c4 100644 --- a/backend/src/main/java/com/yfd/platform/config/JobRunner.java +++ b/backend/src/main/java/com/yfd/platform/config/JobRunner.java @@ -20,6 +20,7 @@ import cn.hutool.core.date.DateUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.yfd.platform.qgc_env.wq.entity.ao.JobBaseAo; import com.yfd.platform.qgc_env.wq.service.EnvWqDataService; +import com.yfd.platform.qgc_job.service.IDwJobService; import com.yfd.platform.qgc_job.service.IEqJobService; import com.yfd.platform.system.domain.QuartzJob; import com.yfd.platform.system.mapper.QuartzJobMapper; @@ -48,7 +49,8 @@ public class JobRunner implements ApplicationRunner { private final QuartzJobMapper quartzJobMapper; private final QuartzManage quartzManage; private final EnvWqDataService envWqDataService; - private final IEqJobService iEqJobService; + private final IEqJobService eqJobService; + private final IDwJobService dwJobService; /** * 项目启动时重新激活启用的定时任务 @@ -71,8 +73,13 @@ public class JobRunner implements ApplicationRunner { ao.setStartTime(DateUtil.beginOfYear(DateUtil.offset(ao.getEndTime(), DateField.YEAR,-1))); } envWqDataService.wqLastData(ao); - - iEqJobService.eqAnchorPointLastDate( ao); log.info("--------------------wq最新一条数据---------------------"); + eqJobService.eqAnchorPointLastDate(ao); + log.info("--------------------eq最新一条数据---------------------"); + eqJobService.eqAnchorPointLastDate(ao); + log.info("--------------------eq最新一条数据---------------------"); + dwJobService.wtLastData(ao); + log.info("--------------------wt最新一条数据---------------------"); + dwJobService.wtvtPointData(ao); } } diff --git a/backend/src/main/java/com/yfd/platform/qgc_env/wt/service/impl/SdWtMonitorServiceImpl.java b/backend/src/main/java/com/yfd/platform/qgc_env/wt/service/impl/SdWtMonitorServiceImpl.java index 5aa6b96f..1e3d7581 100644 --- a/backend/src/main/java/com/yfd/platform/qgc_env/wt/service/impl/SdWtMonitorServiceImpl.java +++ b/backend/src/main/java/com/yfd/platform/qgc_env/wt/service/impl/SdWtMonitorServiceImpl.java @@ -27,7 +27,10 @@ import com.yfd.platform.qgc_env.wt.entity.vo.WaterTempMapVO; import com.yfd.platform.qgc_env.wt.mapper.SdWtMonitorMapper; import com.yfd.platform.qgc_env.wt.utils.SiteAvoidanceUtils; import com.yfd.platform.qgc_env.wt.service.SdWtMonitorService; +import com.yfd.platform.qgc_job.vo.WtLastDataVo; +import com.yfd.platform.qgc_job.vo.WtvtPointVo; import com.yfd.platform.utils.QgcQueryWrapperUtil; +import com.yfd.platform.utils.RedisCacheUtil; import com.yfd.platform.utils.SecurityUtils; import jakarta.annotation.Resource; import lombok.Data; @@ -63,6 +66,9 @@ public class SdWtMonitorServiceImpl implements SdWtMonitorService { @Resource private IDataScopeFilterService dataScopeFilterService; + @Resource + private RedisCacheUtil redisCacheUtil; + @Override public DataSourceResult getEvnmAutoMonitorList(DataSourceRequest dataSourceRequest) { DataSourceLoadOptionsBase loadOptions = dataSourceRequest == null ? null : dataSourceRequest.toDevRequest(); @@ -3044,6 +3050,19 @@ public class SdWtMonitorServiceImpl implements SdWtMonitorService { vo.setAnchoPointState("WT_2"); } } + + if(vo.getEnfc()==1){ + //垂向水温 + WtvtPointVo wtvtPointVo =redisCacheUtil.fromJson(redisCacheUtil.getString("wtvtPointData:" + vo.getStcd()), WtvtPointVo.class); + vo.setWtvtInfo(wtvtPointVo); + }else{ + //河道水温 + WtLastDataVo sdWtrvR = redisCacheUtil.fromJson(redisCacheUtil.getString("wtLastData:" + vo.getStcd()), WtLastDataVo.class); + if(sdWtrvR !=null){ + vo.setTm(sdWtrvR.getTm()); + vo.setTemperature(sdWtrvR.getWt()); + } + } } SiteAvoidanceUtils.calcSiteMapLev(resultList, diff --git a/backend/src/main/java/com/yfd/platform/qgc_job/service/IDwJobService.java b/backend/src/main/java/com/yfd/platform/qgc_job/service/IDwJobService.java new file mode 100644 index 00000000..dff8a481 --- /dev/null +++ b/backend/src/main/java/com/yfd/platform/qgc_job/service/IDwJobService.java @@ -0,0 +1,10 @@ +package com.yfd.platform.qgc_job.service; + +import com.yfd.platform.qgc_env.wq.entity.ao.JobBaseAo; + +public interface IDwJobService { + + void wtLastData(JobBaseAo ao); + + void wtvtPointData(JobBaseAo ao); +} diff --git a/backend/src/main/java/com/yfd/platform/qgc_job/service/impl/DwJobServiceImpl.java b/backend/src/main/java/com/yfd/platform/qgc_job/service/impl/DwJobServiceImpl.java new file mode 100644 index 00000000..894253ff --- /dev/null +++ b/backend/src/main/java/com/yfd/platform/qgc_job/service/impl/DwJobServiceImpl.java @@ -0,0 +1,291 @@ +package com.yfd.platform.qgc_job.service.impl; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.date.DateTime; +import cn.hutool.core.date.DateUtil; +import cn.hutool.json.JSONUtil; +import com.yfd.platform.common.MicroservicDynamicSQLMapper; +import com.yfd.platform.qgc_env.wq.entity.ao.JobBaseAo; +import com.yfd.platform.qgc_job.service.IDwJobService; +import com.yfd.platform.qgc_job.vo.WtLastDataVo; +import com.yfd.platform.qgc_job.vo.WtvtPointDataVo; +import com.yfd.platform.qgc_job.vo.WtvtPointVo; +import com.yfd.platform.utils.RedisCacheUtil; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.Date; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +@Slf4j +@Service +public class DwJobServiceImpl implements IDwJobService { + + private static final String WT_LAST_DATA_KEY_PREFIX = "wtLastData:"; + private static final String WTVT_POINT_DATA_KEY_PREFIX = "wtvtPointData:"; + + @Resource + private MicroservicDynamicSQLMapper microservicDynamicSQLMapper; + + @Resource + private RedisCacheUtil redisCacheUtil; + + @Override + public void wtLastData(JobBaseAo ao) { + if (ao == null) { + throw new RuntimeException("任务参数不能为空"); + } + ao.initTime(); + Date startTime = ao.getStartTime(); + Date endTime = ao.getEndTime(); + Set stcds = ao.getStcds(); + + List> dateList = getInterValDate(startTime, endTime); + Map lastDataMap = new LinkedHashMap<>(); + for (int index = dateList.size() - 1; index >= 0; index--) { + Map dateMap = dateList.get(index); + Date segmentStartTime = dateMap.get("startTime"); + Date segmentEndTime = dateMap.get("endTime"); + List wtLastDataList = getWtLastData(stcds, segmentStartTime, segmentEndTime); + for (WtLastDataVo wtLastDataVo : wtLastDataList) { + if (wtLastDataVo == null || wtLastDataVo.getStcd() == null || lastDataMap.containsKey(wtLastDataVo.getStcd())) { + continue; + } + lastDataMap.put(wtLastDataVo.getStcd(), wtLastDataVo); + } + } + + int count = 0; + for (Map.Entry entry : lastDataMap.entrySet()) { + String stcd = entry.getKey(); + WtLastDataVo wtLastDataVo = entry.getValue(); + WtLastDataVo redisData = redisCacheUtil.fromJson(redisCacheUtil.getString(WT_LAST_DATA_KEY_PREFIX + stcd), WtLastDataVo.class); + if (shouldRefresh(redisData, wtLastDataVo)) { + redisCacheUtil.setString(WT_LAST_DATA_KEY_PREFIX + stcd, JSONUtil.toJsonStr(wtLastDataVo)); + count++; + } else { + redisCacheUtil.delete(WT_LAST_DATA_KEY_PREFIX + stcd); + } + } + log.info("本次更新水温缓存{}条", count); + } + + @Override + public void wtvtPointData(JobBaseAo ao) { + if (ao == null) { + throw new RuntimeException("任务参数不能为空"); + } + ao.initTime(); + Date startTime = ao.getStartTime(); + Date endTime = ao.getEndTime(); + Set stcds = ao.getStcds(); + + List> dateList = getInterValDate(startTime, endTime); + Map> lastDataMap = new LinkedHashMap<>(); + for (int index = dateList.size() - 1; index >= 0; index--) { + Map dateMap = dateList.get(index); + Date segmentStartTime = dateMap.get("startTime"); + Date segmentEndTime = dateMap.get("endTime"); + List wtvtLastDataList = getWtvtPointLastData(stcds, segmentStartTime, segmentEndTime); + if (CollUtil.isEmpty(wtvtLastDataList)) { + continue; + } + Map> groupedMap = new LinkedHashMap<>(); + for (WtvtPointDataVo wtvtPointDataVo : wtvtLastDataList) { + if (wtvtPointDataVo == null || wtvtPointDataVo.getStcd() == null) { + continue; + } + groupedMap.computeIfAbsent(wtvtPointDataVo.getStcd(), key -> new ArrayList<>()).add(wtvtPointDataVo); + } + for (Map.Entry> entry : groupedMap.entrySet()) { + if (!lastDataMap.containsKey(entry.getKey())) { + lastDataMap.put(entry.getKey(), entry.getValue()); + } + } + } + + int count = 0; + for (Map.Entry> entry : lastDataMap.entrySet()) { + String stcd = entry.getKey(); + WtvtPointVo wtvtPointVo = buildWtvtPointVo(stcd, entry.getValue()); + WtvtPointVo redisData = redisCacheUtil.fromJson(redisCacheUtil.getString(WTVT_POINT_DATA_KEY_PREFIX + stcd), WtvtPointVo.class); + if (shouldRefresh(redisData, wtvtPointVo)) { + redisCacheUtil.setString(WTVT_POINT_DATA_KEY_PREFIX + wtvtPointVo.getStcd(), JSONUtil.toJsonStr(wtvtPointVo)); + count++; + } + } + log.info("本次更新垂向水温缓存{}条", count); + } + + private boolean shouldRefresh(WtLastDataVo redisData, WtLastDataVo currentData) { + if (currentData == null || currentData.getTm() == null) { + return false; + } + return redisData == null || redisData.getTm() == null || DateUtil.compare(redisData.getTm(), currentData.getTm()) < 0; + } + + private boolean shouldRefresh(WtvtPointVo redisData, WtvtPointVo currentData) { + if (currentData == null || currentData.getTm() == null) { + return false; + } + return redisData == null || redisData.getTm() == null || DateUtil.compare(redisData.getTm(), currentData.getTm()) < 0; + } + + private List getWtLastData(Set stcds, Date startTime, Date endTime) { + Map paramMap = new HashMap<>(); + paramMap.put("startTime", startTime); + paramMap.put("endTime", endTime); + + StringBuilder sql = new StringBuilder(); + sql.append("SELECT ") + .append(" data.ID AS id, ") + .append(" data.STCD AS stcd, ") + .append(" data.TM AS tm, ") + .append(" data.WT AS wt, ") + .append(" data.REMARK AS remark, ") + .append(" data.FID AS fid, ") + .append(" data.RECORD_USER AS recordUser, ") + .append(" data.RECORD_TIME AS recordTime, ") + .append(" data.MODIFY_USER AS modifyUser, ") + .append(" data.MODIFY_TIME AS modifyTime, ") + .append(" data.IS_DELETED AS isDeleted, ") + .append(" data.DELETE_USER AS deleteUser, ") + .append(" data.DELETE_TIME AS deleteTime ") + .append("FROM ( ") + .append(" SELECT t.*, ROW_NUMBER() OVER (PARTITION BY t.STCD ORDER BY t.TM DESC, t.RECORD_TIME DESC, t.ID DESC) AS rn ") + .append(" FROM SD_WTRV_R t ") + .append(" WHERE NVL(t.IS_DELETED, 0) = 0 ") + .append(" AND t.TM >= #{map.startTime} ") + .append(" AND t.TM <= #{map.endTime} "); + appendInCondition(sql, paramMap, "t.STCD", stcds, "wtLastDataStcd"); + sql.append(") data WHERE data.rn = 1"); + return microservicDynamicSQLMapper.getAllListWithResultType(sql.toString(), paramMap, WtLastDataVo.class); + } + + private List getWtvtPointLastData(Set stcds, Date startTime, Date endTime) { + Map paramMap = new HashMap<>(); + paramMap.put("startTime", startTime); + paramMap.put("endTime", endTime); + + StringBuilder sql = new StringBuilder(); + sql.append("WITH temp AS ( ") + .append(" SELECT STCD, VWT, TM, WTHG ") + .append(" FROM ( ") + .append(" SELECT STCD, VWT, TM, WTHG, DENSE_RANK() OVER (PARTITION BY STCD ORDER BY TM DESC) AS rn ") + .append(" FROM SD_WTVT_R ") + .append(" WHERE NVL(IS_DELETED, 0) = 0 ") + .append(" AND TM >= #{map.startTime} ") + .append(" AND TM <= #{map.endTime} "); + appendInCondition(sql, paramMap, "STCD", stcds, "wtvtPointDataStcd"); + sql.append(" ) ") + .append(" WHERE rn = 1 ") + .append(") ") + .append("SELECT ") + .append(" t1.STCD AS stcd, ") + .append(" t2.TM AS tm, ") + .append(" t1.AVGWT AS avgwt, ") + .append(" t2.VWT AS vwt, ") + .append(" t2.WTHG AS wthg, ") + .append(" t2.MINRN AS minrn, ") + .append(" t2.MAXRN AS maxrn ") + .append("FROM ( ") + .append(" SELECT STCD, ROUND(AVG(VWT), 2) AS AVGWT ") + .append(" FROM temp GROUP BY STCD ") + .append(") t1 ") + .append("INNER JOIN ( ") + .append(" SELECT * FROM ( ") + .append(" SELECT ") + .append(" STCD, VWT, TM, WTHG, ") + .append(" ROW_NUMBER() OVER (PARTITION BY STCD ORDER BY WTHG ASC) AS minrn, ") + .append(" ROW_NUMBER() OVER (PARTITION BY STCD ORDER BY WTHG DESC) AS maxrn ") + .append(" FROM temp ") + .append(" ) WHERE MINRN = 1 OR MAXRN = 1 ") + .append(") t2 ON t1.STCD = t2.STCD"); + return microservicDynamicSQLMapper.getAllListWithResultType(sql.toString(), paramMap, WtvtPointDataVo.class); + } + + private WtvtPointVo buildWtvtPointVo(String stcd, List wtvtPointDataVoList) { + WtvtPointVo wtvtPointVo = new WtvtPointVo(); + wtvtPointVo.setStcd(stcd); + if (CollUtil.isEmpty(wtvtPointDataVoList)) { + return wtvtPointVo; + } + for (WtvtPointDataVo wtvtPointDataVo : wtvtPointDataVoList) { + if (wtvtPointDataVo == null) { + continue; + } + wtvtPointVo.setTm(wtvtPointDataVo.getTm()); + wtvtPointVo.setAvgWt(wtvtPointDataVo.getAvgwt()); + if (Integer.valueOf(1).equals(wtvtPointDataVo.getMinrn())) { + wtvtPointVo.setMinWthg(wtvtPointDataVo.getWthg()); + wtvtPointVo.setMinWthgWt(wtvtPointDataVo.getVwt()); + } + if (Integer.valueOf(1).equals(wtvtPointDataVo.getMaxrn())) { + wtvtPointVo.setMaxWthg(wtvtPointDataVo.getWthg()); + wtvtPointVo.setMaxWthgWt(wtvtPointDataVo.getVwt()); + } + } + return wtvtPointVo; + } + + // 以指定时间间隔分隔开始时间和结束时间，保持旧平台 7 天分段口径 + private List> getInterValDate(Date startTime, Date endTime) { + int num = 7; + List> dateList = new ArrayList<>(); + long interval = DateUtil.betweenDay(startTime, endTime, false); + if (interval <= num) { + Map dateMap = new HashMap<>(); + dateMap.put("startTime", startTime); + dateMap.put("endTime", endTime); + dateList.add(dateMap); + return dateList; + } + while (startTime.getTime() < endTime.getTime()) { + Map dateMap = new HashMap<>(); + DateTime dateTime = DateUtil.offsetDay(startTime, num); + dateMap.put("startTime", startTime); + if (dateTime.getTime() < endTime.getTime()) { + dateMap.put("endTime", DateUtil.offsetSecond(dateTime, -1)); + } else { + dateMap.put("endTime", endTime); + } + dateList.add(dateMap); + startTime = dateTime; + } + dateList.sort(Comparator.comparing(item -> item.get("startTime"))); + return dateList; + } + + private void appendInCondition(StringBuilder sql, + Map paramMap, + String column, + Set values, + String prefix) { + if (CollUtil.isEmpty(values)) { + return; + } + List placeholders = new ArrayList<>(); + int index = 0; + for (String value : values) { + if (value == null || value.trim().isEmpty()) { + continue; + } + String key = prefix + index++; + paramMap.put(key, value.trim()); + placeholders.add("#{map." + key + "}"); + } + if (CollUtil.isEmpty(placeholders)) { + return; + } + sql.append(" AND ").append(column).append(" IN (") + .append(String.join(", ", placeholders)) + .append(") "); + } +} diff --git a/backend/src/main/java/com/yfd/platform/qgc_job/vo/WtLastDataVo.java b/backend/src/main/java/com/yfd/platform/qgc_job/vo/WtLastDataVo.java new file mode 100644 index 00000000..0b2046b2 --- /dev/null +++ b/backend/src/main/java/com/yfd/platform/qgc_job/vo/WtLastDataVo.java @@ -0,0 +1,59 @@ +package com.yfd.platform.qgc_job.vo; + +import com.fasterxml.jackson.annotation.JsonFormat; +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +import java.io.Serializable; +import java.math.BigDecimal; +import java.util.Date; + +@Data +@Schema(description = "水温最新缓存数据") +public class WtLastDataVo implements Serializable { + + private static final long serialVersionUID = 1L; + + @Schema(description = "主键ID") + private String id; + + @Schema(description = "站点编码") + private String stcd; + + @Schema(description = "数据时间") + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8") + private Date tm; + + @Schema(description = "水温") + private BigDecimal wt; + + @Schema(description = "备注") + private String remark; + + @Schema(description = "附件ID") + private String fid; + + @Schema(description = "创建人") + private String recordUser; + + @Schema(description = "创建时间") + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8") + private Date recordTime; + + @Schema(description = "更新人") + private String modifyUser; + + @Schema(description = "更新时间") + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8") + private Date modifyTime; + + @Schema(description = "是否删除") + private Integer isDeleted; + + @Schema(description = "删除人") + private String deleteUser; + + @Schema(description = "删除时间") + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8") + private Date deleteTime; +} diff --git a/backend/src/main/java/com/yfd/platform/qgc_job/vo/WtvtPointDataVo.java b/backend/src/main/java/com/yfd/platform/qgc_job/vo/WtvtPointDataVo.java new file mode 100644 index 00000000..7af5d3ed --- /dev/null +++ b/backend/src/main/java/com/yfd/platform/qgc_job/vo/WtvtPointDataVo.java @@ -0,0 +1,37 @@ +package com.yfd.platform.qgc_job.vo; + +import com.fasterxml.jackson.annotation.JsonFormat; +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +import java.io.Serializable; +import java.util.Date; + +@Data +@Schema(description = "垂向水温锚点中间数据") +public class WtvtPointDataVo implements Serializable { + + private static final long serialVersionUID = 1L; + + @Schema(description = "站点编码") + private String stcd; + + @Schema(description = "数据时间") + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8") + private Date tm; + + @Schema(description = "平均水温") + private Double avgwt; + + @Schema(description = "水温") + private Double vwt; + + @Schema(description = "测量深度") + private Double wthg; + + @Schema(description = "最小深度行号") + private Integer minrn; + + @Schema(description = "最大深度行号") + private Integer maxrn; +} diff --git a/backend/src/main/java/com/yfd/platform/qgc_job/vo/WtvtPointVo.java b/backend/src/main/java/com/yfd/platform/qgc_job/vo/WtvtPointVo.java new file mode 100644 index 00000000..1ec24ba9 --- /dev/null +++ b/backend/src/main/java/com/yfd/platform/qgc_job/vo/WtvtPointVo.java @@ -0,0 +1,37 @@ +package com.yfd.platform.qgc_job.vo; + +import com.fasterxml.jackson.annotation.JsonFormat; +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +import java.io.Serializable; +import java.util.Date; + +@Data +@Schema(description = "垂向水温锚点缓存数据") +public class WtvtPointVo implements Serializable { + + private static final long serialVersionUID = 1L; + + @Schema(description = "站点编码") + private String stcd; + + @Schema(description = "数据时间") + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8") + private Date tm; + + @Schema(description = "平均温度") + private Double avgWt; + + @Schema(description = "最大测量深度") + private Double maxWthg; + + @Schema(description = "最大测量深度水温") + private Double maxWthgWt; + + @Schema(description = "最小测量深度") + private Double minWthg; + + @Schema(description = "最小测量深度水温") + private Double minWthgWt; +}