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

寻找报错

master
Cui YongXing 2024-09-08 01:06:56 +08:00
parent d3bd8c27cc
commit 128d031932
1 changed files with 23 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -344,16 +344,16 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
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);
}
}
Result result = datasourceFeign.addProduct(newBasicId, tableId, data);
log.info("{}添加结束", result);
//executeTheRule(data,map,newBasicId,tableId);
// for (DataValue[] datum : data) {
// for (DataValue dataValue : datum) {
// String key = dataValue.getKey();
// String newKey = map.get(key);
// dataValue.setKey(newKey);
// }
// }
// Result result = datasourceFeign.addProduct(newBasicId, tableId, data);
// log.info("{}添加结束", result);
executeTheRule(data,map,newBasicId,tableId);
}
@ -368,13 +368,15 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
// 初始化第一个任务
tasks.add(() -> {
log.info(dataValues);
return dataValues;
});
// 创建任务链
Future<DataValue[][]> currentFuture = null;
// 提交第一个任务
Future<DataValue[][]> currentFuture = executor.submit(tasks.poll());
for (int i = 1; i <= 4; i++) {
final Future<DataValue[][]> finalCurrentFuture = currentFuture;
log.info(finalCurrentFuture);
Callable<DataValue[][]> task = () -> {
DataValue[][] prevResult = finalCurrentFuture.get();
@ -397,7 +399,15 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
}
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);