From 1ec1f98f3e76e60dcd7e6955ca87c9f5ba576189 Mon Sep 17 00:00:00 2001 From: opensnail <598092184@qq.com> Date: Mon, 9 Sep 2024 11:29:01 +0800 Subject: [PATCH] =?UTF-8?q?fix:(1.2.0-beta1.1):=201.=20=E4=BF=AE=E5=A4=8Dj?= =?UTF-8?q?dk17=E7=9A=84=E7=BA=BF=E7=A8=8B=E6=B1=A0=E7=9A=84=E6=A0=B8?= =?UTF-8?q?=E5=BF=83=E7=8E=B0=E5=9C=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../snailjob/client/job/core/cache/ThreadPoolCache.java | 8 ++++---- .../client/job/core/executor/AbstractJobExecutor.java | 4 +++- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/cache/ThreadPoolCache.java b/snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/cache/ThreadPoolCache.java index ac6a265c7..a2d17a268 100644 --- a/snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/cache/ThreadPoolCache.java +++ b/snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/cache/ThreadPoolCache.java @@ -25,20 +25,20 @@ public class ThreadPoolCache { public static ThreadPoolExecutor createThreadPool(Long taskBatchId, int parallelNum) { if (CACHE_THREAD_POOL.containsKey(taskBatchId)) { ThreadPoolExecutor cacheThreadPool = CACHE_THREAD_POOL.get(taskBatchId); - if (cacheThreadPool.getCorePoolSize() == parallelNum) { + // 大于1说明已经更新了线程池的线程数，为了防止后面任务执行过程任务并行度改变影响已经产生的批次，这里不再做更新操作 + if (cacheThreadPool.getCorePoolSize() > 1) { return cacheThreadPool; } // 若能执行到这里只有分片任务(静态分片、MAP、MapReduce)才会需要多线程支持 - cacheThreadPool.setCorePoolSize(parallelNum); - cacheThreadPool.setMaximumPoolSize(parallelNum); + cacheThreadPool.setCorePoolSize(Math.min(parallelNum, cacheThreadPool.getMaximumPoolSize())); return cacheThreadPool; } Supplier supplier = () -> { ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( // 默认情况先只设置一个线程, 只有分片任务(静态分片、MAP、MapReduce)才会需要多线程支持 - 1, 1, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), + 1, parallelNum, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new CustomizableThreadFactory(MessageFormat.format("snail-job-job-{0}-", taskBatchId))); threadPoolExecutor.allowCoreThreadTimeOut(true); return threadPoolExecutor; diff --git a/snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/executor/AbstractJobExecutor.java b/snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/executor/AbstractJobExecutor.java index 35b800572..f57f1b7fb 100644 --- a/snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/executor/AbstractJobExecutor.java +++ b/snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/executor/AbstractJobExecutor.java @@ -26,6 +26,7 @@ import lombok.extern.slf4j.Slf4j; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -43,7 +44,8 @@ public abstract class AbstractJobExecutor implements IJobExecutor { public void jobExecute(JobContext jobContext) { // 创建可执行的任务 - ThreadPoolExecutor threadPool = ThreadPoolCache.createThreadPool(jobContext.getTaskBatchId(), jobContext.getParallelNum()); + Integer parallelNum = Optional.ofNullable(jobContext.getParallelNum()).orElse(1); + ThreadPoolExecutor threadPool = ThreadPoolCache.createThreadPool(jobContext.getTaskBatchId(), Math.max(1, parallelNum)); ListeningExecutorService decorator = MoreExecutors.listeningDecorator(threadPool); // 将任务添加到时间轮中，到期停止任务