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

修改超时时间

master
Cui YongXing 2024-09-10 19:26:48 +08:00
parent 50e1ad9d8b
commit 9bfc6e0c61
1 changed files with 56 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -43,7 +43,7 @@ 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 = 30000; private final Integer PAGE_SIZE = 100000;
@Resource @Resource
private TaskInfoMapper taskInfoMapper; private TaskInfoMapper taskInfoMapper;
@ -223,46 +223,66 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
private void extracted(Long data, Long taskId, String weight, String finalFieName, String finalJoint, Long basicId, Long newBasicId, Long tableId, HashMap<String, String> map, Integer num) { private void extracted(Long data, Long taskId, String weight, String finalFieName, String finalJoint, Long basicId, Long newBasicId, Long tableId, HashMap<String, String> map, Integer num) {
long count = data / PAGE_SIZE + (data % PAGE_SIZE > 0 ? 1 : 0); long count = data / PAGE_SIZE + (data % PAGE_SIZE > 0 ? 1 : 0);
ExecutorService executorService = Executors.newFixedThreadPool(10);
for (long i = 1; i <= count; i++) { for (long i = 1; i <= count; i++) {
long pageNum = (i - 1) * PAGE_SIZE; long pageNum = (i - 1) * PAGE_SIZE;
long pageSize = Math.min(PAGE_SIZE, data - pageNum); long pageSize = Math.min(PAGE_SIZE, data - pageNum);
if (Weight.high.getValue().equals(weight)) { executorService.submit(() -> {
log.info("执行高级任务");
submitHighPriorityTask(() -> {
getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map, pageSize, num, taskId); getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map, pageSize, num, taskId);
}); });
// if (Weight.high.getValue().equals(weight)) {
// log.info("执行高级任务");
// submitHighPriorityTask(() -> {
// getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map, pageSize, num, taskId);
// });
//
// }
// if (Weight.centre.getValue().equals(weight)) {
// submitMediumPriorityTask(() -> {
// getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map, pageSize, num, taskId);
// });
// }
// if (Weight.low.getValue().equals(weight)) {
//
// submitLowPriorityTask(() -> {
// getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map, pageSize, num, taskId);
// });
//
// }
// if (Weight.urgency.getValue().equals(weight)) {
// log.info("执行紧急任务");
// // 调整线程分配以适应紧急任务
// OptimizedPrioritizedThreadPool.activeEmergencyTasks.set(0);
// OptimizedPrioritizedThreadPool.remainingTasks.set(0);
// submitEmergencyTask(() -> {
// try {
// getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map, pageSize, num, taskId);
// } finally {
// // 减少剩余任务计数
// if (OptimizedPrioritizedThreadPool.remainingTasks.decrementAndGet() == 0) {
// System.out.println("All emergency tasks have completed.");
// }
// }
// });
//
//
// }
} }
if (Weight.centre.getValue().equals(weight)) { executorService.shutdown();
submitMediumPriorityTask(() -> {
getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map, pageSize, num, taskId);
});
}
if (Weight.low.getValue().equals(weight)) {
submitLowPriorityTask(() -> { // 等待所有任务完成（可选）
getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map, pageSize, num, taskId);
});
}
if (Weight.urgency.getValue().equals(weight)) {
log.info("执行紧急任务");
// 调整线程分配以适应紧急任务
OptimizedPrioritizedThreadPool.activeEmergencyTasks.set(0);
OptimizedPrioritizedThreadPool.remainingTasks.set(0);
submitEmergencyTask(() -> {
try { try {
getString(pageNum, finalFieName, finalJoint, basicId, newBasicId, tableId, map, pageSize, num, taskId); if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
} finally { // 如果在指定时间内没有完成，可以选择取消当前正在执行的任务
// 减少剩余任务计数 executorService.shutdownNow();
if (OptimizedPrioritizedThreadPool.remainingTasks.decrementAndGet() == 0) { // 然后根据需要处理超时情况
System.out.println("All emergency tasks have completed.");
}
}
});
} }
} catch (InterruptedException e) {
// 当前线程在等待过程中被中断
executorService.shutdownNow();
// 当前线程需要处理中断情况
Thread.currentThread().interrupt();
} }
} }