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

紧急任务线程修改

master
Cui YongXing 2024-09-05 20:09:28 +08:00
parent 366a3e613e
commit aea5c583b2
2 changed files with 10 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -277,7 +277,7 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
if (Weight.urgency.getValue().equals(weight)){ if (Weight.urgency.getValue().equals(weight)){
log.info("执行紧急任务"); log.info("执行紧急任务");
// 调整线程分配以适应紧急任务 // 调整线程分配以适应紧急任务
adjustForEmergency();
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;
@ -290,8 +290,7 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
} catch (InterruptedException e) { } catch (InterruptedException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
// 任务完成后恢复默认线程分配
restoreDefaults();
} }
long end = System.currentTimeMillis(); long end = System.currentTimeMillis();
log.info("执行时间：{}",end-start); log.info("执行时间：{}",end-start);

11
cloud-task-server/src/main/java/com/muyu/task/server/thread/OptimizedPrioritizedThreadPool.java
View File

@ -1,6 +1,7 @@
package com.muyu.task.server.thread; package com.muyu.task.server.thread;
import java.util.concurrent.*; import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class OptimizedPrioritizedThreadPool { public class OptimizedPrioritizedThreadPool {
private static final ExecutorService executor; private static final ExecutorService executor;
@ -20,7 +21,7 @@ public class OptimizedPrioritizedThreadPool {
private static final int emergencyLowThreads = 5; private static final int emergencyLowThreads = 5;
private static volatile boolean inEmergencyMode = false; private static volatile boolean inEmergencyMode = false;
private static final AtomicInteger activeEmergencyTasks = new AtomicInteger(0);// 紧急任务计数器
static { static {
if (defaultHighThreads + defaultMediumThreads + defaultLowThreads > totalThreads) { if (defaultHighThreads + defaultMediumThreads + defaultLowThreads > totalThreads) {
throw new IllegalArgumentException("Default priority threads exceed total threads."); throw new IllegalArgumentException("Default priority threads exceed total threads.");
@ -42,7 +43,9 @@ public class OptimizedPrioritizedThreadPool {
} }
public static void submitEmergencyTask(Runnable task) { public static void submitEmergencyTask(Runnable task) {
if (activeEmergencyTasks.incrementAndGet()==1){
adjustForEmergency();
}
try { try {
emergencySemaphore.acquire(); emergencySemaphore.acquire();
executor.submit(() -> { executor.submit(() -> {
@ -50,7 +53,9 @@ public class OptimizedPrioritizedThreadPool {
task.run(); task.run();
} finally { } finally {
emergencySemaphore.release(); emergencySemaphore.release();
if (activeEmergencyTasks.decrementAndGet()==0){
restoreDefaults();
}
} }
}); });
} catch (InterruptedException e) { } catch (InterruptedException e) {