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

修改bug

master
Cui YongXing 2024-09-05 18:59:11 +08:00
parent eab767fdb4
commit 40875894c4
3 changed files with 55 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
cloud-task-remote/src/main/java/com/muyu/remote/feign/DatasourceFeign.java
View File

@ -18,13 +18,21 @@ import java.util.List;
@FeignClient(name = "cloud-etl-datasources", fallbackFactory = DatasourceFeignFactory.class) @FeignClient(name = "cloud-etl-datasources", fallbackFactory = DatasourceFeignFactory.class)
public interface DatasourceFeign { public interface DatasourceFeign {
@PostMapping("/dataValue/findTableValue") @PostMapping("/findTableValueToArray")
public Result<List<List<DataValue>>> findTableValue(@RequestParam("basicId") Long basicId, @RequestParam("sql") String sql); public Result<DataValue[][]> findTableValueToArray(@RequestParam("basicId") Long basicId,@RequestParam("sql") String sql);
@PostMapping("/dataValue/findTableValue")
public Result<List<List<DataValue>>> findTableValue(@RequestParam("basicId") Long basicId,
@RequestParam("sql") String sql);
@PostMapping("/product/addProduct") @PostMapping("/product/addProduct")
public Result addProduct(@RequestParam("basicId") Long basicId, @RequestParam("tableId") Long tableId, @RequestBody List<List<DataValue>> listList); public Result addProduct(@RequestParam("basicId") Long basicId,
@RequestParam("tableId") Long tableId,
@RequestBody List<List<DataValue>> listList);
@PostMapping("/addProduct")
public Result addProduct(@RequestParam("basicId") Long basicId, @RequestParam("tableId") Long tableId, @RequestBody DataValue[][] listList);
@PostMapping("/dataValue/findCount") @PostMapping("/dataValue/findCount")
public Result<Long> findCount(@RequestParam("basicId") Long basicId,@RequestParam("sql") String sql); public Result<Long> findCount(@RequestParam("basicId") Long basicId,
@RequestParam("sql") String sql);
} }

15
cloud-task-remote/src/main/java/com/muyu/remote/feign/Factory/DatasourceFeignFactory.java
View File

@ -15,8 +15,16 @@ public class DatasourceFeignFactory implements FallbackFactory<DatasourceFeign>
@Override @Override
public DatasourceFeign create(Throwable e) { public DatasourceFeign create(Throwable e) {
return new DatasourceFeign() { return new DatasourceFeign() {
@Override @Override
public Result findTableValue(Long basicId, String sql) { public Result<DataValue[][]> findTableValueToArray(Long basicId, String sql) {
log.info(e);
return Result.error("网络开小差......");
}
@Override
public Result<List<List<DataValue>>> findTableValue(Long basicId, String sql) {
log.info(e); log.info(e);
return Result.error("网络开小差......"); return Result.error("网络开小差......");
} }
@ -27,6 +35,11 @@ public class DatasourceFeignFactory implements FallbackFactory<DatasourceFeign>
return Result.error("网络开小差......"); return Result.error("网络开小差......");
} }
@Override
public Result addProduct(Long basicId, Long tableId, DataValue[][] listList) {
return null;
}
@Override @Override
public Result<Long> findCount(Long basicId, String sql) { public Result<Long> findCount(Long basicId, String sql) {
log.info(e); log.info(e);

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

@ -41,6 +41,8 @@ import static com.muyu.task.server.thread.OptimizedPrioritizedThreadPool.*;
@Service @Service
public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> implements TaskInfoService { public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> implements TaskInfoService {
private final Integer PAGE_SIZE = 5000;
@Resource @Resource
private TaskInfoMapper taskInfoMapper; private TaskInfoMapper taskInfoMapper;
@Autowired @Autowired
@ -76,7 +78,7 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
respPage.setPages(pages); respPage.setPages(pages);
return respPage; return respPage;
} }
//
// @Override // @Override
// public String findByFieName(Long taskId) { // public String findByFieName(Long taskId) {
// QueryWrapper<TaskInput> wrapper = new QueryWrapper<>(); // QueryWrapper<TaskInput> wrapper = new QueryWrapper<>();
@ -228,8 +230,10 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
Long tableId = one.getTableId(); Long tableId = one.getTableId();
Long newBasicId = one.getBasicId(); Long newBasicId = one.getBasicId();
HashMap<String, String> map = new HashMap<>(); HashMap<String, String> map = new HashMap<>();
String[] newFieName = one.getNewFieName().split(","); String[] newFieName = one.getNewFieName().split(",");
String[] lastFieName = one.getLastFieName().split(","); String[] lastFieName = one.getLastFieName().split(",");
Integer num = lastFieName.length;
for (int i = 0; i < newFieName.length; i++) { for (int i = 0; i < newFieName.length; i++) {
map.put(lastFieName[i], newFieName[i]); map.put(lastFieName[i], newFieName[i]);
fieName += "," + tableNameMap.get(lastFieName[i]); fieName += "," + tableNameMap.get(lastFieName[i]);
@ -241,32 +245,32 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
Long data = countResult.getData(); Long data = countResult.getData();
String finalFieName = fieName; String finalFieName = fieName;
String finalJoint = joint; String finalJoint = joint;
long count = data/5000==0?1:data/5000+1; long count = data/PAGE_SIZE==0?1:data/PAGE_SIZE+1;
if (Weight.high.getValue().equals(weight)){ if (Weight.high.getValue().equals(weight)){
log.info("执行高级任务"); log.info("执行高级任务");
for (long i = 1; i <= count; i++) { for (long i = 1; i <= count; i++) {
long pageNum = (i - 1) * 5000; long pageNum = (i - 1) * PAGE_SIZE;
submitHighPriorityTask(()->{ submitHighPriorityTask(()->{
getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map); getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map,num);
}); });
} }
} }
if (Weight.centre.getValue().equals(weight)){ if (Weight.centre.getValue().equals(weight)){
log.info("执行中级任务"); log.info("执行中级任务");
for (long i = 1; i <= count; i++) { for (long i = 1; i <= count; i++) {
long pageNum = (i - 1) * 5000; long pageNum = (i - 1) * PAGE_SIZE;
System.out.println(pageNum); System.out.println(pageNum);
submitMediumPriorityTask(()->{ submitMediumPriorityTask(()->{
getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map); getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map,num);
}); });
} }
} }
if (Weight.low.getValue().equals(weight)){ if (Weight.low.getValue().equals(weight)){
log.info("执行低级任务"); log.info("执行低级任务");
for (long i = 1; i <= count; i++) { for (long i = 1; i <= count; i++) {
long pageNum = (i - 1) * 5000; long pageNum = (i - 1) * PAGE_SIZE;
submitLowPriorityTask(()->{ submitLowPriorityTask(()->{
getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map); getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map,num);
}); });
} }
} }
@ -275,10 +279,10 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
// 调整线程分配以适应紧急任务 // 调整线程分配以适应紧急任务
adjustForEmergency(); adjustForEmergency();
for (long i = 1; i <= count; i++) { for (long i = 1; i <= count; i++) {
long pageNum = (i - 1) * 5000; long pageNum = (i - 1) * PAGE_SIZE;
submitEmergencyTask(()->{ submitEmergencyTask(()->{
getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map); getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map,num);
}); });
} }
// 任务完成后恢复默认线程分配 // 任务完成后恢复默认线程分配
@ -291,19 +295,26 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
@NotNull @NotNull
private void getString(Long pageNum,String fieName,String joint,Long basicId,Long newBasicId,Long tableId,HashMap<String, String> map ) { private void getString(Long pageNum,String fieName,String joint,Long basicId,Long newBasicId,Long tableId,HashMap<String, String> map ,Integer num) {
String sqlSelect = " SELECT " + fieName + " FROM " + joint +" limit "+pageNum +",5000 "; String sqlSelect = " SELECT " + fieName + " FROM " + joint +" limit "+pageNum +","+PAGE_SIZE;
log.info(sqlSelect); log.info(sqlSelect);
Result<List<List<DataValue>>> tableValueResult = datasourceFeign.findTableValue(basicId, sqlSelect); Result<DataValue[][]> tableValueResult = datasourceFeign.findTableValueToArray(basicId, sqlSelect);
List<List<DataValue>> tableValue = tableValueResult.getData(); DataValue[][] data = tableValueResult.getData();
for (List<DataValue> dataValues : tableValue) { for (DataValue[] datum : data) {
for (DataValue dataValue : dataValues) { 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);
} }
} }
Result result = datasourceFeign.addProduct(newBasicId, tableId, tableValue); // for (List<DataValue> dataValues : tableValue) {
// for (DataValue dataValue : dataValues) {
// String key = dataValue.getKey();
// String newKey = map.get(key);
// dataValue.setKey(newKey);
// }
// }
Result result = datasourceFeign.addProduct(newBasicId, tableId, data);
log.info(result); log.info(result);
} }