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

寻找报错

master
Cui YongXing 2024-09-06 09:41:48 +08:00
parent ae5005949a
commit d04aedbfdd
3 changed files with 41 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -19,7 +19,10 @@ import java.util.List;
@FeignClient(name = "cloud-etl-datasources", fallbackFactory = DatasourceFeignFactory.class)
public interface DatasourceFeign {
@PostMapping("/dataValue/findTableValueToArray")
public Result<DataValue[][]> findTableValueToArray(@RequestParam("basicId") Long basicId,@RequestParam("sql") String sql);
public Result<DataValue[][]> findTableValueToArray(@RequestParam("basicId") Long basicId,
@RequestParam("sql") String sql,
@RequestParam("one") Long one,
@RequestParam("two") Integer two);
@PostMapping("/dataValue/findTableValue")
public Result<List<List<DataValue>>> findTableValue(@RequestParam("basicId") Long basicId,
@ -30,7 +33,9 @@ public interface DatasourceFeign {
@RequestParam("tableId") Long tableId,
@RequestBody List<List<DataValue>> listList);
@PostMapping("/product/addProduct")
public Result addProduct(@RequestParam("basicId") Long basicId, @RequestParam("tableId") Long tableId, @RequestBody DataValue[][] listList);
public Result addProduct(@RequestParam("basicId") Long basicId,
@RequestParam("tableId") Long tableId,
@RequestBody DataValue[][] listList);
@PostMapping("/dataValue/findCount")
public Result<Long> findCount(@RequestParam("basicId") Long basicId,

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

@ -17,8 +17,9 @@ public class DatasourceFeignFactory implements FallbackFactory<DatasourceFeign>
return new DatasourceFeign() {
@Override
public Result<DataValue[][]> findTableValueToArray(Long basicId, String sql) {
public Result<DataValue[][]> findTableValueToArray(Long basicId, String sql, Integer one, Integer two) {
log.info(e);
return Result.error("网络开小差......");
}

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

@ -247,12 +247,17 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
String finalFieName = fieName;
String finalJoint = joint;
long count = data/PAGE_SIZE==0?1:data/PAGE_SIZE+1;
long pageSize;
if (Weight.high.getValue().equals(weight)){
log.info("执行高级任务");
for (long i = 1; i <= count; i++) {
long pageNum = (i - 1) * PAGE_SIZE;
pageSize = data - pageNum;
if (pageSize>=PAGE_SIZE){
pageSize=PAGE_SIZE;
}
submitHighPriorityTask(()->{
getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map,num);
getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map,pageSize,num);
});
}
@ -261,9 +266,12 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
log.info("执行中级任务");
for (long i = 1; i <= count; i++) {
long pageNum = (i - 1) * PAGE_SIZE;
System.out.println(pageNum);
pageSize = data - pageNum;
if (pageSize>=PAGE_SIZE){
pageSize=PAGE_SIZE;
}
submitMediumPriorityTask(()->{
getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map,num);
getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map,pageSize,num);
});
}
}
@ -271,8 +279,12 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
log.info("执行低级任务");
for (long i = 1; i <= count; i++) {
long pageNum = (i - 1) * PAGE_SIZE;
pageSize = data - pageNum;
if (pageSize>=PAGE_SIZE){
pageSize=PAGE_SIZE;
}
submitLowPriorityTask(()->{
getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map,num);
getString(pageNum, finalFieName, finalJoint,basicId,newBasicId,tableId,map,pageSize,num);
});
}
}
@ -281,11 +293,15 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
// 调整线程分配以适应紧急任务
OptimizedPrioritizedThreadPool.activeEmergencyTasks.set(0);
OptimizedPrioritizedThreadPool.remainingTasks.set(0);
for (long i = 1; i <= count+1; i++) {
for (long i = 1; i <= count; i++) {
long pageNum = (i - 1) * PAGE_SIZE;
pageSize = data - pageNum;
if (pageSize>=PAGE_SIZE){
pageSize=PAGE_SIZE;
}
submitEmergencyTask(()->{
try {
getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map, num);
getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map,pageSize, num);
} finally {
// 减少剩余任务计数
if (OptimizedPrioritizedThreadPool.remainingTasks.decrementAndGet() == 0) {
@ -304,12 +320,20 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
@NotNull
private void getString(Long pageNum,String fieName,String joint,Long basicId,Long newBasicId,Long tableId,HashMap<String, String> map ,Integer num) {
private void getString(Long pageNum,
String fieName,
String joint,
Long basicId,
Long newBasicId,
Long tableId,
HashMap<String, String> map ,
Long one,
Integer two) {
String sqlSelect = " SELECT " + fieName + " FROM " + joint +" limit "+pageNum +","+PAGE_SIZE;
log.info(sqlSelect);
//log.info("执行{}查询的方法",sqlSelect);
Result<DataValue[][]> tableValueResult = datasourceFeign.findTableValueToArray(basicId, sqlSelect);
Result<DataValue[][]> tableValueResult = datasourceFeign.findTableValueToArray(basicId, sqlSelect,one,two);
log.info(tableValueResult);
DataValue[][] data = tableValueResult.getData();
log.info("执行{}查询的方法结束",sqlSelect);