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

微调

Browse Source
This commit is contained in:
wr
2025-06-17 16:49:31 +08:00
parent 34d49847ab
commit 67bfba23eb
4 changed files with 79 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
migration-influxdb/migration-influxdb-read-boot/src/main/java/com/njcn/migration/read/controller/MigrationInfluxDBController.java
View File

@@ -87,4 +87,26 @@ public class MigrationInfluxDBController {
fileInputStream.close(); fileInputStream.close();
responseOutputStream.close(); responseOutputStream.close();
} }
@GetMapping(value = "/influxdbCs",produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
@ApiOperation(value ="influxdb数据同步->天数按小时进行分组同步测试", produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
public void influxdbCs() {
System.out.println("--------------------------------influxdb同步------------------------------------");
// 获取当前时间
LocalDateTime now = LocalDateTime.now();
// 减去2个小时
LocalDateTime oneHourAgo = now.minusHours(2);
// 将分钟和秒设置为0
LocalDateTime result = oneHourAgo.truncatedTo(ChronoUnit.HOURS);
// 加上59分钟59秒
LocalDateTime modifiedResult = result.plusMinutes(59).plusSeconds(59);
LineCountEvaluateParam param = new LineCountEvaluateParam();
param.setIsManual(false);
param.setStartTime(result.format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
param.setEndTime(modifiedResult.format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
migrationService.hourseLineDataBacthSysc(param);
migrationService.hourseDevDataBacthSysc(param);
}
} }

2
migration-influxdb/migration-influxdb-read-boot/src/main/java/com/njcn/migration/read/job/MigrationInfluxDBJob.java
View File

@@ -43,6 +43,7 @@ public class MigrationInfluxDBJob {
private final RmpEventDetailMapper detailMapper; private final RmpEventDetailMapper detailMapper;
@Scheduled(cron = "0 2 * * * ?") @Scheduled(cron = "0 2 * * * ?")
public void InfluxDBJob() { public void InfluxDBJob() {
System.out.println("--------------------------------influxdb同步------------------------------------");
// 获取当前时间 // 获取当前时间
LocalDateTime now = LocalDateTime.now(); LocalDateTime now = LocalDateTime.now();
// 减去2个小时 // 减去2个小时
@@ -62,6 +63,7 @@ public class MigrationInfluxDBJob {
@Scheduled(cron = "0 0 22 * * ?") @Scheduled(cron = "0 0 22 * * ?")
public void mapJob() throws IOException { public void mapJob() throws IOException {
System.out.println("--------------------------------excel文件同步------------------------------------");
File file = new File("/usr/local/jar/sj.xlsx"); File file = new File("/usr/local/jar/sj.xlsx");
List<LineTimeDto> excelDataV = EasyExcel.read(file) List<LineTimeDto> excelDataV = EasyExcel.read(file)
.head(LineTimeDto.class) .head(LineTimeDto.class)

56
migration-influxdb/migration-influxdb-read-boot/src/main/java/com/njcn/migration/read/service/impl/MigrationServiceImpl.java
View File

@@ -60,41 +60,45 @@ public class MigrationServiceImpl implements MigrationService {
final Integer[] num = {0}; final Integer[] num = {0};
map.forEach((lineId, time) -> { map.forEach((lineId, time) -> {
num[0] = num[0] + 1; num[0] = num[0] + 1;
System.out.println("当前总监测点数量"+size+"当前第"+num[0]+"：-》"+lineId+" "+param.getStartTime()+" "+param.getEndTime()+"剩余"+(size-num[0])); LineCountEvaluateParam evaluateParam = new LineCountEvaluateParam();
String format=null; evaluateParam.setLineId(Arrays.asList(lineId));
evaluateParam.setStartTime(param.getStartTime());
evaluateParam.setEndTime(param.getEndTime());
if (!param.getIsManual()) { if (!param.getIsManual()) {
if (StrUtil.isNotBlank(time)) { if (StrUtil.isNotBlank(time)) {
param.setStartTime(time); evaluateParam.setStartTime(time);
} }
} }
param.setLineId(Arrays.asList(lineId));
List<DataV> dataVS = dataV.listDataV(param); String format = null;
List<DataV> dataVS = dataV.listDataV(evaluateParam);
if (CollUtil.isNotEmpty(dataVS)) { if (CollUtil.isNotEmpty(dataVS)) {
if (!param.getIsManual()) { if (!param.getIsManual()) {
format = dataVS.get(0).getTimeId(); format = dataVS.get(0).getTimeId();
} }
migrationInsertFeignClient.insertDataV(dataVS); migrationInsertFeignClient.insertDataV(dataVS);
} }
migrationInsertFeignClient.insertDataFlicker(dataFlicker.listDataFlicker(param)); migrationInsertFeignClient.insertDataFlicker(dataFlicker.listDataFlicker(evaluateParam));
migrationInsertFeignClient.insertDataFluc(dataFluc.listDataFluc(param)); migrationInsertFeignClient.insertDataFluc(dataFluc.listDataFluc(evaluateParam));
migrationInsertFeignClient.insertDataHarmphasicI(dataHarmphasicI.listDataHarmphasicI(param)); migrationInsertFeignClient.insertDataHarmphasicI(dataHarmphasicI.listDataHarmphasicI(evaluateParam));
migrationInsertFeignClient.insertDataHarmphasicV(dataHarmphasicV.listDataHarmphasicV(param)); migrationInsertFeignClient.insertDataHarmphasicV(dataHarmphasicV.listDataHarmphasicV(evaluateParam));
migrationInsertFeignClient.insertDataHarmpowerP(dataHarmpowerP.listDataHarmpowerP(param)); migrationInsertFeignClient.insertDataHarmpowerP(dataHarmpowerP.listDataHarmpowerP(evaluateParam));
migrationInsertFeignClient.insertDataHarmpowerQ(dataHarmpowerQ.listDataHarmpowerQ(param)); migrationInsertFeignClient.insertDataHarmpowerQ(dataHarmpowerQ.listDataHarmpowerQ(evaluateParam));
migrationInsertFeignClient.insertDataHarmpowerS(dataHarmpowerS.listDataHarmpowerS(param)); migrationInsertFeignClient.insertDataHarmpowerS(dataHarmpowerS.listDataHarmpowerS(evaluateParam));
migrationInsertFeignClient.insertDataHarmrateI(dataHarmRateI.listDataHarmrateI(param)); migrationInsertFeignClient.insertDataHarmrateI(dataHarmRateI.listDataHarmrateI(evaluateParam));
migrationInsertFeignClient.insertDataHarmrateV(dataHarmRateV.listDataHarmrateV(param)); migrationInsertFeignClient.insertDataHarmrateV(dataHarmRateV.listDataHarmrateV(evaluateParam));
migrationInsertFeignClient.insertDataI(dataI.listDataI(param)); migrationInsertFeignClient.insertDataI(dataI.listDataI(evaluateParam));
migrationInsertFeignClient.insertDataInharmI(dataInharmI.listDataInharmI(param)); migrationInsertFeignClient.insertDataInharmI(dataInharmI.listDataInharmI(evaluateParam));
migrationInsertFeignClient.insertDataInharmV(dataInharmV.listDataInharmV(param)); migrationInsertFeignClient.insertDataInharmV(dataInharmV.listDataInharmV(evaluateParam));
migrationInsertFeignClient.insertDataPlt(dataPlt.listDataPlt(param)); migrationInsertFeignClient.insertDataPlt(dataPlt.listDataPlt(evaluateParam));
migrationInsertFeignClient.batchInsertion(eventDetail.getRawData(param)); migrationInsertFeignClient.batchInsertion(eventDetail.getRawData(evaluateParam));
System.out.println("定时任务_当前总监测点数量" + size + "当前第" + num[0] + "：-》" + lineId + " 文件时间 " + time + " |最新数据时间" + format + "| " + evaluateParam.getStartTime() + " " + evaluateParam.getEndTime() + "剩余" + (size - num[0]));
if (!param.getIsManual() && StrUtil.isNotBlank(format)) { if (!param.getIsManual() && StrUtil.isNotBlank(format)) {
TimeUtil.putLineTime(lineId, format); TimeUtil.putLineTime(lineId, format);
} }
}); });
System.gc();
} }
@Override @Override
@@ -105,30 +109,32 @@ public class MigrationServiceImpl implements MigrationService {
final Integer[] num = {0}; final Integer[] num = {0};
map.forEach((lineId, time) -> { map.forEach((lineId, time) -> {
num[0] = num[0] + 1; num[0] = num[0] + 1;
System.out.println("当前总终端数量"+size+"当前第"+num[0]+"：-》"+lineId+" "+param.getStartTime()+" "+param.getEndTime()+"剩余"+(size-num[0])); LineCountEvaluateParam evaluateParam = new LineCountEvaluateParam();
evaluateParam.setLineId(Arrays.asList(lineId));
evaluateParam.setStartTime(param.getStartTime());
evaluateParam.setEndTime(param.getEndTime());
String format = null; String format = null;
if (!param.getIsManual()) { if (!param.getIsManual()) {
if (StrUtil.isNotBlank(time)) { if (StrUtil.isNotBlank(time)) {
param.setStartTime(time); param.setStartTime(time);
} }
} }
param.setLineId(Arrays.asList(lineId)); List<PqsCommunicate> pqsCommunicates = pqsCommunicate.listPqsCommunicate(evaluateParam);
List<PqsCommunicate> pqsCommunicates = pqsCommunicate.listPqsCommunicate(param);
if (CollUtil.isNotEmpty(pqsCommunicates)) { if (CollUtil.isNotEmpty(pqsCommunicates)) {
if (!param.getIsManual()) { if (!param.getIsManual()) {
format = pqsCommunicates.get(0).getTimeId(); format = pqsCommunicates.get(0).getTimeId();
} }
migrationInsertFeignClient.insertPqsCommunicate(pqsCommunicates); migrationInsertFeignClient.insertPqsCommunicate(pqsCommunicates);
} }
System.out.println("当前总终端数量" + size + "当前第" + num[0] + "：-》" + lineId + " 文件时间 " + time + " |最新数据时间" + format + "| " + evaluateParam.getStartTime() + " " + evaluateParam.getEndTime() + "剩余" + (size - num[0]));
if (!param.getIsManual() && StrUtil.isNotBlank(format)) { if (!param.getIsManual() && StrUtil.isNotBlank(format)) {
TimeUtil.putDevTime(lineId, format); TimeUtil.putDevTime(lineId, format);
} }
}); });
System.gc();
} }
@Override @Override
@Async("asyncInfluxDBExecutor")
public void initializeExcel() { public void initializeExcel() {
File file = new File("/usr/local/jar/sj.xlsx"); File file = new File("/usr/local/jar/sj.xlsx");
Map<String, String> map = TimeUtil.getLineMap(); Map<String, String> map = TimeUtil.getLineMap();

2
migration-influxdb/migration-influxdb-read-boot/src/main/java/com/njcn/migration/read/util/TimeUtil.java
View File

@@ -132,5 +132,7 @@ public class TimeUtil {
excelWriter.finish(); excelWriter.finish();
TimeUtil.putAllLineTime(excelDataV.stream().collect(Collectors.toMap(LineTimeDto::getLineId, LineTimeDto::getTimeData))); TimeUtil.putAllLineTime(excelDataV.stream().collect(Collectors.toMap(LineTimeDto::getLineId, LineTimeDto::getTimeData)));
TimeUtil.putAllDevTime(excelCommunicates.stream().collect(Collectors.toMap(DevTimeDto::getDevId, DevTimeDto::getTimeData))); TimeUtil.putAllDevTime(excelCommunicates.stream().collect(Collectors.toMap(DevTimeDto::getDevId, DevTimeDto::getTimeData)));
System.out.println(" line "+TimeUtil.getLineMap());
System.out.println(" dev "+TimeUtil.getDevMap());
} }
} }