etl
/
cloud-etl-task
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

寻找报错

master
Cui YongXing 2024-09-08 00:47:05 +08:00
parent 5fa93b9d7b
commit d47e8e59f7
1 changed files with 65 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

87
cloud-task-server/src/main/java/com/muyu/task/server/service/impl/TaskInfoServiceImpl.java
View File

@ -29,7 +29,7 @@ import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
import static com.muyu.task.server.thread.OptimizedPrioritizedThreadPool.*;
@ -202,7 +202,7 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
//查询表的数量
Long count = getCount(joint, basicId);
//查询和添加
extracted(count, weight,fieName,joint, basicId, newBasicId, tableId, map, num);
extracted(count, weight, fieName, joint, basicId, newBasicId, tableId, map, num);
long end = System.currentTimeMillis();
//log.info("执行时间：{}",end-start);
return null;
@ -216,7 +216,7 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
return data;
}
private void extracted(Long data, String weight,String finalFieName,String finalJoint, Long basicId, Long newBasicId, Long tableId, HashMap<String, String> map, Integer num) {
private void extracted(Long data, String weight, String finalFieName, String finalJoint, Long basicId, Long newBasicId, Long tableId, HashMap<String, String> map, Integer num) {
long count = data / PAGE_SIZE + (data % PAGE_SIZE > 0 ? 1 : 0);
for (long i = 1; i <= count; i++) {
long pageNum = (i - 1) * PAGE_SIZE;
@ -335,35 +335,78 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
HashMap<String, String> map,
Long one,
Integer two) {
String sqlSelect = " SELECT " + fieName + " FROM " + joint + " LIMIT "+ PAGE_SIZE +" OFFSET " + pageNum ;
String sqlSelect = " SELECT " + fieName + " FROM " + joint + " LIMIT " + PAGE_SIZE + " OFFSET " + pageNum;
log.info(sqlSelect);
//log.info("执行{}查询的方法",sqlSelect);
Result<DataValue[][]> tableValueResult = datasourceFeign.findTableValueToArray(basicId, sqlSelect, one, two);
log.info(tableValueResult);
DataValue[][] data = tableValueResult.getData();
log.info("执行{}查询的方法结束", sqlSelect);
for (DataValue[] datum : data) {
for (DataValue dataValue : datum) {
String key = dataValue.getKey();
String newKey = map.get(key);
dataValue.setKey(newKey);
}
}
log.info("{}查询结束", sqlSelect);
log.info("执行{}添加的方法", sqlSelect);
Result result = datasourceFeign.addProduct(newBasicId, tableId, data);
log.info("{}添加结束", result);
// for (List<DataValue> dataValues : tableValue) {
// for (DataValue dataValue : dataValues) {
// String key = dataValue.getKey();
// String newKey = map.get(key);
// dataValue.setKey(newKey);
// }
// }
executeTheRule(data,map,newBasicId,tableId);
}
private void executeTheRule(DataValue[][] dataValues,HashMap<String,String> map, Long newBasicId,
Long tableId) {
// 创建一个单线程的ExecutorService
ExecutorService executor = Executors.newSingleThreadExecutor();
// 创建一个链表来保存任务
LinkedList<Callable<DataValue[][]>> tasks = new LinkedList<>();
// 初始化第一个任务
tasks.add(() -> {
return dataValues;
});
// 创建任务链
Future<DataValue[][]> currentFuture = null;
for (int i = 1; i <= 4; i++) {
final Future<DataValue[][]> finalCurrentFuture = currentFuture;
Callable<DataValue[][]> task = () -> {
DataValue[][] prevResult = finalCurrentFuture.get();
return prevResult;
};
// 提交任务并更新当前Future
currentFuture = executor.submit(task);
// 等待当前任务完成
try {
currentFuture.get();
System.out.println("Task " + i + " completed with result:");
} catch (InterruptedException | ExecutionException e) {
Thread.currentThread().interrupt();
System.out.println("Task execution failed: " + e.getMessage());
break;
}
}
try {
DataValue[][] afterFilteringDataValue = currentFuture.get();
for (DataValue[] datum : afterFilteringDataValue) {
for (DataValue dataValue : datum) {
String key = dataValue.getKey();
String newKey = map.get(key);
dataValue.setKey(newKey);
}
}
Result result = datasourceFeign.addProduct(newBasicId, tableId, afterFilteringDataValue);
log.info("{}添加结束", result);
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new RuntimeException(e);
}
// 关闭ExecutorService
executor.shutdown();
}
}