Middleware/cn-data-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

修改最新数据来了不更新通讯表数据，通讯表由前置状态维护

Browse Source
This commit is contained in:
hzj
2026-06-10 10:05:57 +08:00
parent 5cb4ab0e8f
commit 5090e023b1
3 changed files with 29 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/pojo/dto/PqsCommunicateDto.java
View File

@@ -20,6 +20,9 @@ public class PqsCommunicateDto {
private Integer type;
//是否更新updateTime标志；数据上送更新1，状态翻转不更新0
private Integer flag=0;
//是否更新、pqs_communicate标志，0更新，1不更新
private Integer updateCommunicateFlag=0;
}

1
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/LnDataDealServiceImpl.java
View File

@@ -137,6 +137,7 @@ public class LnDataDealServiceImpl implements LnDataDealService {
pqsCommunicateDto.setDevId(temp);
pqsCommunicateDto.setType(1);
pqsCommunicateDto.setFlag(1);
pqsCommunicateDto.setUpdateCommunicateFlag(1);
updatefutures.add(CompletableFuture.runAsync(() -> iPqsCommunicate.insertion(pqsCommunicateDto), executor));

45
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbPqsCommunicateImpl.java
View File

@@ -112,28 +112,33 @@ public class InfluxdbPqsCommunicateImpl implements IPqsCommunicate {
public void insertion(PqsCommunicateDto pqsCommunicateDto) {
// log.info("进出Influxdb实现类");
//获取最新一条数据
long start = System.currentTimeMillis();
if(Objects.equals(pqsCommunicateDto.getUpdateCommunicateFlag(),0)){
long start = System.currentTimeMillis();
PqsCommunicate dto = new PqsCommunicate();
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(PqsCommunicate.class);
influxQueryWrapper.eq(PqsCommunicate::getDevId,pqsCommunicateDto.getDevId()).timeDesc().limit(1);
List<PqsCommunicate> pqsCommunicates = pqsCommunicateMapper.selectByQueryWrapper(influxQueryWrapper);
PqsCommunicate dto = new PqsCommunicate();
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(PqsCommunicate.class);
influxQueryWrapper.eq(PqsCommunicate::getDevId,pqsCommunicateDto.getDevId()).timeDesc().limit(1);
List<PqsCommunicate> pqsCommunicates = pqsCommunicateMapper.selectByQueryWrapper(influxQueryWrapper);
long end1 = System.currentTimeMillis();
long end1 = System.currentTimeMillis();
log.info("查询通讯表最新数据：" + (end1 - start) + " ms");
PqsCommunicate pqsCommunicate = new PqsCommunicate();
pqsCommunicate.setTime(LocalDateTime.parse(pqsCommunicateDto.getTime(), DATE_TIME_FORMATTER).atZone(ZoneId.systemDefault()).toInstant());
pqsCommunicate.setDevId(pqsCommunicateDto.getDevId());
pqsCommunicate.setType(pqsCommunicateDto.getType());
//如果不存数据或者状态不一样则插入数据
//可能存在掉线后最后一组数据还未入库，添加时间判断
if(CollectionUtils.isEmpty(pqsCommunicates)|| (!Objects.equals( pqsCommunicates.get(0).getType(),pqsCommunicateDto.getType())&&pqsCommunicates.get(0).getTime().isBefore(pqsCommunicate.getTime()))){
pqsCommunicateMapper.insertOne(pqsCommunicate);
log.info("查询通讯表最新数据：" + (end1 - start) + " ms");
PqsCommunicate pqsCommunicate = new PqsCommunicate();
pqsCommunicate.setTime(LocalDateTime.parse(pqsCommunicateDto.getTime(), DATE_TIME_FORMATTER).atZone(ZoneId.systemDefault()).toInstant());
pqsCommunicate.setDevId(pqsCommunicateDto.getDevId());
pqsCommunicate.setType(pqsCommunicateDto.getType());
//如果不存数据或者状态不一样则插入数据
//可能存在掉线后最后一组数据还未入库，添加时间判断
if(CollectionUtils.isEmpty(pqsCommunicates)|| (!Objects.equals( pqsCommunicates.get(0).getType(),pqsCommunicateDto.getType())&&pqsCommunicates.get(0).getTime().isBefore(pqsCommunicate.getTime()))){
pqsCommunicateMapper.insertOne(pqsCommunicate);
}
long end2 = System.currentTimeMillis();
log.info("更新通讯表最新数据：" + (end2 - end1) + " ms");
}
long end3 = System.currentTimeMillis();
}
long end2 = System.currentTimeMillis();
log.info("更新通讯表最新数据：" + (end2 - end1) + " ms");
//更新mysql数据
DevComFlagDTO devComFlagDTO = new DevComFlagDTO();
@@ -144,8 +149,8 @@ public class InfluxdbPqsCommunicateImpl implements IPqsCommunicate {
devComFlagDTO.setStatus(pqsCommunicateDto.getType());
deviceFeignClient.updateDevComFlag(devComFlagDTO);
long end3 = System.currentTimeMillis();
log.info("更新device表状态和时间：" + (end3 - end2) + " ms");
long end4 = System.currentTimeMillis();
log.info("更新device表状态和时间：" + (end4 - end3) + " ms");
}