Other/data-migration
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.oracle同步influxdb代码,监测点运行中断状态

Browse Source
This commit is contained in:
wr
2024-09-29 16:22:34 +08:00
parent 53cc3c85e3
commit 401d2a4e97
10 changed files with 288 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
influx-data/influx-source/src/main/java/com/njcn/influx/bo/param/TableEnum.java
View File

@@ -28,7 +28,9 @@ public enum TableEnum {
DATAINHARMV("DataInharmV","电压间谐波幅值数据表", 4),
DATAI("DataI","谐波电流幅值数据表", 4),
DATAPLT("DataPlt","长时闪变数据表", 1),
DATAV("DataV","谐波电压幅值数据表", 4);
DATAV("DataV","谐波电压幅值数据表", 4),
COMINFORMATION("ComInfoRmation","监测点状态监测数据", 4),
;
private final String code;

53
influx-data/influx-source/src/main/java/com/njcn/influx/bo/po/InfluxDBComInfoRmation.java Normal file
View File

@@ -0,0 +1,53 @@
package com.njcn.influx.bo.po;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import com.njcn.influx.utils.InstantDateSerializer;
import com.njcn.oracle.bo.po.ComInfoRmation;
import lombok.Data;
import org.influxdb.annotation.Column;
import org.influxdb.annotation.Measurement;
import org.influxdb.annotation.TimeColumn;
import java.time.Instant;
import java.time.ZoneId;
/**
* 类的介绍：
*
* @author xuyang
* @version 1.0.0
* @createTime 2022/7/12 9:55
*/
@Data
@Measurement(name = "pqs_communicate")
public class InfluxDBComInfoRmation {
@TimeColumn
@Column(name = "time")
@JsonSerialize(using = InstantDateSerializer.class)
private Instant time;
@Column(name = "dev_id")
private String devId;
@Column(name = "description")
private String description;
@Column(name = "type")
private Integer type;
public static InfluxDBComInfoRmation oralceToInfluxDB(ComInfoRmation comInfoRmation) {
if (comInfoRmation == null) {
return null;
}
InfluxDBComInfoRmation influxDBDataCommunicate = new InfluxDBComInfoRmation();
Instant instant = comInfoRmation.getUpdateTime().atZone(ZoneId.systemDefault()).toInstant();
influxDBDataCommunicate.setTime(instant);
influxDBDataCommunicate.setDevId(comInfoRmation.getLineIndex());
influxDBDataCommunicate.setDescription(comInfoRmation.getDescription());
influxDBDataCommunicate.setType(comInfoRmation.getType());
return influxDBDataCommunicate;
}
}

7
influx-data/influx-source/src/main/java/com/njcn/influx/imapper/InfluxDBComInfoRmationMapper.java Normal file
View File

@@ -0,0 +1,7 @@
package com.njcn.influx.imapper;
import com.njcn.influx.base.InfluxDbBaseMapper;
import com.njcn.influx.bo.po.InfluxDBComInfoRmation;
public interface InfluxDBComInfoRmationMapper extends InfluxDbBaseMapper<InfluxDBComInfoRmation> {
}

94
influx-data/influx-source/src/main/java/com/njcn/influx/service/impl/OracleToInfluxDBServiceImpl.java
View File

@@ -1,9 +1,11 @@
package com.njcn.influx.service.impl;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.core.text.StrPool;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.njcn.influx.bo.param.TableEnum;
import com.njcn.influx.bo.po.JobDetailHoursInfluxDB;
@@ -245,36 +247,68 @@ public class OracleToInfluxDBServiceImpl implements OracleToInfluxDBService {
StopWatch stopWatch = new StopWatch();
stopWatch.start();
List list = new ArrayList(Collections.emptyList());
//获取监测点最新的数据时间，单监测点查询数据
List<lineTimeDto> lineTimeList = lineTimeMapper.getLineTime();
lineTimeList.forEach(item->{
if("ComInfoRmation".equals(tableName)){
MigrationParam migration = new MigrationParam();
migration.setLineIds(Collections.singletonList(item.getLineIndex()));
migration.setStartTime(item.getUpdateTime().minusHours(2));
migration.setEndTime(item.getUpdateTime());
System.out.println("当前监测点为------------------------------------"+item.getLineIndex());
System.out.println("执行扫描起始时间------------------------------------"+item.getUpdateTime().minusHours(2).format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
System.out.println("执行扫描结束时间------------------------------------"+item.getUpdateTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
migration.setStartTime(dataAsynParam.getStartDateTime());
migration.setEndTime(dataAsynParam.getEndDateTime());
System.out.println("执行扫描起始时间------------------------------------"+dataAsynParam.getStartDateTime().minusHours(2).format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
System.out.println("执行扫描结束时间------------------------------------"+dataAsynParam.getEndDateTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
list.addAll(executor.queryData(migration));
});
System.out.println("查询到的数据++++++++++++++"+list.size());
//反射獲取linid的值并把linid的值替换成mysql对应的linid，并记录未匹配的lineid
Iterator iterator = list.iterator();
while (iterator.hasNext()) {
try{
Object obj = iterator.next();
//获取
Field id = obj.getClass().getDeclaredField("lineid");
id.setAccessible(true); //暴力访问id
String id1 = id.get(obj).toString();
if (!IdMappingCache.LineIdMapping.containsKey(id1)){
log.info(tableName+"表---Oralcet数据同步到InfluxDB未找mysql中到lineid匹配的lineid"+id1);
iterator.remove();
}else {
id.set(obj, IdMappingCache.LineIdMapping.get(id1));
System.out.println("查询到的数据++++++++++++++"+list.size());
//反射获取linid的值并把linid的值替换成mysql对应的devid，并记录未匹配的devid
Iterator iterator = list.iterator();
while (iterator.hasNext()) {
try{
Object obj = iterator.next();
//获取
Field id = obj.getClass().getDeclaredField("lineIndex");
id.setAccessible(true); //暴力访问id
Object o = id.get(obj);
if(ObjectUtil.isNotNull(o)){
int index = Integer.parseInt(o.toString())/10;
if (!IdMappingCache.DevIdMapping.containsKey(index+"")){
log.info(tableName+"表---Oralcet数据同步到InfluxDB未找mysql中到devid匹配的devid"+index);
iterator.remove();
}else {
id.set(obj, IdMappingCache.DevIdMapping.get(index+""));
}
}
}catch (Exception e){
e.printStackTrace();
}
}
}else{
//获取监测点最新的数据时间，单监测点查询数据
List<lineTimeDto> lineTimeList = lineTimeMapper.getLineTime();
lineTimeList.forEach(item->{
MigrationParam migration = new MigrationParam();
migration.setLineIds(Collections.singletonList(item.getLineIndex()));
migration.setStartTime(item.getUpdateTime().minusHours(2));
migration.setEndTime(item.getUpdateTime());
System.out.println("当前监测点为------------------------------------"+item.getLineIndex());
System.out.println("执行扫描起始时间------------------------------------"+item.getUpdateTime().minusHours(2).format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
System.out.println("执行扫描结束时间------------------------------------"+item.getUpdateTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
list.addAll(executor.queryData(migration));
});
System.out.println("查询到的数据++++++++++++++"+list.size());
//反射獲取linid的值并把linid的值替换成mysql对应的linid，并记录未匹配的lineid
Iterator iterator = list.iterator();
while (iterator.hasNext()) {
try{
Object obj = iterator.next();
//获取
Field id = obj.getClass().getDeclaredField("lineid");
id.setAccessible(true); //暴力访问id
String id1 = id.get(obj).toString();
if (!IdMappingCache.LineIdMapping.containsKey(id1)){
log.info(tableName+"表---Oralcet数据同步到InfluxDB未找mysql中到lineid匹配的lineid"+id1);
iterator.remove();
}else {
id.set(obj, IdMappingCache.LineIdMapping.get(id1));
}
}catch (Exception e){
e.printStackTrace();
}
}catch (Exception e){
e.printStackTrace();
}
}
//采用弱引用接受，后续手动调用gc后，会清空该对象
@@ -351,7 +385,11 @@ public class OracleToInfluxDBServiceImpl implements OracleToInfluxDBService {
DataAsynParam dataAsynParam1 = new DataAsynParam();
dataAsynParam1.setEndDateTime(startDateTime1.minusHours(-1).minusSeconds(1));
dataAsynParam1.setStartDateTime(startDateTime1);
dataAsynParam1.setTableNames(TableEnum.getExecutableTypes());
if(CollUtil.isEmpty(dataAsynParam.getTableNames())){
dataAsynParam1.setTableNames(TableEnum.getExecutableTypes());
}else{
dataAsynParam1.setTableNames(dataAsynParam.getTableNames());
}
log.info("执行"+startDateTime1+"时刻数据");
this.hourseDataBacthSysc(dataAsynParam1);
}