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

测试线程池

master
陈思豪 2024-09-08 10:32:45 +08:00
parent e8516c4d2e
commit d94ad26737
2 changed files with 29 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
cloud-task-remote/src/main/java/com/muyu/task/feign/DataValueClient.java
View File

@ -22,23 +22,35 @@ public interface DataValueClient {
/**
* 根据sql查询数据
*
* @param basicId
* @param sql
* @return
*/
@PostMapping("/DataValue/findTableValue")
public Result<List<List<DataValue>>> findTableValue(@RequestParam("basicId") Long basicId,@RequestParam("sql") String sql);
public Result<List<List<DataValue>>> findTableValue(@RequestParam("basicId") Long basicId, @RequestParam("sql") String sql);
/**
* 根据sql查询数据条数
*
* @param basicId
* @param sql
* @return
*/
@PostMapping("/DataValue/findCount")
public Integer findCount(@RequestParam("basicId") Long basicId,@RequestParam("sql") String sql);
public Integer findCount(@RequestParam("basicId") Long basicId, @RequestParam("sql") String sql);
//添加
@PostMapping("/DataValue/addTable")
public Result addTableDataValue(@RequestParam("basicId") Long basicId,@RequestParam("tableId") Long tableId, @RequestBody List<List<DataValue>> dataValue);
public Result addTableDataValue(@RequestParam("basicId") Long basicId, @RequestParam("tableId") Long tableId, @RequestBody List<List<DataValue>> dataValue);
//二维数组查询
@PostMapping("/findTableValueToArray")
public DataValue[][] findTableValueToArray(@RequestParam("basicId") Long basicId, @RequestParam("sql") String sql, @RequestParam("one") Long one, @RequestParam("two") Integer two);
//添加数据值
@PostMapping("/product/addProduct")
public Result addProduct(@RequestParam("basicId") Long basicId, @RequestParam("tableId") Long tableId, @RequestBody DataValue[][] listList);
}

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

@ -110,6 +110,7 @@ public class TaskServiceImpl extends ServiceImpl<TaskMapper, TaskInfo> implement
TaskOutputInfo taskOutputInfo = taskOutputService.getOne(outputInfo);
String[] outPutFileName = taskOutputInfo.getLastFileName().split(",");
String[] newFileName = taskOutputInfo.getNewFileName().split(",");
Integer two = newFileName.length;
HashMap<String, String> fieldAsNameMap = new HashMap<>();
HashMap<String, String> newAndOldMap = new HashMap<>();
HashSet<String> basicList = new HashSet<>();
@ -195,7 +196,7 @@ public class TaskServiceImpl extends ServiceImpl<TaskMapper, TaskInfo> implement
String finalSql = sql;
log.info("sql为{}",finalSql);
executeUrgently(() -> {
selectAndAdd(count, finalDatabaseId, basicId, finalSql, tableId,newAndOldMap);
selectAndAdd(count, finalDatabaseId, basicId, finalSql, tableId,newAndOldMap,two);
});
}
@ -204,7 +205,7 @@ public class TaskServiceImpl extends ServiceImpl<TaskMapper, TaskInfo> implement
Long finalDatabaseId = databaseId;
String finalSql = sql;
executeHigh(() -> {
selectAndAdd(count, finalDatabaseId, basicId, finalSql, tableId,newAndOldMap);
selectAndAdd(count, finalDatabaseId, basicId, finalSql, tableId,newAndOldMap,two);
});
}
@ -213,7 +214,7 @@ public class TaskServiceImpl extends ServiceImpl<TaskMapper, TaskInfo> implement
Long finalDatabaseId = databaseId;
String finalSql = sql;
executeMedium(() -> {
selectAndAdd(count, finalDatabaseId, basicId, finalSql, tableId,newAndOldMap);
selectAndAdd(count, finalDatabaseId, basicId, finalSql, tableId,newAndOldMap,two);
});
}
@ -222,7 +223,7 @@ public class TaskServiceImpl extends ServiceImpl<TaskMapper, TaskInfo> implement
Long finalDatabaseId = databaseId;
String finalSql = sql;
executeLow(() -> {
selectAndAdd(count, finalDatabaseId, basicId, finalSql, tableId,newAndOldMap);
selectAndAdd(count, finalDatabaseId, basicId, finalSql, tableId,newAndOldMap,two);
});
}
@ -233,27 +234,24 @@ public class TaskServiceImpl extends ServiceImpl<TaskMapper, TaskInfo> implement
}
private void selectAndAdd(Integer count,Long databaseId,Long basicId,String sql,Long tableId,
HashMap<String,String> newAndOldMap) {
int pageSize = 1000;
HashMap<String,String> newAndOldMap,Integer two) {
int pageSize = 10000;
long firstArray = 0L;
int totalSegments = (int) Math.ceil((double) count / pageSize);
for (int i = 0; i < totalSegments; i++) {
log.info("调用第{}次", i);
String limit = " limit " + i * pageSize + "," + pageSize;
firstArray = count - (long) i * pageSize;
firstArray = Math.min(pageSize, count - i * pageSize);
if (firstArray >= pageSize) {
firstArray = pageSize;
}
String limitSelect = sql + limit;
log.info("执行查询语句为{}", limitSelect);
log.info(databaseId);
Result<List<List<DataValue>>> tableValue = dataValueClient.findTableValue(databaseId, limitSelect);
List<List<DataValue>> data = tableValue.getData();
log.info("远程调用完毕,调用数量{}",data.size());
for (List<DataValue> datum : data) {
log.info(datum.toString());
for (DataValue dataValue : datum) {
DataValue[][] tableValue = dataValueClient.findTableValueToArray(databaseId, limitSelect, firstArray,two);
log.info("远程调用完毕,调用数量{}",tableValue.length);
for (DataValue[] dataValues : tableValue) {
for (DataValue dataValue : dataValues) {
log.info(dataValue.toString());
String key = dataValue.getKey();
log.info("传来的值的键为{}",key);
@ -264,9 +262,10 @@ public class TaskServiceImpl extends ServiceImpl<TaskMapper, TaskInfo> implement
dataValue.setKey(newKey);
}
}
Result result = dataValueClient.addTableDataValue(basicId, tableId, data);
Result result = dataValueClient.addProduct(basicId, tableId, tableValue);
log.info("添加结果为{}", result);
log.info("添加到queue里成功");
}
// static class Worker implements Runnable {