Microservice/pqs
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

1.趋势图查询添加多线程处理

Browse Source 
2.变压器策略调整
This commit is contained in:
chendaofei
2025-11-11 14:33:16 +08:00
parent b7e3bb9796
commit e9ac024f73
3 changed files with 424 additions and 341 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
pqs-device/pq-device/pq-device-boot/src/main/java/com/njcn/device/pq/service/impl/PqsTransformerServiceImpl.java
View File

@@ -8,6 +8,7 @@ import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.njcn.common.pojo.enums.common.DataStateEnum;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.device.line.mapper.LineMapper;
import com.njcn.device.pq.enums.LineBaseEnum;
@@ -198,6 +199,59 @@ public class PqsTransformerServiceImpl extends ServiceImpl<PqsTransformerMapper,
PqsTflgass one = iPqsTflgassService.getOne(new LambdaQueryWrapper<PqsTflgass>()
.eq(PqsTflgass::getTfIndex, id));
List<String> arr = new ArrayList<>();
if (Objects.nonNull(one)) {
arr.add(one.getLogicBefore());
arr.add(one.getLogicNext());
}
voltage.put("bind", arr);
//1.获取当前变电站下，所拥有的母线条数
List<TerminalTree> downVoltage = lineMapper.getvoltage(subId, 1);
if(CollUtil.isNotEmpty(arr)) {
downVoltage.forEach(item->{
if(Objects.equals(item.getId(),arr.get(1))){
item.setLevel(LineBaseEnum.SUB_V_LEVEL.getCode());
}
});
}
//获取所有母线层级
List<Line> UpBusList = lineMapper.selectList(new LambdaQueryWrapper<Line>().eq(Line::getLevel, LineBaseEnum.SUB_V_LEVEL.getCode()).eq(Line::getState, DataStateEnum.ENABLE.getCode()).orderByAsc(Line::getSort));
List<TerminalTree> upTreeList = UpBusList.stream().map(it -> {
TerminalTree terminalTree = new TerminalTree();
terminalTree.setId(it.getId());
terminalTree.setName(it.getName());
terminalTree.setPid(it.getPids().split(StrUtil.COMMA)[LineBaseEnum.SUB_LEVEL.getCode()]);
terminalTree.setSort(it.getSort());
if(CollUtil.isNotEmpty(arr)) {
if (it.getId().equals(arr.get(0))) {
terminalTree.setLevel(LineBaseEnum.SUB_V_LEVEL.getCode());
}
}
return terminalTree;
}).collect(Collectors.toList());
//获取母线id
//List<String> ids = downVoltage.stream().map(TerminalTree::getId).collect(Collectors.toList());
List<TerminalTree> upList = TreedChildren(upTreeList, "变压器上节点设备树", subId);
voltage.put("upNode", upList);
List<TerminalTree> downList = TreedChildren(downVoltage, "变压器下节点设备树", subId);
voltage.put("downNode", downList);
return voltage;
}
public Map<String, Object> getGeneratrixOld(String subId, String id) {
Map<String, Object> voltage = new HashMap<>();
//先判断当前，变电站是否已添加了节点
PqsTflgass one = iPqsTflgassService.getOne(new LambdaQueryWrapper<PqsTflgass>()
.eq(PqsTflgass::getTfIndex, id));
//1.获取当前变电站下，所拥有的母线条数
List<TerminalTree> downVoltage = lineMapper.getvoltage(subId, 1);
//获取母线id
@@ -373,6 +427,7 @@ public class PqsTransformerServiceImpl extends ServiceImpl<PqsTransformerMapper,
return voltage;
}
@Override
public List<TerminalTree> getTransformerTree() {
List<TerminalTree> terminalTrees = this.baseMapper.listTransformer();

1
pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/controller/HistoryResultController.java
View File

@@ -61,6 +61,7 @@ public class HistoryResultController extends BaseController {
public HttpResult<List<HistoryDataResultVO>> getHistoryResult(@RequestBody @Validated HistoryParam historyParam) {
String methodDescribe = getMethodDescribe("getHistoryResult");
List<HistoryDataResultVO> list;
//针对河北现场特殊处理，pmswifi_开头监测点为无线装置，需要查询influxdb,其他测点查询oracle
if (HistoryDataSource == 1 && !historyParam.getLineId()[0].contains("pmswifi_")) {
list = oracleResultService.getHistoryResult(historyParam);
} else {

81
pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/impl/HistoryResultServiceImpl.java
View File

@@ -5,6 +5,7 @@ import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.njcn.common.pojo.constant.BizParamConstant;
@@ -54,6 +55,9 @@ import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
@@ -62,8 +66,11 @@ import java.math.BigDecimal;
import java.math.RoundingMode;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -102,16 +109,24 @@ public class HistoryResultServiceImpl implements HistoryResultService {
private final RStatDataVDMapper dataVDMapper;
private final RStatDataHarmRateVDMapper dataHarmRateVDMapper;
private final RStatDataHarmRateIDMapper dataHarmRateIDMapper;
@Autowired
@Qualifier("asyncExecutor")
private Executor asyncExecutor;
@Override
public List<HistoryDataResultVO> getHistoryResult(HistoryParam historyParam) {
System.out.println("begin============================="+ LocalDateTime.now());
TimeInterval timeInterval = new TimeInterval();
List<HistoryDataResultVO> historyDataResultVOList = new ArrayList<>();
//获取监测点
String[] points = historyParam.getLineId();
Integer number = 0;
for (int i = 0; i < points.length; i++) {
HistoryDataResultVO historyDataResultVO;
long a = timeInterval.intervalMs();
List<EventDetail> eventDetailList = eventDetailFeignClient.getEventDetailData(points[i], historyParam.getSearchBeginTime(), historyParam.getSearchEndTime()).getData();
List<EventDetailVO> eventDetailVOList = new ArrayList<>();
if (!eventDetailList.isEmpty()) {
for (EventDetail eventdetail : eventDetailList) {
@@ -122,9 +137,21 @@ public class HistoryResultServiceImpl implements HistoryResultService {
eventDetailVOList.add(eventDetailVO);
}
}
//获取监测点的数据单位，kV还是v kW还是w
PqsDeviceUnit pqsDeviceUnit = commTerminalGeneralClient.lineUnitDetail(points[i]).getData();
//获取监测点信息
LineDevGetDTO lineDetailDataVO = commTerminalGeneralClient.getMonitorDetail(points[i]).getData();
//获取限值
Overlimit overlimit = commTerminalGeneralClient.getOverLimitData(points[i]).getData();
long b = timeInterval.intervalMs();
System.out.println("feign调用："+(b-a));
//获取指标
String[] contions = historyParam.getCondition();
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (int j = 0; j < contions.length; j++) {
Integer number = 0;
if ("40".equals(contions[j]) || "41".equals(contions[j]) || "42".equals(contions[j]) || "43".equals(contions[j])
|| "44".equals(contions[j]) || "45".equals(contions[j]) || "50".equals(contions[j]) || "51".equals(contions[j])
|| "52".equals(contions[j])) {
@@ -133,20 +160,33 @@ public class HistoryResultServiceImpl implements HistoryResultService {
if ("46".equals(contions[j]) || "47".equals(contions[j]) || "48".equals(contions[j]) || "49".equals(contions[j])) {
number = historyParam.getInHarmonic();
}
historyDataResultVO = getCondition(historyParam.getSearchBeginTime(), historyParam.getSearchEndTime(), points[i], contions[j], number, historyParam.getValueType(), historyParam.getPtType());
/* if (points.length == 1 && j == 0) {
historyDataResultVO.setEventDetail(eventDetailVOList);
} else if (points.length > 1) {
historyDataResultVO.setEventDetail(eventDetailVOList);
}*/
historyDataResultVO.setEventDetail(eventDetailVOList);
historyDataResultVOList.add(historyDataResultVO);
//每个指标独立一个线程去查库，针对现场网络环境比较复杂的场景，正常场景看不太出来效果
asyncMethodWithCustomExecutor(contions[j],number,historyParam,eventDetailVOList,points[i],pqsDeviceUnit,lineDetailDataVO,overlimit,historyDataResultVOList,futures);
}
// 等待所有任务完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
}
}
System.out.println("end============================="+LocalDateTime.now());
System.out.println("共计耗时："+timeInterval.intervalMs()/1000.0);
return historyDataResultVOList;
}
//处理指标查询
public void asyncMethodWithCustomExecutor(String target,Integer number,HistoryParam historyParam,List<EventDetailVO> eventDetailVOList,String point,PqsDeviceUnit pqsDeviceUnit,LineDevGetDTO lineDetailDataVO,Overlimit overlimit,List<HistoryDataResultVO> historyDataResultVOList,List<CompletableFuture<Void>> futures) {
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
HistoryDataResultVO historyDataResultVO = getCondition(historyParam.getSearchBeginTime(), historyParam.getSearchEndTime(), point, target, number, historyParam.getValueType(), historyParam.getPtType(),pqsDeviceUnit,lineDetailDataVO,overlimit);
historyDataResultVO.setEventDetail(eventDetailVOList);
synchronized (historyDataResultVOList) { // 同步块保证线程安全
historyDataResultVOList.add(historyDataResultVO);
}
System.out.println("执行异步任务，线程名：" + Thread.currentThread().getName());
}, asyncExecutor);
futures.add(future);
}
@Override
public List<HistoryDataResultVO> getHistoryLineData(NormHistoryParam normHistoryParam) {
List<HistoryDataResultVO> historyDataResultVOList = new ArrayList<>();
@@ -160,9 +200,9 @@ public class HistoryResultServiceImpl implements HistoryResultService {
* 查询稳态数据分析
*/
@SneakyThrows
private HistoryDataResultVO getCondition(String startTime, String endTime, String lineId, String contion, Integer number, Integer valueType, Integer ptType) {
public HistoryDataResultVO getCondition(String startTime, String endTime, String lineId, String contion, Integer number, Integer valueType, Integer ptType,PqsDeviceUnit pqsDeviceUnit,LineDevGetDTO lineDetailDataVO,Overlimit overlimit) {
HistoryDataResultVO historyDataResultVO = new HistoryDataResultVO();
QueryResultLimitVO queryResultLimitVO = getQueryResult(startTime, endTime, lineId, contion, number, valueType, ptType);
QueryResultLimitVO queryResultLimitVO = getQueryResult(startTime, endTime, lineId, contion, number, valueType, ptType,pqsDeviceUnit,lineDetailDataVO,overlimit);
List<HarmonicHistoryData> harmonicHistoryDataList = queryResultLimitVO.getHarmonicHistoryDataList();
BeanUtil.copyProperties(queryResultLimitVO, historyDataResultVO);
//时间轴
@@ -207,7 +247,6 @@ public class HistoryResultServiceImpl implements HistoryResultService {
} else {
//按时间分组
Map<Instant, List<HarmonicHistoryData>> map = harmonicHistoryDataList.stream().collect(Collectors.groupingBy(HarmonicHistoryData::getTime, TreeMap::new, Collectors.toList()));
Float maxI = null;
Float minI = null;
for (Map.Entry<Instant, List<HarmonicHistoryData>> entry : map.entrySet()) {
@@ -241,13 +280,10 @@ public class HistoryResultServiceImpl implements HistoryResultService {
objects[3] = c;
maxI = max(maxI, c.floatValue());
minI = min(minI, c.floatValue());
}
}
List<Object> list = new ArrayList<>(Arrays.asList(objects));
objectListData.add(list);
}
historyDataResultVO.setMaxValue(maxI);
@@ -256,7 +292,6 @@ public class HistoryResultServiceImpl implements HistoryResultService {
historyDataResultVO.setLowerLimit(queryResultLimitVO.getLowerLimit());
historyDataResultVO.setValue(objectListData);
}
} else {
return historyDataResultVO;
}
@@ -284,19 +319,12 @@ public class HistoryResultServiceImpl implements HistoryResultService {
return ding;
}
private QueryResultLimitVO getQueryResult(String startTime, String endTime, String lineId, String contion, Integer number, Integer valueType, Integer ptType) {
PqsDeviceUnit pqsDeviceUnit = commTerminalGeneralClient.lineUnitDetail(lineId).getData();
private QueryResultLimitVO getQueryResult(String startTime, String endTime, String lineId, String contion, Integer number, Integer valueType, Integer ptType,PqsDeviceUnit pqsDeviceUnit,LineDevGetDTO lineDetailDataVO,Overlimit overlimit) {
QueryResultLimitVO queryResultLimitVO = new QueryResultLimitVO();
if (!lineId.isEmpty()) {
Float topLimit = 0f;
Float lowerLimit = 0f;
//获取监测点信息
LineDevGetDTO lineDetailDataVO = commTerminalGeneralClient.getMonitorDetail(lineId).getData();
//获取限值
Overlimit overlimit = commTerminalGeneralClient.getOverLimitData(lineId).getData();
if (Objects.isNull(overlimit)) {
//对配网没有限值的统一处理
DictData dictData = dicDataFeignClient.getDicDataById(lineDetailDataVO.getVoltageLevel()).getData();
@@ -305,7 +333,6 @@ public class HistoryResultServiceImpl implements HistoryResultService {
overlimit = COverlimitUtil.globalAssemble(voltageLevel, 10f, 10f, shortVal, 1, 1);
}
//组装sql语句
StringBuilder stringBuilder = new StringBuilder();
stringBuilder.append(InfluxDBTableConstant.TIME + " >= '").append(startTime).append(InfluxDBTableConstant.START_TIME).append("' and ").append(InfluxDBTableConstant.TIME).append(" <= '").append(endTime).append(InfluxDBTableConstant.END_TIME).append("' and (");