【发布时间】：2020-03-11 05:47:45
【问题描述】：
我遇到了一个问题，如果我在创建池后尝试将 ThreadPoolExecutor 的核心池大小调整为不同的数字，然后间歇性地，一些任务会被 RejectedExecutionException 拒绝，即使我提交的任务数永远不要超过queueSize + maxPoolSize。
我要解决的问题是扩展ThreadPoolExecutor，它会根据线程池队列中的待处理执行来调整其核心线程的大小。我需要这个，因为默认情况下，ThreadPoolExecutor 只有在队列已满时才会创建一个新的Thread。
这是一个小型的自包含的 Pure Java 8 程序来演示该问题。
import static java.lang.Math.max;
import static java.lang.Math.min;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolResizeTest {
public static void main(String[] args) throws Exception {
// increase the number of iterations if unable to reproduce
// for me 100 iterations have been enough
int numberOfExecutions = 100;
for (int i = 1; i <= numberOfExecutions; i++) {
executeOnce();
}
}
private static void executeOnce() throws Exception {
int minThreads = 1;
int maxThreads = 5;
int queueCapacity = 10;
ThreadPoolExecutor pool = new ThreadPoolExecutor(
minThreads, maxThreads,
0, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(queueCapacity),
new ThreadPoolExecutor.AbortPolicy()
);
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
scheduler.scheduleAtFixedRate(() -> resizeThreadPool(pool, minThreads, maxThreads),
0, 10, TimeUnit.MILLISECONDS);
CompletableFuture<Void> taskBlocker = new CompletableFuture<>();
try {
int totalTasksToSubmit = queueCapacity + maxThreads;
for (int i = 1; i <= totalTasksToSubmit; i++) {
// following line sometimes throws a RejectedExecutionException
pool.submit(() -> {
// block the thread and prevent it from completing the task
taskBlocker.join();
});
// Thread.sleep(10); //enabling even a small sleep makes the problem go away
}
} finally {
taskBlocker.complete(null);
scheduler.shutdown();
pool.shutdown();
}
}
/**
* Resize the thread pool if the number of pending tasks are non-zero.
*/
private static void resizeThreadPool(ThreadPoolExecutor pool, int minThreads, int maxThreads) {
int pendingExecutions = pool.getQueue().size();
int approximateRunningExecutions = pool.getActiveCount();
/*
* New core thread count should be the sum of pending and currently executing tasks
* with an upper bound of maxThreads and a lower bound of minThreads.
*/
int newThreadCount = min(maxThreads, max(minThreads, pendingExecutions + approximateRunningExecutions));
pool.setCorePoolSize(newThreadCount);
pool.prestartAllCoreThreads();
}
}
如果我从不提交超过 queueCapacity+maxThreads 的请求，为什么池会抛出 RejectedExecutionException。我从不更改最大线程数，因此根据 ThreadPoolExecutor 的定义，它应该将任务容纳在 Thread 或队列中。
当然，如果我从不调整池大小，那么线程池永远不会拒绝任何提交。这也很难调试，因为在提交中添加任何类型的延迟都会使问题消失。
关于如何修复 RejectedExecutionException 的任何指示？
【问题讨论】：
-
为什么不通过包装现有的
ExecutorService来提供您自己的ExecutorService实现，它会重新提交由于调整大小而提交失败的任务？ -
@daniu 这是一种解决方法。问题的重点是，如果我从不提交超过 queueCapacity+maxThreads 的内容，为什么池会抛出 RejectedExecutionException。我从不更改最大线程数，因此根据 ThreadPoolExecutor 的定义，它应该将任务容纳在线程中或队列中。
-
好吧，我好像误解了你的问题。它是什么？你想知道为什么会发生这种行为，或者你是如何解决这个问题的吗？
-
是的，将我的实现更改为执行器服务是不可行的，因为很多代码都引用了 ThreadPoolExecutor。因此，如果我仍然想要一个可调整大小的 ThreadPoolExecutor，我需要知道如何修复它。做这样的事情的正确方法可能是扩展 ThreadPoolExecutor 并访问它的一些受保护变量，并在超类共享的锁上的同步块内更新池大小。
-
扩展
ThreadPoolExecutor很可能是个坏主意，在这种情况下您是否也不需要更改现有代码？最好提供一些实际代码如何访问执行程序的示例。如果它使用了许多特定于ThreadPoolExecutor的方法（即不在ExecutorService中），我会感到惊讶。
标签： java multithreading