fix: 优化水温站锚点数据获取

This commit is contained in:
tangwei 2026-07-10 13:54:04 +08:00
parent f2e6b406fa
commit 14930678e6
7 changed files with 463 additions and 3 deletions

View File

@ -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);
}
}

View File

@ -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,

View File

@ -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);
}

View File

@ -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<String> stcds = ao.getStcds();
List<Map<String, Date>> dateList = getInterValDate(startTime, endTime);
Map<String, WtLastDataVo> lastDataMap = new LinkedHashMap<>();
for (int index = dateList.size() - 1; index >= 0; index--) {
Map<String, Date> dateMap = dateList.get(index);
Date segmentStartTime = dateMap.get("startTime");
Date segmentEndTime = dateMap.get("endTime");
List<WtLastDataVo> 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<String, WtLastDataVo> 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<String> stcds = ao.getStcds();
List<Map<String, Date>> dateList = getInterValDate(startTime, endTime);
Map<String, List<WtvtPointDataVo>> lastDataMap = new LinkedHashMap<>();
for (int index = dateList.size() - 1; index >= 0; index--) {
Map<String, Date> dateMap = dateList.get(index);
Date segmentStartTime = dateMap.get("startTime");
Date segmentEndTime = dateMap.get("endTime");
List<WtvtPointDataVo> wtvtLastDataList = getWtvtPointLastData(stcds, segmentStartTime, segmentEndTime);
if (CollUtil.isEmpty(wtvtLastDataList)) {
continue;
}
Map<String, List<WtvtPointDataVo>> 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<String, List<WtvtPointDataVo>> entry : groupedMap.entrySet()) {
if (!lastDataMap.containsKey(entry.getKey())) {
lastDataMap.put(entry.getKey(), entry.getValue());
}
}
}
int count = 0;
for (Map.Entry<String, List<WtvtPointDataVo>> 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<WtLastDataVo> getWtLastData(Set<String> stcds, Date startTime, Date endTime) {
Map<String, Object> 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<WtvtPointDataVo> getWtvtPointLastData(Set<String> stcds, Date startTime, Date endTime) {
Map<String, Object> 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<WtvtPointDataVo> 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<Map<String, Date>> getInterValDate(Date startTime, Date endTime) {
int num = 7;
List<Map<String, Date>> dateList = new ArrayList<>();
long interval = DateUtil.betweenDay(startTime, endTime, false);
if (interval <= num) {
Map<String, Date> dateMap = new HashMap<>();
dateMap.put("startTime", startTime);
dateMap.put("endTime", endTime);
dateList.add(dateMap);
return dateList;
}
while (startTime.getTime() < endTime.getTime()) {
Map<String, Date> 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<String, Object> paramMap,
String column,
Set<String> values,
String prefix) {
if (CollUtil.isEmpty(values)) {
return;
}
List<String> 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(") ");
}
}

View File

@ -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;
}

View File

@ -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;
}

View File

@ -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;
}