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

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
hzj
2024-01-15 20:33:18 +08:00
parent ef7f63bbec c3f7491598
commit 1239b0d97a
13 changed files with 208 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
oracle-data/oracle-source/src/main/java/com/njcn/oracle/bo/po/JobDetail.java
View File

@@ -53,11 +53,13 @@ public class JobDetail {
*/
@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;

1
oracle-data/oracle-source/src/main/java/com/njcn/oracle/mybatis/service/impl/ReplenishMybatisServiceImpl.java
View File

@@ -22,7 +22,6 @@ public abstract class ReplenishMybatisServiceImpl<M extends BatchBaseMapper<T>,
@Override
@Transactional(rollbackFor = Exception.class)
public void insertBatchBySlice(List<T> data, int size) throws RuntimeException{
int totalCount = data.size();
int idxLimit = Math.min(size, totalCount);
List<T> dataTemp = new ArrayList<>(data);

19
oracle-data/oracle-source/src/main/java/com/njcn/oracle/service/DataSyncService.java
View File

@@ -1,19 +0,0 @@
package com.njcn.oracle.service;
import com.njcn.oracle.bo.param.DataAsynParam;
import com.njcn.oracle.mybatis.service.IReplenishMybatisService;
import java.time.LocalDate;
/**
* Description:
* Date: 2024/1/8 16:16【需求编号】
*
* @author clam
* @version V1.0.0
*/
public interface DataSyncService {
void dataSync(IReplenishMybatisService executor, String temp, LocalDate migration);
void dataBacthSysc(DataAsynParam dataAsynParam);
}

1
oracle-data/oracle-source/src/main/java/com/njcn/oracle/service/JobDetailService.java
View File

@@ -9,7 +9,6 @@ import com.baomidou.mybatisplus.extension.service.IService;
import java.util.List;
/**
*
* Description:
* Date: 2024/1/8 12:27【需求编号】
*

112
oracle-data/oracle-source/src/main/java/com/njcn/oracle/service/impl/DataSyncServiceImpl.java
View File

@@ -1,112 +0,0 @@
package com.njcn.oracle.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.extra.spring.SpringUtil;
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.DataSyncService;
import com.njcn.oracle.service.JobDetailService;
import com.njcn.oracle.util.LocalDateUtil;
import lombok.RequiredArgsConstructor;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
/**
* Description:
* Date: 2024/1/8 16:16【需求编号】
*
* @author clam
* @version V1.0.0
*/
@Service
@RequiredArgsConstructor
public class DataSyncServiceImpl implements DataSyncService {
private final JobDetailService jobDetailService;
@Override
public void dataSync(IReplenishMybatisService executor, String temp, LocalDate date) {
StopWatch stopWatch = new StopWatch();
stopWatch.start();
LocalDateTime localDateTime = date.atStartOfDay();
LocalDateTime tempStartTime = LocalDateTimeUtil.beginOfDay(localDateTime);
LocalDateTime tempEndTime = LocalDateTimeUtil.endOfDay(localDateTime);
MigrationParam migration = new MigrationParam();
migration.setStartTime(tempStartTime);
migration.setEndTime(tempEndTime);
List list = executor.queryData(migration);
JobDetail jobDetail = new JobDetail();
jobDetail.setTableName(temp);
jobDetail.setExcuteDate(date);
JobDetail select = jobDetailService.select(jobDetail);
if (Objects.nonNull(select)){
if (select.getState()==2) {
jobDetail.setState(0);
jobDetail.setRowCount(list.size());
jobDetail.setUpdateTime(LocalDateTime.now());
jobDetailService.updateByMultiId(jobDetail);
}else {
return;
}
}else {
jobDetail.setState(0);
jobDetail.setRowCount(list.size());
jobDetail.setUpdateTime(LocalDateTime.now());
jobDetailService.save(jobDetail);
}
try {
if(CollectionUtil.isNotEmpty(list)){
executor.clearTargetData(migration);
executor.insertBatchByDB(list);
}
stopWatch.stop();
jobDetail.setState(1);
jobDetail.setDuration(stopWatch.getTotalTimeSeconds());
jobDetail.setUpdateTime(LocalDateTime.now());
jobDetailService.updateByMultiId(jobDetail);
}catch (
Exception e
){
jobDetail.setState(2);
jobDetail.setUpdateTime(LocalDateTime.now());
jobDetailService.updateByMultiId(jobDetail);
}
}
@Override
@Async
public void dataBacthSysc(DataAsynParam dataAsynParam) {
List<String> tableNames = dataAsynParam.getTableNames();
tableNames.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);
}
List<LocalDate> dateList = LocalDateUtil.getDateList(dataAsynParam.getStartTime(), dataAsynParam.getEndTime());
IReplenishMybatisService finalExecutor = executor;
dateList.forEach(date->{
this.dataSync(finalExecutor,temp,date);
});
});
}
}

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

@@ -10,6 +10,7 @@ import com.njcn.oracle.bo.param.ServiceTypeEnum;
import com.njcn.oracle.util.LocalDateUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.time.LocalDate;
import java.util.ArrayList;
@@ -24,7 +25,6 @@ import com.njcn.oracle.service.JobDetailService;
import org.springframework.util.CollectionUtils;
/**
*
* Description:
* Date: 2024/1/8 12:27【需求编号】
*
@@ -38,9 +38,10 @@ public class JobDetailServiceImpl extends MppServiceImpl<JobDetailMapper, JobDet
@Override
public JobDetail select(JobDetail jobDetail) {
JobDetail jobDetail1 = this.lambdaQuery().eq(JobDetail::getTableName,jobDetail.getTableName()).
eq(JobDetail::getExcuteDate,jobDetail.getExcuteDate()).one();
return jobDetail1;
return this.lambdaQuery()
.eq(JobDetail::getTableName, jobDetail.getTableName())
.eq(JobDetail::getExcuteDate, jobDetail.getExcuteDate())
.one();
}
@Override
@@ -53,7 +54,7 @@ public class JobDetailServiceImpl extends MppServiceImpl<JobDetailMapper, JobDet
List<JobDetail> jobDetails = this.getBaseMapper().selectList(queryWrapper);
List<String> tableNames = new ArrayList<>();
List<String> tableNames;
if (CollectionUtils.isEmpty(jobQueryParam.getTableNames())) {
tableNames = ServiceTypeEnum.getExecutableTypes();
} else {
@@ -88,10 +89,10 @@ public class JobDetailServiceImpl extends MppServiceImpl<JobDetailMapper, JobDet
@Override
public boolean jobRemove(JobDetail jobDetail) {
QueryWrapper<JobDetail> queryWrapper = new QueryWrapper<>();
queryWrapper.lambda().eq(JobDetail::getTableName,jobDetail.getTableName()).
eq(JobDetail::getExcuteDate,jobDetail.getExcuteDate());
boolean remove = this.remove(queryWrapper);
return remove;
queryWrapper.lambda()
.eq(JobDetail::getTableName, jobDetail.getTableName())
.eq(JobDetail::getExcuteDate, jobDetail.getExcuteDate());
return this.remove(queryWrapper);
}

2
oracle-data/oracle-source/src/main/java/com/njcn/oracle/util/LocalDateUtil.java
View File

@@ -16,11 +16,9 @@ public class LocalDateUtil {
public static List<LocalDate> getDateList(LocalDate startDate, LocalDate endDate) {
List<LocalDate> dateList = new ArrayList<>();
long daysBetween = ChronoUnit.DAYS.between(startDate, endDate);
for (long i = 0; i <= daysBetween; i++) {
dateList.add(startDate.plusDays(i));
}
return dateList;
}
}

29
oracle-data/oracle-target/src/main/java/com/njcn/oracle/controller/DataSyncController.java
View File

@@ -1,14 +1,7 @@
package com.njcn.oracle.controller;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.njcn.oracle.bo.param.DataAsynParam;
import com.njcn.oracle.bo.param.MigrationParam;
import com.njcn.oracle.mybatis.service.IReplenishMybatisService;
import com.njcn.oracle.service.DataSyncService;
import com.njcn.oracle.util.LocalDateUtil;
import com.njcn.oracle.service.IOracleService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiOperation;
@@ -16,14 +9,6 @@ import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.ModelAndView;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.List;
import java.util.Objects;
/**
* Description:
@@ -41,11 +26,12 @@ import java.util.Objects;
public class DataSyncController {
private final DataSyncService dataSyncService;
private final IOracleService oracleService;
/**
* 将时间区间拆解为24小时分区，小于24小时，仅查询当前时间区间，目前不支持指定时分秒
* 暂且不用多线程处理，因无法控制数据指标数和时间范围多大，担心
*
* @param dataAsynParam
*/
@PostMapping("/dataSync")
@@ -53,17 +39,10 @@ public class DataSyncController {
@ApiImplicitParam(name = "dataAsynParam", value = "数据同步参数", required = true)
public Boolean dataSync(@RequestBody DataAsynParam dataAsynParam) {
dataSyncService.dataBacthSysc(dataAsynParam);
oracleService.dataBacthSysc(dataAsynParam);
return true;// HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, true, "数据同步");
}
}

7
oracle-data/oracle-target/src/main/java/com/njcn/oracle/service/IOracleService.java Normal file
View File

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

148
oracle-data/oracle-target/src/main/java/com/njcn/oracle/service/impl/OracleServiceImpl.java Normal file
View File

@@ -0,0 +1,148 @@
package com.njcn.oracle.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.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.IOracleService;
import com.njcn.oracle.service.JobDetailService;
import com.njcn.oracle.util.LocalDateUtil;
import lombok.Data;
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.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
@Data
@Service
@RequiredArgsConstructor
public class OracleServiceImpl implements IOracleService {
private final static String PACKAGE_PREFIX = "com.njcn.oracle.service.impl.";
private final static String PACKAGE_SUFFIX = "ServiceImpl";
private final JobDetailService jobDetailService;
@Value("${business.slice:2}")
private int slice;
/**
* 批量处理数据表的数据迁移
*
* @param dataAsynParam 业务参数
*/
@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");
//日志记录
JobDetail jobDetail = new JobDetail();
jobDetail.setTableName(tableName);
jobDetail.setExcuteDate(date);
jobDetail = jobDetailService.select(jobDetail);
if (Objects.nonNull(jobDetail)) {
if (jobDetail.getState() == 2 && i == 0 && size != 0) {
//第一片执行时返现是失败的，则再次执行
jobDetail.setState(0);
jobDetail.setRowCount(size);
jobDetail.setUpdateTime(LocalDateTime.now());
jobDetailService.updateByMultiId(jobDetail);
} else if (jobDetail.getState() == 0 && i > 0 && jobDetail.getRowCount() + size != 0) {
// 处理中，后续时间片的处理，累加记录数
jobDetail.setState(0);
jobDetail.setRowCount(jobDetail.getRowCount() + size);
jobDetail.setUpdateTime(LocalDateTime.now());
jobDetailService.updateByMultiId(jobDetail);
} else {
System.gc();
break;
}
} else {
if (size > 0) {
jobDetail = new JobDetail();
jobDetail.setTableName(tableName);
jobDetail.setExcuteDate(date);
jobDetail.setState(0);
jobDetail.setRowCount(size);
jobDetail.setUpdateTime(LocalDateTime.now());
jobDetailService.save(jobDetail);
}
}
try {
if (CollectionUtil.isNotEmpty(weakReferenceData.get())) {
//执行目标库的数据处理
executor.clearTargetData(migration);
executor.insertBatchByDB(weakReferenceData.get());
//最后一片时修改状态
}
if (i + 1 == slice && Objects.nonNull(jobDetail)) {
stopWatch.stop();
jobDetail.setState(1);
jobDetail.setDuration(stopWatch.getTotalTimeSeconds());
jobDetail.setUpdateTime(LocalDateTime.now());
jobDetailService.updateByMultiId(jobDetail);
}
} catch (Exception exception) {
exception.printStackTrace();
jobDetail.setState(2);
jobDetail.setUpdateTime(LocalDateTime.now());
jobDetailService.updateByMultiId(jobDetail);
}
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");
}
});
});
}
}

6
oracle-data/oracle-target/src/main/resources/application.yml
View File

@@ -56,7 +56,7 @@ spring:
strict: false
datasource:
master:
url: jdbc:oracle:thin:@192.168.1.51:1521:pqsbase
url: jdbc:oracle:thin:@192.168.1.101:1521:pqsbase
username: pqsadmin
password: Pqsadmin123
driver-class-name: oracle.jdbc.driver.OracleDriver
@@ -82,3 +82,7 @@ mybatis-plus:
#指定主键生成策略
id-type: assign_uuid
#分片次数，一定为24的约数，1 2 3 4 6 8 12 24
business:
slice: 4

21
oracle-data/oracle-target/src/main/resources/static/index.html
View File

@@ -5,7 +5,6 @@
<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>
@@ -86,7 +85,11 @@
</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="消耗时长"></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>
@@ -181,7 +184,8 @@
const query = () => {
formInline.loading = true
console.log('submit!')
fetch('http://192.168.1.130:8091/jobDetail/jobQuery', {
let url = window.location.origin + '/jobDetail/jobQuery';
fetch(url, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
@@ -255,6 +259,16 @@
}
})
}
// 处理数据精度
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',
@@ -288,6 +302,7 @@
handleSizeChange,
move,
jobRemove,
formatNumber,
}
},
})

1
oracle-data/oracle-target/src/main/resources/static/locale.js
View File

@@ -137,4 +137,3 @@
return zhCn;
}));
已禁用缩放