gtsoft/gtsoft-snail-job-server
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects 1 Releases Wiki Activity

fix:(1.2.0-beta1.1): 1. 修复jdk17的线程池的核心现场

Browse Source
This commit is contained in:
opensnail 2024-09-09 11:29:01 +08:00
parent 6a9b26f692
commit 1ec1f98f3e
2 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/cache/ThreadPoolCache.java vendored
View File

@ -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<ThreadPoolExecutor> 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;

4
snail-job-client/snail-job-client-job-core/src/main/java/com/aizuda/snailjob/client/job/core/executor/AbstractJobExecutor.java
View File

@ -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);
// 将任务添加到时间轮中，到期停止任务