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

oralce同步到influxDB

Browse Source
This commit is contained in:
hzj
2024-01-16 15:51:00 +08:00
parent a2afabe80c
commit aa23ed6199
22 changed files with 75111 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1,65 @@
package com.njcn.influx.bo.param;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
/**
* Description:
* Date: 2024/1/4 13:54【需求编号】
*
* @author clam
* @version V1.0.0
*/
public enum TableEnum {
DATAFLICKER("DataFlicker","电压闪变数据表"),
DATAFLUC("DataFluc","电压波动数据表"),
DATAHARMPHASICI("DataHarmphasicI","谐波电流角度数据表"),
DATAHARMPHASICV("DataHarmphasicV","谐波电压角度数据表"),
DATAHARMPOWERP("DataHarmpowerP","有功功率数据表"),
DATAHARMPOWERQ("DataHarmpowerQ","无功功率数据表"),
DATAHARMPOWERS("DataHarmpowerS","视在功率数据表"),
DATAHARMRATEI("DataHarmrateI","谐波电流含有率数据表"),
DATAHARMRATEV("DataHarmrateV","谐波电压含有率数据表"),
DATAINHARMI("DataInharmI","电流简谐波幅值数据表"),
DATAINHARMV("DataInharmV","电压间谐波幅值数据表"),
DATAI("DataI","谐波电流幅值数据表"),
DATAPLT("DataPlt","长时闪变数据表"),
DATAV("DataV","谐波电压幅值数据表");
private final String code;
private final String value;
TableEnum(String code, String value) {
this.code = code;
this.value = value;
}
/**
* 获取所有可执行的组件类型，
* 除了"目录"类型，其他都可以执行
* @return
*/
public static List<String> getExecutableTypes() {
return Arrays.stream(TableEnum.values()).map(tmep->{
return tmep.code;
}).collect(Collectors.toList());
}
/**
* 通过code获取枚举值
* @param code
* @return
*/
public static String getValueByCode(String code) {
for (TableEnum item : TableEnum.values()) {
if (item.code.equals(code)) {
return item.value;
}
}
return null;
}
}

8
influx-data/influx-source/src/main/java/com/njcn/influx/bo/po/InfluxDBDataV.java
View File

@@ -247,7 +247,7 @@ public class InfluxDBDataV {
influxDBDataV.setPhasicType(dataV.getPhasicType());
influxDBDataV.setQualityFlag(dataV.getQualityflag()+"");
influxDBDataV.setValueType(valueType);
//T相时处理setRmsLvr
//T相时特殊处理RmsLvr将Rmsab，Rmsbc，Rmsca转纯的abc3相里
if(Objects.equals(dataV.getPhasicType(),"T")){
InfluxDBDataV influxDBDataVA = new InfluxDBDataV();
@@ -261,8 +261,6 @@ public class InfluxDBDataV {
influxDBDataVC.setPhasicType("C");
if (valueType.equals("AVG")) {
influxDBDataVA.setRmsLvr(dataV.getRmsab());
@@ -296,7 +294,7 @@ public class InfluxDBDataV {
influxDBDataVList.add(influxDBDataVC);
}else {
}
if (valueType.equals("AVG")) {
influxDBDataV.setVNeg(dataV.getVNeg());
@@ -557,7 +555,7 @@ public class InfluxDBDataV {
influxDBDataV.setV49(dataV.getV49Cp95());
influxDBDataV.setV50(dataV.getV50Cp95());
}
}

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

@@ -0,0 +1,76 @@
package com.njcn.influx.bo.po;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableName;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.github.jeffreyning.mybatisplus.anno.MppMultiId;
import lombok.Data;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Objects;
/**
*
* Description:
* Date: 2024/1/8 12:27【需求编号】
*
* @author clam
* @version V1.0.0
*/
@TableName(value = "JOB_DETAIL_INFLUXDB")
@Data
public class JobDetailInfluxDB {
/**
* 指标表名
*/
@MppMultiId(value = "TABLE_NAME")
private String tableName;
@TableField(exist = false)
private String tableName_CN;
/**
* 起始时间
*/
@JsonFormat(
pattern = "yyyy-MM-dd"
)
@MppMultiId(value = "EXCUTE_DATE")
private LocalDate excuteDate;
/**
* 状态（0-执行中、1-成功、2-失败,3未执行）
*/
@TableField(value = "STATE")
private Integer state;
/**
* 记录数
*/
@TableField(value = "\"ROW_COUNT\"")
private Integer rowCount;
@JsonFormat(
pattern = "yyyy-MM-dd HH:mm:ss"
)
@TableField(value = "UPDATE_TIME")
private LocalDateTime updateTime;
@TableField(value = "DURATION")
private Double duration;
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
JobDetailInfluxDB jobDetail = (JobDetailInfluxDB) o;
return Objects.equals(tableName, jobDetail.tableName) && Objects.equals(excuteDate, jobDetail.excuteDate);
}
}

18
influx-data/influx-source/src/main/java/com/njcn/influx/mapper/JobDetailInfluxDBMapper.java Normal file
View File

@@ -0,0 +1,18 @@
package com.njcn.influx.mapper;
import com.github.jeffreyning.mybatisplus.base.MppBaseMapper;
import com.njcn.influx.bo.po.JobDetailInfluxDB;
import java.util.List;
/**
*
* Description:
* Date: 2024/1/8 12:27【需求编号】
*
* @author clam
* @version V1.0.0
*/
public interface JobDetailInfluxDBMapper extends MppBaseMapper<JobDetailInfluxDB> {
}

2
influx-data/influx-source/src/main/java/com/njcn/influx/service/InfluxDBBaseService.java
View File

@@ -13,7 +13,7 @@ public interface InfluxDBBaseService<T> {
/***
* 当批量插入数据量过大时，可以指定尺寸交给mybatis,每次插入多少条记录
* 当批量插入数据量过大时，可以指定尺寸交，influxdb最大5000行
* @author hongawen
* @date 2023/6/6 9:33
* @param data 数据集合

22
influx-data/influx-source/src/main/java/com/njcn/influx/service/JobDetailInfluxDBService.java Normal file
View File

@@ -0,0 +1,22 @@
package com.njcn.influx.service;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.github.jeffreyning.mybatisplus.service.IMppService;
import com.njcn.influx.bo.po.JobDetailInfluxDB;
import com.njcn.oracle.bo.param.JobQueryParam;
/**
* Description:
* Date: 2024/1/8 12:27【需求编号】
*
* @author clam
* @version V1.0.0
*/
public interface JobDetailInfluxDBService extends IMppService<JobDetailInfluxDB> {
JobDetailInfluxDB select(JobDetailInfluxDB jobDetail);
IPage<JobDetailInfluxDB> selectByParam(JobQueryParam jobQueryParam);
boolean jobRemove(JobDetailInfluxDB jobDetail);
}

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

@@ -0,0 +1,7 @@
package com.njcn.influx.service;
import com.njcn.oracle.bo.param.DataAsynParam;
public interface OracleToInfluxDBService {
void dataBacthSysc(DataAsynParam dataAsynParam);
}

5
influx-data/influx-source/src/main/java/com/njcn/influx/service/impl/InfluxDBBaseServiceImpl.java
View File

@@ -7,6 +7,7 @@ import org.influxdb.InfluxDB;
import org.influxdb.annotation.Measurement;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
@@ -24,8 +25,8 @@ import java.util.List;
public class InfluxDBBaseServiceImpl<T> implements InfluxDBBaseService<T> {
private final InfluxDB influxDb;
private final String database="pqsbase_zl";
@Value("${spring.influx.database}")
private String database;
@Override
public void insertBatchBySlice(List<T> data, int size) {
int totalCount = data.size();

92
influx-data/influx-source/src/main/java/com/njcn/influx/service/impl/JobDetailInfluxDBServiceImpl.java Normal file
View File

@@ -0,0 +1,92 @@
package com.njcn.influx.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
import com.njcn.influx.bo.po.JobDetailInfluxDB;
import com.njcn.influx.mapper.JobDetailInfluxDBMapper;
import com.njcn.influx.service.JobDetailInfluxDBService;
import com.njcn.oracle.bo.param.JobQueryParam;
import com.njcn.oracle.bo.param.ServiceTypeEnum;
import com.njcn.oracle.util.LocalDateUtil;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import java.time.LocalDate;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
/**
* Description:
* Date: 2024/1/8 12:27【需求编号】
*
* @author clam
* @version V1.0.0
*/
@Service
public class JobDetailInfluxDBServiceImpl extends MppServiceImpl<JobDetailInfluxDBMapper, JobDetailInfluxDB> implements JobDetailInfluxDBService {
@Override
public JobDetailInfluxDB select(JobDetailInfluxDB jobDetail) {
return this.lambdaQuery()
.eq(JobDetailInfluxDB::getTableName, jobDetail.getTableName())
.eq(JobDetailInfluxDB::getExcuteDate, jobDetail.getExcuteDate())
.one();
}
@Override
public IPage<JobDetailInfluxDB> selectByParam(JobQueryParam jobQueryParam) {
IPage<JobDetailInfluxDB> page = new Page<>(jobQueryParam.getCurrentPage(), jobQueryParam.getPageSize());
QueryWrapper<JobDetailInfluxDB> queryWrapper = new QueryWrapper<>();
queryWrapper.lambda().between(JobDetailInfluxDB::getExcuteDate, jobQueryParam.getStartTime(), jobQueryParam.getEndTime()).
in(!CollectionUtils.isEmpty(jobQueryParam.getStates()), JobDetailInfluxDB::getState, jobQueryParam.getStates()).
in(!CollectionUtils.isEmpty(jobQueryParam.getTableNames()), JobDetailInfluxDB::getTableName, jobQueryParam.getTableNames());
List<JobDetailInfluxDB> jobDetails = this.getBaseMapper().selectList(queryWrapper);
List<String> tableNames;
if (CollectionUtils.isEmpty(jobQueryParam.getTableNames())) {
tableNames = ServiceTypeEnum.getExecutableTypes();
} else {
tableNames = jobQueryParam.getTableNames();
}
List<LocalDate> dateList = LocalDateUtil.getDateList(jobQueryParam.getStartTime(), jobQueryParam.getEndTime());
List<String> finalTableNames = tableNames;
dateList.stream().forEach(localDate -> {
finalTableNames.stream().forEach(temp -> {
JobDetailInfluxDB jobDetail = new JobDetailInfluxDB();
jobDetail.setTableName(temp);
jobDetail.setExcuteDate(localDate);
if (jobDetails.contains(jobDetail)) {
return;
}
jobDetail.setTableName_CN(ServiceTypeEnum.getValueByCode(temp));
jobDetail.setState(3);
jobDetails.add(jobDetail);
});
});
List<JobDetailInfluxDB> jobDetailIPage = jobDetails.stream().filter(temp->jobQueryParam.getStates().contains(temp.getState()+"")).sorted(Comparator.comparing(JobDetailInfluxDB::getExcuteDate).reversed()).
skip((jobQueryParam.getCurrentPage() - 1) * jobQueryParam.getPageSize()).
limit(jobQueryParam.getPageSize()).collect(Collectors.toList());
jobDetailIPage.stream().forEach(temp -> temp.setTableName_CN(ServiceTypeEnum.getValueByCode(temp.getTableName())));
page.setRecords(jobDetailIPage);
page.setTotal(Long.parseLong(jobDetails.size() + ""));
return page;
}
@Override
public boolean jobRemove(JobDetailInfluxDB jobDetail) {
QueryWrapper<JobDetailInfluxDB> queryWrapper = new QueryWrapper<>();
queryWrapper.lambda()
.eq(JobDetailInfluxDB::getTableName, jobDetail.getTableName())
.eq(JobDetailInfluxDB::getExcuteDate, jobDetail.getExcuteDate());
return this.remove(queryWrapper);
}
}

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

@@ -0,0 +1,184 @@
package com.njcn.influx.service.impl;
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.extra.spring.SpringUtil;
import com.njcn.influx.bo.po.JobDetailInfluxDB;
import com.njcn.influx.service.JobDetailInfluxDBService;
import com.njcn.influx.service.OracleToInfluxDBService;
import com.njcn.oracle.bo.param.DataAsynParam;
import com.njcn.oracle.bo.param.MigrationParam;
import com.njcn.oracle.bo.po.JobDetail;
import com.njcn.oracle.mybatis.service.IReplenishMybatisService;
import com.njcn.oracle.service.JobDetailService;
import com.njcn.oracle.util.LocalDateUtil;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;
import java.lang.ref.WeakReference;
import java.lang.reflect.Method;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* Description:
* Date: 2024/1/16 14:49【需求编号】
*
* @author clam
* @version V1.0.0
*/
@Service
@RequiredArgsConstructor
public class OracleToInfluxDBServiceImpl implements OracleToInfluxDBService {
private final static String PACKAGE_PREFIX = "com.njcn.oracle.service.impl.";
private final static String PACKAGE_SUFFIX = "ServiceImpl";
private final JobDetailInfluxDBService jobDetailInfluxDBService;
private final InfluxDBBaseServiceImpl influxDBBaseService;
@Value("${business.slice:2}")
private int slice;
@Override
@Async
public void dataBacthSysc(DataAsynParam dataAsynParam) {
Runtime runtime = Runtime.getRuntime();
System.out.println("开始执行前总堆内存为：" + runtime.totalMemory() / (1024 * 1024) + " MB");
System.out.println("开始执行前已用堆内存为：" + (runtime.totalMemory() - runtime.freeMemory()) / (1024 * 1024) + " MB");
System.out.println("开始执行前空闲堆内存为：" + runtime.freeMemory() / (1024 * 1024) + " MB");
//目前且作为2片，后续将该属性提取到配置文件中
List<String> tableNames = dataAsynParam.getTableNames();
//嵌套循环，先循环指标，再循环日期
tableNames.stream().forEach(tableName -> {
IReplenishMybatisService executor;
try {
executor = (IReplenishMybatisService) SpringUtil.getBean(Class.forName(PACKAGE_PREFIX + tableName + PACKAGE_SUFFIX));
} catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
List<LocalDate> dateList = LocalDateUtil.getDateList(dataAsynParam.getStartTime(), dataAsynParam.getEndTime());
dateList.forEach(date -> {
//程序监听
StopWatch stopWatch = new StopWatch();
stopWatch.start();
//分片下，每段时间的小时数
int sliceHour = 24 / slice;
for (int i = 0; i < slice; i++) {
String dateStr = LocalDateTimeUtil.format(date, DatePattern.NORM_DATE_PATTERN);
LocalDateTime startTime = LocalDateTimeUtil.parse(dateStr + StrPool.C_SPACE + (sliceHour * i < 10 ? "0" + sliceHour * i : sliceHour * i) + ":00:00", DatePattern.NORM_DATETIME_PATTERN);
LocalDateTime endTime = LocalDateTimeUtil.parse(dateStr + StrPool.C_SPACE + (sliceHour * (i + 1) - 1 < 10 ? "0" + (sliceHour * (i + 1) - 1) : sliceHour * (i + 1) - 1) + ":59:59", DatePattern.NORM_DATETIME_PATTERN);
//查询该时区的数据，并准备入库
MigrationParam migration = new MigrationParam();
migration.setStartTime(startTime);
migration.setEndTime(endTime);
//采用弱引用接受，后续手动调用gc后，会清空该对象
WeakReference<List> weakReferenceData = new WeakReference<>(executor.queryData(migration));
int size = 0;
if (CollectionUtil.isNotEmpty(weakReferenceData.get())) {
size = weakReferenceData.get().size();
}
System.out.println(tableName + "查到" + size + "条数据后总堆内存为：" + runtime.totalMemory() / (1024 * 1024) + " MB");
System.out.println(tableName + "查到" + size + "条数据后已用堆内存为：" + (runtime.totalMemory() - runtime.freeMemory()) / (1024 * 1024) + " MB");
System.out.println(tableName + "查到" + size + "条数据后空闲堆内存为：" + runtime.freeMemory() / (1024 * 1024) + " MB");
//日志记录
JobDetailInfluxDB jobDetailInfluxDB = new JobDetailInfluxDB();
jobDetailInfluxDB.setTableName(tableName);
jobDetailInfluxDB.setExcuteDate(date);
jobDetailInfluxDB = jobDetailInfluxDBService.select(jobDetailInfluxDB);
if (Objects.nonNull(jobDetailInfluxDB)) {
if (jobDetailInfluxDB.getState() == 2 && i == 0 && size != 0) {
//第一片执行时返现是失败的，则再次执行
jobDetailInfluxDB.setState(0);
jobDetailInfluxDB.setRowCount(size);
jobDetailInfluxDB.setUpdateTime(LocalDateTime.now());
jobDetailInfluxDBService.updateByMultiId(jobDetailInfluxDB);
} else if (jobDetailInfluxDB.getState() == 0 && i > 0 && jobDetailInfluxDB.getRowCount() + size != 0) {
// 处理中，后续时间片的处理，累加记录数
jobDetailInfluxDB.setState(0);
jobDetailInfluxDB.setRowCount(jobDetailInfluxDB.getRowCount() + size);
jobDetailInfluxDB.setUpdateTime(LocalDateTime.now());
jobDetailInfluxDBService.updateByMultiId(jobDetailInfluxDB);
} else {
System.gc();
break;
}
} else {
if (size > 0) {
jobDetailInfluxDB = new JobDetailInfluxDB();
jobDetailInfluxDB.setTableName(tableName);
jobDetailInfluxDB.setExcuteDate(date);
jobDetailInfluxDB.setState(0);
jobDetailInfluxDB.setRowCount(size);
jobDetailInfluxDB.setUpdateTime(LocalDateTime.now());
jobDetailInfluxDBService.save(jobDetailInfluxDB);
}
}
try {
if (CollectionUtil.isNotEmpty(weakReferenceData.get())) {
//执行目标库的数据处理
Class<?> clazz = null;
Class<?> clazz2 = null;
//获取Table表对应的influxdb对应的表的实体类调用oralceToInfluxDB方法及oralceToInfluxDB的入参clazz2
try {
clazz = Class.forName("com.njcn.influx.bo.po.InfluxDB" + tableName);
clazz2 = Class.forName("com.njcn.oracle.bo.po." + tableName);
} catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
Method method = null;
try {
method = clazz.getDeclaredMethod("oralceToInfluxDB",clazz2);
} catch (NoSuchMethodException e) {
throw new RuntimeException(e);
}
method.setAccessible(true);
Method finalMethod = method;
List list1 =(List) weakReferenceData.get().stream().flatMap(po -> {
try {
Object invoke = finalMethod.invoke(null,po);
Object invoke1 = invoke;
//返回oralce转inflaux，flicker等表是1-1，还有1-4，这是判断返回是否是集合如何是集合继续扁平化
return invoke1 instanceof List ? ((List<?>) invoke1).stream() : Stream.of(invoke1);
} catch (Exception e) {
throw new RuntimeException(e);
}
}).map(item-> (Object) item).collect(Collectors.toList());
//插入influxdb
influxDBBaseService.insertBatchBySlice(list1,5000);
//最后一片时修改状态
}
if (i + 1 == slice && Objects.nonNull(jobDetailInfluxDB)) {
stopWatch.stop();
jobDetailInfluxDB.setState(1);
jobDetailInfluxDB.setDuration(stopWatch.getTotalTimeSeconds());
jobDetailInfluxDB.setUpdateTime(LocalDateTime.now());
jobDetailInfluxDBService.updateByMultiId(jobDetailInfluxDB);
}
} catch (Exception exception) {
exception.printStackTrace();
jobDetailInfluxDB.setState(2);
jobDetailInfluxDB.setUpdateTime(LocalDateTime.now());
jobDetailInfluxDBService.updateByMultiId(jobDetailInfluxDB);
}
System.out.println("执行后总堆内存为：" + runtime.totalMemory() / (1024 * 1024) + " MB");
System.out.println("执行后已用堆内存为：" + (runtime.totalMemory() - runtime.freeMemory()) / (1024 * 1024) + " MB");
System.out.println("执行后空闲堆内存为：" + runtime.freeMemory() / (1024 * 1024) + " MB");
}
});
});
}
}

7
influx-data/influx-target/pom.xml
View File

@@ -18,19 +18,12 @@
</properties>
<dependencies>
<!--oracle数据源-->
<dependency>
<groupId>com.njcn</groupId>
<artifactId>influx-source</artifactId>
<version>1.0.0</version>
</dependency>
<!--oracle数据源-->
<dependency>
<groupId>com.njcn</groupId>
<artifactId>oracle-source</artifactId>
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>

1
influx-data/influx-target/src/main/java/com/njcn/InfluxDataApplication.java
View File

@@ -15,6 +15,7 @@ import org.springframework.scheduling.annotation.EnableAsync;
* @date 2022/11/10
*/
@Slf4j
@EnableAsync
@MapperScan("com.njcn.**.mapper")
@SpringBootApplication(scanBasePackages = "com.njcn",exclude = {SecurityAutoConfiguration.class, SecurityFilterAutoConfiguration.class})
public class InfluxDataApplication {

51
influx-data/influx-target/src/main/java/com/njcn/influx/controller/JobDetailInfluxDBController.java Normal file
View File

@@ -0,0 +1,51 @@
package com.njcn.influx.controller;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.njcn.influx.bo.po.JobDetailInfluxDB;
import com.njcn.influx.service.JobDetailInfluxDBService;
import com.njcn.oracle.bo.param.JobQueryParam;
import com.njcn.oracle.bo.po.JobDetail;
import com.njcn.oracle.service.JobDetailService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiOperation;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* Description:
* Date: 2024/1/9 9:13【需求编号】
*
* @author clam
* @version V1.0.0
*/
@Validated
@Slf4j
@RestController
@RequestMapping("/jobDetail")
@Api(tags = "同步任务")
@AllArgsConstructor
public class JobDetailInfluxDBController {
private final JobDetailInfluxDBService jobDetailInfluxDBService;
@PostMapping("/jobQuery")
@ApiOperation("任务查询")
@ApiImplicitParam(name = "jobQueryParam", value = "任务查询参数", required = true)
public IPage<JobDetailInfluxDB> jobQuery(@RequestBody JobQueryParam jobQueryParam){
IPage<JobDetailInfluxDB> jobDetails = jobDetailInfluxDBService.selectByParam(jobQueryParam);
return jobDetails;//HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, jobDetails, "任务查询");
}
@PostMapping("/jobRemove")
@ApiOperation("任务移除")
@ApiImplicitParam(name = "jobQueryParam", value = "任务移除参数", required = true)
public boolean jobRemove(@RequestBody JobDetailInfluxDB jobDetail){
boolean b = jobDetailInfluxDBService.jobRemove(jobDetail);
return b;
}
}

60
influx-data/influx-target/src/main/java/com/njcn/influx/controller/OracleToInfluxDBController.java
View File

@@ -4,6 +4,7 @@ import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.njcn.influx.service.InfluxDBBaseService;
import com.njcn.influx.service.OracleToInfluxDBService;
import com.njcn.oracle.bo.param.DataAsynParam;
import com.njcn.oracle.bo.param.MigrationParam;
import com.njcn.oracle.mybatis.service.IReplenishMybatisService;
@@ -42,69 +43,14 @@ import java.util.stream.Stream;
@AllArgsConstructor
public class OracleToInfluxDBController {
private final InfluxDBBaseService influxDBBaseService;
private final OracleToInfluxDBService oracleToInfluxDBService;
@PostMapping("/dataSync")
@ApiOperation("数据同步")
@ApiImplicitParam(name = "dataAsynParam", value = "数据同步参数", required = true)
@SneakyThrows
public Boolean dataSync(@RequestBody DataAsynParam dataAsynParam){
dataAsynParam.getTableNames().stream().forEach(temp->{
IReplenishMybatisService executor = null;
try {
executor = (IReplenishMybatisService) SpringUtil.getBean(Class.forName("com.njcn.oracle.service.impl." + temp + "ServiceImpl"));
} catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
IReplenishMybatisService finalExecutor = executor;
MigrationParam migration = new MigrationParam();
LocalDateTime localDateTime = dataAsynParam.getStartTime().atStartOfDay();
LocalDateTime tempStartTime = LocalDateTimeUtil.beginOfDay(localDateTime);
LocalDateTime tempEndTime = LocalDateTimeUtil.endOfDay(localDateTime);
migration.setStartTime(tempStartTime);
migration.setEndTime(tempEndTime);
migration.setStartTime(tempStartTime);
migration.setEndTime(tempEndTime);
List list = finalExecutor.queryData(migration);
if (CollectionUtil.isEmpty(list)) {
return;
}
Class<?> clazz = null;
Class<?> clazz2 = null;
try {
clazz = Class.forName("com.njcn.influx.bo.po.InfluxDB" + temp);
clazz2 = Class.forName("com.njcn.oracle.bo.po." + temp);
} catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
Method method = null;
try {
method = clazz.getDeclaredMethod("oralceToInfluxDB",clazz2);
} catch (NoSuchMethodException e) {
throw new RuntimeException(e);
}
method.setAccessible(true);
Method finalMethod = method;
List list1 =(List) list.stream().flatMap(po -> {
try {
Object invoke = finalMethod.invoke(null,po);
Object invoke1 = invoke;
//返回可能是集合
return invoke1 instanceof List ? ((List<?>) invoke1).stream() : Stream.of(invoke1);
} catch (Exception e) {
throw new RuntimeException(e);
}
}).map(item-> (Object) item).collect(Collectors.toList());
influxDBBaseService.insertBatchBySlice(list1,5000);
});
oracleToInfluxDBService.dataBacthSysc(dataAsynParam);
return true;// HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, true, "数据同步");
}

16
influx-data/influx-target/src/main/resources/application.yml
View File

@@ -1,5 +1,7 @@
#文件位置配置
business:
#分片次数，一定为24的约数，1 2 3 4 6 8 12 24
slice: 4
#处理波形数据位置
# wavePath: C://comtrade
wavePath: /usr/local/comtrade
@@ -23,10 +25,10 @@ server:
spring:
#influxDB内容配置
influx:
url: http://192.168.1.102:8086
url: http://192.168.1.24:8086
user: admin
password: 123456
database: pqsbase_zl
database: test
mapper-location: com.njcn.influx.imapper
application:
name: oracle-data
@@ -87,11 +89,11 @@ spring:
username: pqsadmin
password: Pqsadmin123
driver-class-name: oracle.jdbc.driver.OracleDriver
target:
url: jdbc:oracle:thin:@192.168.1.51:1521:pqsbase
username: pqsadmin_hn
password: pqsadmin
driver-class-name: oracle.jdbc.driver.OracleDriver
# target:
# url: jdbc:oracle:thin:@192.168.1.51:1521:pqsbase
# username: pqsadmin_hn
# password: pqsadmin
# driver-class-name: oracle.jdbc.driver.OracleDriver
#mybatis配置信息
mybatis-plus:
#别名扫描

1
influx-data/influx-target/src/main/resources/static/element.css Normal file
View File

File diff suppressed because one or more lines are too long

57311
influx-data/influx-target/src/main/resources/static/element.js Normal file
View File

File diff suppressed because one or more lines are too long

323
influx-data/influx-target/src/main/resources/static/index.html Normal file
View File

@@ -0,0 +1,323 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>迁移</title>
<link rel="stylesheet" href="./element.css" />
<script src="./vue.js"></script>
<script src="./element.js"></script>
<script src="./locale.js"></script>
</head>
<body>
<div id="app" >
<el-form :inline="true" :model="formInline" class="demo-form-inline">
<el-form-item label="时间区间">
<el-date-picker
v-model="formInline.date"
type="daterange"
range-separator="-"
start-placeholder="开始时间"
end-placeholder="结束时间"
:size="size"
value-format="YYYY-MM-DD"
/>
</el-form-item>
<el-form-item label="数据指标">
<el-select
v-model="formInline.tableNames"
placeholder="请选择指标"
filterable
clearable
multiple
collapse-tags
style="width: 360px"
>
<el-option label="电压闪变数据表-DataFlicker" value="DataFlicker"></el-option>
<el-option label="电压波动数据表-DataFluc" value="DataFluc"></el-option>
<el-option label="谐波电流角度数据表-DataHarmphasicI" value="DataHarmphasicI"></el-option>
<el-option label="谐波电压角度数据表-DataHarmphasicV" value="DataHarmphasicV"></el-option>
<el-option label="有功功率数据表-DataHarmpowerP" value="DataHarmpowerP"></el-option>
<el-option label="无功功率数据表-DataHarmpowerQ" value="DataHarmpowerQ"></el-option>
<el-option label="视在功率数据表-DataHarmpowerS" value="DataHarmpowerS"></el-option>
<el-option label="谐波电流含有率数据表-DataHarmrateI" value="DataHarmrateI"></el-option>
<el-option label="谐波电压含有率数据表-DataHarmrateV" value="DataHarmrateV"></el-option>
<el-option label="电流简谐波幅值数据表-DataInharmI" value="DataInharmI"></el-option>
<el-option label="电压间谐波幅值数据表-DataInharmV" value="DataInharmV"></el-option>
<el-option label="谐波电流幅值数据表-DataI" value="DataI"></el-option>
<el-option label="长时闪变数据表-DataPlt" value="DataPlt"></el-option>
<el-option label="谐波电压幅值数据表-DataV" value="DataV"></el-option>
</el-select>
</el-form-item>
<el-form-item >
<el-checkbox v-model="formInline.checkAll" @change="handleCheckAllChange">{{formInline.checkAll?'取消全选':'全选'}}</el-checkbox>
</el-form-item>
<el-form-item label="数据状态">
<el-select v-model="formInline.states" placeholder="请选择状态" clearable multiple collapse-tags>
<el-option label="执行中" value="0"></el-option>
<el-option label="成功" value="1"></el-option>
<el-option label="失败" value="2"></el-option>
<el-option label="未执行" value="3"></el-option>
</el-select>
</el-form-item>
<el-form-item>
<el-button type="primary" @click="query">查询</el-button>
<el-button type="primary" @click="move()">迁移</el-button>
</el-form-item>
</el-form>
<el-table :data="tableData" style="width: 100%; height: calc(100vh - 120px)" stripe align="center" v-loading="formInline.loading">
<el-table-column prop="tableName" label="指标名"></el-table-column>
<el-table-column prop="tableName_CN" label="表名"></el-table-column>
<el-table-column prop="excuteDate" label="日期(年月日)">
<!-- <template #default="scope"> {{scope.row.startTime&&scope.row.startTime.split(' ')[0]}} </template>-->
</el-table-column>
<el-table-column prop="rowCount" label="条目数"></el-table-column>
<el-table-column prop="updateTime" label="执行时间"></el-table-column>
<el-table-column prop="duration" label="消耗时长" >
<template #default="scope">
{{ formatNumber(scope.row, 'duration') }}
</template>
</el-table-column>
<el-table-column prop="name" label="状态">
<template #default="scope">
<el-tag effect="dark" type="danger" v-if="scope.row.state === 2">失败</el-tag>
<el-tag effect="dark" type="primary" v-else-if="scope.row.state === 0">执行中</el-tag>
<el-tag effect="dark" type="success" v-else-if="scope.row.state === 1">成功</el-tag>
<el-tag effect="dark" type="danger" v-else-if="scope.row.state === 3">未执行</el-tag>
</template>
</el-table-column>
<el-table-column prop="name" label="操作">
<template #default="scope">
<el-popconfirm title="确定删除此条记录吗?" @confirm="jobRemove(scope.row)">
<template #reference>
<el-button type="danger" text v-if="scope.row.state === 1">删除</el-button>
</template>
</el-popconfirm>
<el-button type="primary" text v-if="scope.row.state === 2 || scope.row.state === 3" @click="move(scope.row)">迁移</el-button>
</template>
</el-table-column>
</el-table>
<div style="padding-top: 10px; display: flex; justify-content: flex-end">
<el-pagination
background
layout="total,prev, pager, next"
:total="formInline.total"
:page-size="formInline.pageSize"
:current-page="formInline.currentPage"
@current-change="handleCurrentChange"
@size-change="handleSizeChange"
></el-pagination>
</div>
</div>
<script>
const { createApp, ref, reactive } = Vue
createApp({
setup() {
const tableData = ref([])
const message = ref('Hello Vue!')
// 格式化时间YYYY-MM-DD
const formatTime = (date) => {
let year = date.getFullYear()
let month = date.getMonth() + 1
let day = date.getDate()
if (month < 10) {
month = '0' + month
}
if (day < 10) {
day = '0' + day
}
return `${year}-${month}-${day}`
}
const formInline = reactive({
loading: true,
states: ['0', '1'],
checkAll:true,
tableNames: [
'DataFlicker',
'DataFluc',
'DataHarmphasicI',
'DataHarmphasicV',
'DataHarmpowerP',
'DataHarmpowerQ',
'DataHarmpowerS',
'DataHarmrateI',
'DataHarmrateV',
'DataInharmI',
'DataInharmV',
'DataI',
'DataPlt',
'DataV',
],
tableNames2: [
'DataFlicker',
'DataFluc',
'DataHarmphasicI',
'DataHarmphasicV',
'DataHarmpowerP',
'DataHarmpowerQ',
'DataHarmpowerS',
'DataHarmrateI',
'DataHarmrateV',
'DataInharmI',
'DataInharmV',
'DataI',
'DataPlt',
'DataV',
],
date: [formatTime(new Date()), formatTime(new Date())],
total: 0,
currentPage: 1,
pageSize: 20,
})
const query = () => {
formInline.loading = true
console.log('submit!')
let url = window.location.origin + '/jobDetail/jobQuery';
fetch(url, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
states: formInline.states,
tableNames: formInline.tableNames,
startTime: formInline.date[0],
endTime: formInline.date[1],
currentPage: formInline.currentPage,
pageSize: formInline.pageSize,
}),
})
.then((res) => {
formInline.loading = false
return res.json()
})
.then((res) => {
console.log(res)
res.records.forEach(item=>{
for(let key in item){
if(!item[key] && item[key] !== 0){
item[key] = '/'
}
}
})
tableData.value = res.records
formInline.total = res.total
})
}
query()
const handleCurrentChange = (val) => {
formInline.currentPage = val
query()
}
const handleSizeChange = (val) => {
formInline.currentPage = 1
formInline.pageSize = val
query()
}
const move = (row) => {
let obj = {
tableNames: formInline.tableNames,
startTime: formInline.date[0],
endTime: formInline.date[1],
}
if (row) {
obj = {
tableNames: [row.tableName],
startTime: row.excuteDate,
endTime: row.excuteDate,
}
}
fetch('/data/dataSync', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(obj),
})
.then((res) => {
return res.json()
})
.then((res) => {
if (res) {
ElementPlus.ElMessage.success('数据迁移成功')
query()
} else {
ElementPlus.ElMessage.error('数据迁移失败')
}
})
}
// 处理数据精度
const formatNumber = (row,column) =>{
const duration = row[column];
if(/^-?\d*\.?\d+$/.test(duration)){
let time = Number(duration);
return (time/60).toFixed(2) + " 分钟";
}
return duration;
}
const jobRemove = (row) => {
fetch('/jobDetail/jobRemove', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
tableName: row.tableName,
excuteDate: row.excuteDate,
}),
})
.then((res) => {
return res.json()
})
.then((res) => {
if (res) {
ElementPlus.ElMessage.success('数据删除成功')
query()
} else {
ElementPlus.ElMessage.error('数据删除失败')
}
})
}
const handleCheckAllChange = ()=>{
console.log(formInline.checkAll)
if(formInline.checkAll){
formInline.tableNames = JSON.parse(JSON.stringify(formInline.tableNames2))
}else{
formInline.tableNames = []
}
}
return {
handleCheckAllChange,
message,
formInline,
query,
tableData,
handleCurrentChange,
handleSizeChange,
move,
jobRemove,
formatNumber,
}
},
})
.use(ElementPlus, {
locale: ElementPlusLocaleZhCn,
})
.mount('#app')
</script>
</body>
<style>
* {
margin: 0;
padding: 0;
}
#app {
padding: 10px;
}
</style>
</html>

139
influx-data/influx-target/src/main/resources/static/locale.js Normal file
View File

@@ -0,0 +1,139 @@
/*! Element Plus v2.4.4 */
(function (global, factory) {
typeof exports === 'object' && typeof module !== 'undefined' ? module.exports = factory() :
typeof define === 'function' && define.amd ? define(factory) :
(global = typeof globalThis !== 'undefined' ? globalThis : global || self, global.ElementPlusLocaleZhCn = factory());
})(this, (function () { 'use strict';
var zhCn = {
name: "zh-cn",
el: {
colorpicker: {
confirm: "\u786E\u5B9A",
clear: "\u6E05\u7A7A"
},
datepicker: {
now: "\u6B64\u523B",
today: "\u4ECA\u5929",
cancel: "\u53D6\u6D88",
clear: "\u6E05\u7A7A",
confirm: "\u786E\u5B9A",
selectDate: "\u9009\u62E9\u65E5\u671F",
selectTime: "\u9009\u62E9\u65F6\u95F4",
startDate: "\u5F00\u59CB\u65E5\u671F",
startTime: "\u5F00\u59CB\u65F6\u95F4",
endDate: "\u7ED3\u675F\u65E5\u671F",
endTime: "\u7ED3\u675F\u65F6\u95F4",
prevYear: "\u524D\u4E00\u5E74",
nextYear: "\u540E\u4E00\u5E74",
prevMonth: "\u4E0A\u4E2A\u6708",
nextMonth: "\u4E0B\u4E2A\u6708",
year: "\u5E74",
month1: "1 \u6708",
month2: "2 \u6708",
month3: "3 \u6708",
month4: "4 \u6708",
month5: "5 \u6708",
month6: "6 \u6708",
month7: "7 \u6708",
month8: "8 \u6708",
month9: "9 \u6708",
month10: "10 \u6708",
month11: "11 \u6708",
month12: "12 \u6708",
weeks: {
sun: "\u65E5",
mon: "\u4E00",
tue: "\u4E8C",
wed: "\u4E09",
thu: "\u56DB",
fri: "\u4E94",
sat: "\u516D"
},
months: {
jan: "\u4E00\u6708",
feb: "\u4E8C\u6708",
mar: "\u4E09\u6708",
apr: "\u56DB\u6708",
may: "\u4E94\u6708",
jun: "\u516D\u6708",
jul: "\u4E03\u6708",
aug: "\u516B\u6708",
sep: "\u4E5D\u6708",
oct: "\u5341\u6708",
nov: "\u5341\u4E00\u6708",
dec: "\u5341\u4E8C\u6708"
}
},
select: {
loading: "\u52A0\u8F7D\u4E2D",
noMatch: "\u65E0\u5339\u914D\u6570\u636E",
noData: "\u65E0\u6570\u636E",
placeholder: "\u8BF7\u9009\u62E9"
},
cascader: {
noMatch: "\u65E0\u5339\u914D\u6570\u636E",
loading: "\u52A0\u8F7D\u4E2D",
placeholder: "\u8BF7\u9009\u62E9",
noData: "\u6682\u65E0\u6570\u636E"
},
pagination: {
goto: "\u524D\u5F80",
pagesize: "\u6761/\u9875",
total: "\u5171 {total} \u6761",
pageClassifier: "\u9875",
page: "\u9875",
prev: "\u4E0A\u4E00\u9875",
next: "\u4E0B\u4E00\u9875",
currentPage: "\u7B2C {pager} \u9875",
prevPages: "\u5411\u524D {pager} \u9875",
nextPages: "\u5411\u540E {pager} \u9875",
deprecationWarning: "\u4F60\u4F7F\u7528\u4E86\u4E00\u4E9B\u5DF2\u88AB\u5E9F\u5F03\u7684\u7528\u6CD5\uFF0C\u8BF7\u53C2\u8003 el-pagination \u7684\u5B98\u65B9\u6587\u6863"
},
messagebox: {
title: "\u63D0\u793A",
confirm: "\u786E\u5B9A",
cancel: "\u53D6\u6D88",
error: "\u8F93\u5165\u7684\u6570\u636E\u4E0D\u5408\u6CD5!"
},
upload: {
deleteTip: "\u6309 delete \u952E\u53EF\u5220\u9664",
delete: "\u5220\u9664",
preview: "\u67E5\u770B\u56FE\u7247",
continue: "\u7EE7\u7EED\u4E0A\u4F20"
},
table: {
emptyText: "\u6682\u65E0\u6570\u636E",
confirmFilter: "\u7B5B\u9009",
resetFilter: "\u91CD\u7F6E",
clearFilter: "\u5168\u90E8",
sumText: "\u5408\u8BA1"
},
tree: {
emptyText: "\u6682\u65E0\u6570\u636E"
},
transfer: {
noMatch: "\u65E0\u5339\u914D\u6570\u636E",
noData: "\u65E0\u6570\u636E",
titles: ["\u5217\u8868 1", "\u5217\u8868 2"],
filterPlaceholder: "\u8BF7\u8F93\u5165\u641C\u7D22\u5185\u5BB9",
noCheckedFormat: "\u5171 {total} \u9879",
hasCheckedFormat: "\u5DF2\u9009 {checked}/{total} \u9879"
},
image: {
error: "\u52A0\u8F7D\u5931\u8D25"
},
pageHeader: {
title: "\u8FD4\u56DE"
},
popconfirm: {
confirmButtonText: "\u786E\u5B9A",
cancelButtonText: "\u53D6\u6D88"
}
}
};
return zhCn;
}));

34
influx-data/influx-target/src/main/resources/static/sjzb.json Normal file
View File

@@ -0,0 +1,34 @@
{
"DataFlicker": "电压闪变数据表",
"DataFluc": "电压波动数据表",
"DataHarmphasicI": "谐波电流角度数据表",
"DataHarmphasicV": "谐波电压角度数据表",
"DataHarmpowerP": "有功功率数据表",
"DataHarmpowerQ": "无功功率数据表",
"DataHarmpowerS": "视在功率数据表",
"DataHarmrateI": "谐波电流含有率数据表",
"DataHarmrateV": "谐波电压含有率数据表",
"DataInharmI": "电流简谐波幅值数据表",
"DataInharmV": "电压间谐波幅值数据表",
"DataI": "谐波电流幅值数据表",
"DataPlt": "长时闪变数据表",
"DataV": "谐波电压幅值数据表",
"DayFlicker": "电压闪变日统计表",
"DayFluc": "电压波动日统计表",
"DayHarmphasicI": "谐波电流角度日统计表",
"DayHarmphasicV": "谐波电压角度日统计表",
"DayHarmpowerP": "有功功率日统计表",
"DayHarmpowerQ": "无功功率日统计表",
"DayHarmpowerS": "视在功率日统计表",
"DayHarmrateI": "谐波电流含有率日统计表",
"DayHarmrateV": "谐波电压含有率日统计表",
"DayInharmI": "电流简谐波幅值日统计表",
"DayInharmV": "电压间谐波幅值日统计表",
"DayI": "谐波电流幅值日统计表",
"DayPlt": "长时闪变日统计表",
"DayV": "谐波电压幅值日统计表",
"LimitRate": "合格率统计表",
"LimitTarget": "合格率指标表"
}
帮我转换成标签
<el-option label="电压闪变数据表" value="DataFlicker" />

16508
influx-data/influx-target/src/main/resources/static/vue.js Normal file
View File

File diff suppressed because it is too large Load Diff

2
oracle-data/oracle-source/src/main/java/com/njcn/oracle/service/impl/JobDetailServiceImpl.java
View File

@@ -75,7 +75,7 @@ public class JobDetailServiceImpl extends MppServiceImpl<JobDetailMapper, JobDet
});
});
List<JobDetail> jobDetailIPage = jobDetails.stream().sorted(Comparator.comparing(JobDetail::getExcuteDate).reversed()).
List<JobDetail> jobDetailIPage = jobDetails.stream().filter(temp->jobQueryParam.getStates().contains(temp.getState()+"")).sorted(Comparator.comparing(JobDetail::getExcuteDate).reversed()).
skip((jobQueryParam.getCurrentPage() - 1) * jobQueryParam.getPageSize()).
limit(jobQueryParam.getPageSize()).collect(Collectors.toList());
jobDetailIPage.stream().forEach(temp -> temp.setTableName_CN(ServiceTypeEnum.getValueByCode(temp.getTableName())));