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

添加联查信息方法

master
陈思豪 2024-09-06 19:24:51 +08:00
parent 2950c9a577
commit 31ffd97900
4 changed files with 155 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
cloud-task-server/src/main/java/com/muyu/service/impl/TaskServiceImpl.java
View File

@ -4,20 +4,18 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.muyu.common.core.domain.Result; import com.muyu.common.core.domain.Result;
import com.muyu.common.core.utils.StringUtils; import com.muyu.common.core.utils.StringUtils;
import com.muyu.domain.TaskInfo; import com.muyu.domain.*;
import com.muyu.domain.TaskInputInfo;
import com.muyu.domain.TaskJoinInfo;
import com.muyu.domain.TaskOutputInfo;
import com.muyu.domain.req.TaskInfoReq; import com.muyu.domain.req.TaskInfoReq;
import com.muyu.domain.taskenum.Weight; import com.muyu.domain.taskenum.Weight;
import com.muyu.mapper.TaskMapper; import com.muyu.mapper.TaskMapper;
import com.muyu.service.TaskInputService; import com.muyu.service.TaskInputService;
import com.muyu.service.TaskJoinService; import com.muyu.service.TaskJoinService;
import com.muyu.service.TaskOutputService; import com.muyu.service.TaskOutputService;
import com.muyu.task.SegmentTask;
import com.muyu.service.TaskService; import com.muyu.service.TaskService;
import com.muyu.task.feign.DataValueClient; import com.muyu.task.feign.DataValueClient;
import jakarta.annotation.Resource;
import lombok.extern.log4j.Log4j2; import lombok.extern.log4j.Log4j2;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@ -259,8 +257,60 @@ public class TaskServiceImpl extends ServiceImpl<TaskMapper, TaskInfo> implement
} }
public void selectAndAdd(){ static class SegmentTask implements Runnable, Comparable<SegmentTask> {
@Resource
private DataValueClient dataValueClient;
private final String sql;
private final Long basicId;
private final Long tableId;
private final Weight weight;
HashMap<String, String> map = new HashMap<>();
public SegmentTask(String sql, Long basicId, Long tableId, HashMap<String, String> map, Weight weight) {
this.sql = sql;
this.basicId = basicId;
this.tableId = tableId;
this.map = map;
this.weight = weight;
}
@Override
public void run() {
log.info("开始执行任务：" + basicId + " " + sql + " " + weight + " " + tableId, tableId);
// 查询数据
Result tableValue = dataValueClient.findTableValue(basicId, sql);
log.info("远程调用完毕，调用的值为{}", tableValue.getData());
List<List<DataValue>> data = (List<List<DataValue>>) tableValue.getData();
log.info("查询结果：{}", data);
for (List<DataValue> datum : data) {
for (DataValue dataValue : datum) {
String key = dataValue.getKey();
String newKey = map.get(key);
dataValue.setKey(newKey);
}
}
log.info("开始添加：{}", data);
Result result = dataValueClient.addTableDataValue(basicId, tableId, data);
log.info("添加完毕字段：{}", result);
// 处理结果
// dataValueList.accept(tableValue.getData());
}
public Weight geWeight() {
return weight;
}
@Override
public int compareTo(@NotNull SegmentTask o) {
return Integer.compare(o.geWeight().getWeight(), this.weight.getWeight());
}
} }
} }

23
cloud-task-server/src/main/java/com/muyu/task/PriorityQueryExecutor.java
View File

@ -1,23 +0,0 @@
package com.muyu.task;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @PackageName:com.muyu.task
* @ClassName:PriorityQueryExecutor
* @Description:
* @author: ￥陈思豪￥
* @date: 2024/9/6 10:31
*/
public class PriorityQueryExecutor {
private static final ExecutorService executor = Executors.newFixedThreadPool(50);
public void execute(SegmentTask task) {
}
}

18
cloud-task-server/src/main/java/com/muyu/task/PriorityThreadPool.java 100644
View File

@ -0,0 +1,18 @@
//package com.muyu.task;
//
//import java.util.concurrent.*;
//
///**
// * @PackageName:com.muyu.task
// * @ClassName:PriorityQueryExecutor
// * @Description:
// * @author: ￥陈思豪￥
// * @date: 2024/9/6 10:31
// */
//public class PriorityThreadPool {
//
// private static final ExecutorService executor;
// private static final Semaphore s
//
//
//}

160
cloud-task-server/src/main/java/com/muyu/task/SegmentTask.java
View File

@ -1,80 +1,80 @@
package com.muyu.task; //package com.muyu.task;
//
/** ///**
* @PackageName:com.muyu.task // * @PackageName:com.muyu.task
* @ClassName:SementTask // * @ClassName:SementTask
* @Description: // * @Description:
* @author: ￥陈思豪￥ // * @author: ￥陈思豪￥
* @date: 2024/9/5 19:57 // * @date: 2024/9/5 19:57
*/ // */
//
import com.muyu.common.core.domain.Result; //import com.muyu.common.core.domain.Result;
import com.muyu.domain.DataValue; //import com.muyu.domain.DataValue;
import com.muyu.domain.taskenum.Weight; //import com.muyu.domain.taskenum.Weight;
import com.muyu.task.feign.DataValueClient; //import com.muyu.task.feign.DataValueClient;
import jakarta.annotation.Resource; //import jakarta.annotation.Resource;
import lombok.extern.log4j.Log4j2; //import lombok.extern.log4j.Log4j2;
import org.jetbrains.annotations.NotNull; //import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
//
import java.util.HashMap; //import java.util.HashMap;
import java.util.List; //import java.util.List;
//
@Log4j2 //@Log4j2
public class SegmentTask implements Runnable, Comparable<SegmentTask> { //public class SegmentTask implements Runnable, Comparable<SegmentTask> {
//
//
@Resource // @Resource
private DataValueClient dataValueClient; // private DataValueClient dataValueClient;
//
private final String sql; // private final String sql;
private final Long basicId; // private final Long basicId;
private final Long tableId; // private final Long tableId;
private final Weight weight; // private final Weight weight;
HashMap<String, String> map = new HashMap<>(); // HashMap<String, String> map = new HashMap<>();
//
public SegmentTask(String sql, Long basicId, Long tableId, HashMap<String, String> map, Weight weight) { // public SegmentTask(String sql, Long basicId, Long tableId, HashMap<String, String> map, Weight weight) {
this.sql = sql; // this.sql = sql;
this.basicId = basicId; // this.basicId = basicId;
this.tableId = tableId; // this.tableId = tableId;
this.map=map; // this.map=map;
this.weight = weight; // this.weight = weight;
} // }
//
@Override // @Override
public void run() { // public void run() {
//
log.info("开始执行任务：" + basicId + " " + sql + " " + weight + " " + tableId, tableId); // log.info("开始执行任务：" + basicId + " " + sql + " " + weight + " " + tableId, tableId);
// 查询数据 // // 查询数据
Result tableValue = dataValueClient.findTableValue(basicId, sql); // Result tableValue = dataValueClient.findTableValue(basicId, sql);
log.info("远程调用完毕，调用的值为{}",tableValue.getData()); // log.info("远程调用完毕，调用的值为{}",tableValue.getData());
List<List<DataValue>> data = (List<List<DataValue>>) tableValue.getData(); // List<List<DataValue>> data = (List<List<DataValue>>) tableValue.getData();
log.info("查询结果：{}", data); // log.info("查询结果：{}", data);
for (List<DataValue> datum : data) { // for (List<DataValue> datum : data) {
for (DataValue dataValue : datum) { // for (DataValue dataValue : datum) {
String key = dataValue.getKey(); // String key = dataValue.getKey();
String newKey = map.get(key); // String newKey = map.get(key);
dataValue.setKey(newKey); // dataValue.setKey(newKey);
} // }
} // }
log.info("开始添加：{}",data); // log.info("开始添加：{}",data);
//
Result result = dataValueClient.addTableDataValue(basicId, tableId, data); // Result result = dataValueClient.addTableDataValue(basicId, tableId, data);
log.info("添加完毕字段：{}",result); // log.info("添加完毕字段：{}",result);
// 处理结果 // // 处理结果
// dataValueList.accept(tableValue.getData()); //// dataValueList.accept(tableValue.getData());
//
//
} // }
//
public Weight geWeight() { // public Weight geWeight() {
return weight; // return weight;
} // }
//
//
@Override // @Override
public int compareTo(@NotNull SegmentTask o) { // public int compareTo(@NotNull SegmentTask o) {
return Integer.compare(o.weight.getWeight(), this.weight.getWeight()); // return Integer.compare(o.weight.getWeight(), this.weight.getWeight());
} // }
} //}
//