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

添加在线监测表同步

Browse Source
This commit is contained in:
hzj
2024-05-10 08:53:51 +08:00
parent 660e14e149
commit bb0966a189
17 changed files with 403 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1,34 @@
package com.njcn.influx.bo.po;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import java.time.LocalDateTime;
import com.github.jeffreyning.mybatisplus.anno.MppMultiId;
import lombok.Data;
/**
*
* Description:
* Date: 2024/5/9 14:25【需求编号】
*
* @author clam
* @version V1.0.0
*/
@Data
@TableName(value = "r_stat_onlinerate_d")
public class PqsOnlinerateMysql {
@MppMultiId(value = "time_id")
private LocalDateTime timeId;
@MppMultiId(value = "dev_index")
private String devIndex;
@TableField(value = "online_min")
private Integer onlineMin;
@TableField(value = "offline_min")
private Integer offlineMin;
}

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

@@ -0,0 +1,34 @@
package com.njcn.influx.bo.po;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import java.time.LocalDateTime;
import lombok.Data;
/**
*
* Description:
* Date: 2024/5/9 10:42【需求编号】
*
* @author clam
* @version V1.0.0
*/
@Data
@TableName(value = "PQS_ONLINERATE")
public class PqsOnlineratePO {
@TableField(value = "TIMEID")
private LocalDateTime timeid;
@TableField(value = "DEV_INDEX")
private Integer devIndex;
@TableField(value = "ONLINEMIN")
private Integer onlinemin;
@TableField(value = "OFFLINEMIN")
private Integer offlinemin;
}

12
influx-data/influx-source/src/main/java/com/njcn/influx/config/IdMappingCache.java
View File

@@ -1,7 +1,9 @@
package com.njcn.influx.config;
import com.njcn.influx.bo.po.PqDeviceBak;
import com.njcn.influx.bo.po.PqLineBak;
import com.njcn.influx.mapper.PqLineBakMapper;
import com.njcn.influx.service.IPqDeviceBakService;
import com.njcn.influx.service.PqLineBakService;
import io.swagger.v3.oas.annotations.servers.Server;
import org.springframework.beans.factory.annotation.Autowired;
@@ -26,7 +28,8 @@ public class IdMappingCache {
@Autowired
private PqLineBakService pqLineBakService;
@Autowired
private IPqDeviceBakService pqDeviceBakService;
public static Map<String, String> IdMapping = new HashMap<>();
@PostConstruct
@@ -37,6 +40,13 @@ public class IdMappingCache {
String line_id = row.getLineId();
IdMapping.put(line_id,id );
}
List<PqDeviceBak> list = pqDeviceBakService.list();
for (PqDeviceBak row : list) {
String id = row.getId();
String dev_id = row.getDevId()+"";
IdMapping.put(dev_id,id );
}
}
public String getDataById(String id) {

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

@@ -0,0 +1,15 @@
package com.njcn.influx.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.njcn.influx.bo.po.PqsOnlinerateMysql;
/**
*
* Description:
* Date: 2024/5/9 14:25【需求编号】
*
* @author clam
* @version V1.0.0
*/
public interface PqsOnlinerateMysqlMapper extends BaseMapper<PqsOnlinerateMysql> {
}

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

@@ -0,0 +1,15 @@
package com.njcn.influx.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.njcn.influx.bo.po.PqsOnlineratePO;
/**
*
* Description:
* Date: 2024/5/9 10:42【需求编号】
*
* @author clam
* @version V1.0.0
*/
public interface PqsOnlineratePOMapper extends BaseMapper<PqsOnlineratePO> {
}

16
influx-data/influx-source/src/main/java/com/njcn/influx/mapper/mapping/PqsOnlinerateMysqlMapper.xml Normal file
View File

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.njcn.influx.mapper.PqsOnlinerateMysqlMapper">
<resultMap id="BaseResultMap" type="com.njcn.influx.bo.po.PqsOnlinerateMysql">
<!--@mbg.generated-->
<!--@Table pqs_onlinerate-->
<id column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<id column="DEV_INDEX" jdbcType="VARCHAR" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
</resultMap>
<sql id="Base_Column_List">
<!--@mbg.generated-->
TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN
</sql>
</mapper>

65
influx-data/influx-source/src/main/java/com/njcn/influx/mapper/mapping/PqsOnlineratePOMapper.xml Normal file
View File

@@ -0,0 +1,65 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.njcn.influx.mapper.PqsOnlineratePOMapper">
<resultMap id="BaseResultMap" type="com.njcn.influx.bo.po.PqsOnlineratePO">
<!--@mbg.generated-->
<!--@Table PQS_ONLINERATE-->
<id column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<id column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
<result column="TIMEID" jdbcType="TIMESTAMP" property="timeid" />
<result column="DEV_INDEX" jdbcType="DECIMAL" property="devIndex" />
<result column="ONLINEMIN" jdbcType="DECIMAL" property="onlinemin" />
<result column="OFFLINEMIN" jdbcType="DECIMAL" property="offlinemin" />
</resultMap>
<sql id="Base_Column_List">
<!--@mbg.generated-->
TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN, TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN,
TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN, TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN,
TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN, TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN,
TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN, TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN,
TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN, TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN,
TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN, TIMEID, DEV_INDEX, ONLINEMIN, OFFLINEMIN
</sql>
</mapper>

1
influx-data/influx-source/src/main/java/com/njcn/influx/service/JobDetailHoursInfluxDBService.java
View File

@@ -2,7 +2,6 @@ package com.njcn.influx.service;
import com.github.jeffreyning.mybatisplus.service.IMppService;
import com.njcn.influx.bo.po.JobDetailHoursInfluxDB;
import com.njcn.oracle.bo.po.JobDetailHours;
/**
*

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

@@ -6,4 +6,6 @@ public interface OracleToInfluxDBService {
void dataBacthSysc(DataAsynParam dataAsynParam);
void hourseDataBacthSysc(DataAsynParam dataAsynParam);
void AsyncData(DataAsynParam dataAsynParam);
}

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

@@ -0,0 +1,16 @@
package com.njcn.influx.service;
import com.njcn.influx.bo.po.PqsOnlinerateMysql;
import com.baomidou.mybatisplus.extension.service.IService;
/**
*
* Description:
* Date: 2024/5/9 14:25【需求编号】
*
* @author clam
* @version V1.0.0
*/
public interface PqsOnlinerateMysqlService extends IService<PqsOnlinerateMysql>{
}

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

@@ -0,0 +1,21 @@
package com.njcn.influx.service;
import com.njcn.influx.bo.po.PqsOnlineratePO;
import com.baomidou.mybatisplus.extension.service.IService;
import com.njcn.oracle.bo.param.DataAsynParam;
/**
*
* Description:
* Date: 2024/5/9 10:42【需求编号】
*
* @author clam
* @version V1.0.0
*/
public interface PqsOnlineratePOService extends IService<PqsOnlineratePO>{
void minutesDataBacthSysc(DataAsynParam dataAsynParam);
void AsyncData(DataAsynParam dataAsynParam);
}

3
influx-data/influx-source/src/main/java/com/njcn/influx/service/impl/JobDetailHoursInfluxDBServiceImpl.java
View File

@@ -4,8 +4,7 @@ import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
import com.njcn.influx.bo.po.JobDetailHoursInfluxDB;
import com.njcn.influx.mapper.JobDetailHoursInfluxDBMapper;
import com.njcn.influx.service.JobDetailHoursInfluxDBService;
import com.njcn.oracle.bo.po.JobDetailHours;
import com.njcn.oracle.mybatis.mapper.JobDetailHoursMapper;
import org.springframework.stereotype.Service;
/**

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

@@ -16,11 +16,7 @@ import com.njcn.influx.service.JobHistoryLogInfluxdbService;
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.param.ServiceTypeEnum;
import com.njcn.oracle.bo.po.JobDetailHours;
import com.njcn.oracle.bo.po.JobHistoryLog;
import com.njcn.oracle.mybatis.service.IReplenishMybatisService;
import com.njcn.oracle.service.JobDetailHoursService;
import com.njcn.oracle.util.LocalDateUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -216,6 +212,7 @@ public class OracleToInfluxDBServiceImpl implements OracleToInfluxDBService {
}
//按小时来同步数据
@Override
// @Async
public void hourseDataBacthSysc(DataAsynParam dataAsynParam) {
Runtime runtime = Runtime.getRuntime();
System.out.println("开始执行前总堆内存为：" + runtime.totalMemory() / (1024 * 1024) + " MB");
@@ -258,8 +255,8 @@ public class OracleToInfluxDBServiceImpl implements OracleToInfluxDBService {
System.out.println("执行扫描结束时间------------------------------------"+endTime.format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
List list = executor.queryData(migration);
System.out.println("查询到的数据++++++++++++++"+list.size());
List list = executor.queryData(migration);
System.out.println("查询到的数据++++++++++++++"+list.size());
//反射獲取linid的值并把linid的值替换成mysql对应的linid，并记录未匹配的lineid
Iterator iterator = list.iterator();
while (iterator.hasNext()) {
@@ -346,4 +343,20 @@ public class OracleToInfluxDBServiceImpl implements OracleToInfluxDBService {
});
}
@Override
@Async
public void AsyncData(DataAsynParam dataAsynParam) {
LocalDateTime startDateTime1 = dataAsynParam.getStartDateTime();
LocalDateTime endDateTime1 = dataAsynParam.getEndDateTime();
while (endDateTime1.isBefore(startDateTime1)) {
startDateTime1 = startDateTime1.minusHours(1);
DataAsynParam dataAsynParam1 = new DataAsynParam();
dataAsynParam1.setEndDateTime(startDateTime1.minusHours(-1).minusSeconds(1));
dataAsynParam1.setStartDateTime(startDateTime1);
dataAsynParam1.setTableNames(TableEnum.getExecutableTypes());
log.info("执行"+startDateTime1+"时刻数据");
this.hourseDataBacthSysc(dataAsynParam1);
}
}
}

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

@@ -0,0 +1,23 @@
package com.njcn.influx.service.impl;
import com.baomidou.dynamic.datasource.annotation.DS;
import org.springframework.stereotype.Service;
import org.springframework.beans.factory.annotation.Autowired;
import java.util.List;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.njcn.influx.mapper.PqsOnlinerateMysqlMapper;
import com.njcn.influx.bo.po.PqsOnlinerateMysql;
import com.njcn.influx.service.PqsOnlinerateMysqlService;
/**
*
* Description:
* Date: 2024/5/9 14:25【需求编号】
*
* @author clam
* @version V1.0.0
*/
@Service
@DS("target")
public class PqsOnlinerateMysqlServiceImpl extends ServiceImpl<PqsOnlinerateMysqlMapper, PqsOnlinerateMysql> implements PqsOnlinerateMysqlService{
}

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

@@ -0,0 +1,86 @@
package com.njcn.influx.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.LocalDateTimeUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.njcn.influx.bo.param.TableEnum;
import com.njcn.influx.bo.po.PqsOnlinerateMysql;
import com.njcn.influx.config.IdMappingCache;
import com.njcn.influx.service.IPqDeviceBakService;
import com.njcn.influx.service.PqsOnlinerateMysqlService;
import com.njcn.oracle.bo.param.DataAsynParam;
import com.njcn.oracle.bo.po.DataFlicker;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.beans.factory.annotation.Autowired;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.njcn.influx.bo.po.PqsOnlineratePO;
import com.njcn.influx.mapper.PqsOnlineratePOMapper;
import com.njcn.influx.service.PqsOnlineratePOService;
import org.springframework.transaction.annotation.Transactional;
/**
*
* Description:
* Date: 2024/5/9 10:42【需求编号】
*
* @author clam
* @version V1.0.0
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class PqsOnlineratePOServiceImpl extends ServiceImpl<PqsOnlineratePOMapper, PqsOnlineratePO> implements PqsOnlineratePOService{
private final PqsOnlinerateMysqlService pqsOnlinerateMysqlService;
@Override
public void minutesDataBacthSysc(DataAsynParam dataAsynParam) {
List<PqsOnlineratePO> list = this.lambdaQuery().between(PqsOnlineratePO::getTimeid, LocalDateTimeUtil.beginOfDay(dataAsynParam.getStartDateTime()), LocalDateTimeUtil.endOfDay(dataAsynParam.getStartDateTime()).minusSeconds(1)).list();
log.info(dataAsynParam.getStartDateTime()+"-----数据量："+list.size());
if (CollectionUtil.isNotEmpty(list)) {
List<PqsOnlinerateMysql> collect = new ArrayList<>();
list.stream().forEach(temp -> {
PqsOnlinerateMysql pqsOnlinerateMysql = new PqsOnlinerateMysql();
if (!IdMappingCache.IdMapping.containsKey(temp.getDevIndex()+"")){
return;
}else {
pqsOnlinerateMysql.setDevIndex(IdMappingCache.IdMapping.get(temp.getDevIndex()+""));
}
pqsOnlinerateMysql.setTimeId(temp.getTimeid());
pqsOnlinerateMysql.setOnlineMin(temp.getOnlinemin());
pqsOnlinerateMysql.setOfflineMin(temp.getOfflinemin());
collect.add(pqsOnlinerateMysql);
});
QueryWrapper<PqsOnlinerateMysql> queryWrapper = new QueryWrapper<>();
queryWrapper.lambda().between(PqsOnlinerateMysql::getTimeId, LocalDateTimeUtil.beginOfDay(dataAsynParam.getStartDateTime()), LocalDateTimeUtil.endOfDay(dataAsynParam.getStartDateTime()).minusSeconds(1));
pqsOnlinerateMysqlService.remove(queryWrapper);
pqsOnlinerateMysqlService.saveBatch(collect);
}
}
@Override
@Async
public void AsyncData(DataAsynParam dataAsynParam) {
LocalDateTime startDateTime1 = dataAsynParam.getStartDateTime();
LocalDateTime endDateTime1 = dataAsynParam.getEndDateTime();
while (endDateTime1.isBefore(startDateTime1)) {
log.info("执行"+startDateTime1+"时刻数据");
DataAsynParam dataAsynParam1 = new DataAsynParam();
dataAsynParam1.setStartDateTime(startDateTime1);
this.minutesDataBacthSysc(dataAsynParam1);
startDateTime1 = startDateTime1.minusDays(1);
}
}
}