为什么我不能在 Java 中第二次将任务添加到线程池中？

Question

我创建了一个线程池来处理任务，处理完任务后，我发现我不能添加和启动另一个任务？如何修复？如果我用executor = new ThreadPoolExecutor(3, 3, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory("timeOutThread"));更改执行器，它会正常运行。但如果因为超时而取消任务，这会导致内存泄漏吗？

ExecutorService executor =   new ThreadPoolExecutor(3,
3, 0L,
TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1),
new NamedThreadFactory(
"timeOutThread")); 
List<Callable<String>> callableList = new ArrayList<>();
IntStream.range(0, 3).forEach(index -> {
callableList.add(() -> request(index));
});
List<Future<String>> futureList = executor.invokeAll(callableList, 1, TimeUnit.SECONDS); 
for (int i = 0; i < futureList.size(); i++) {
Future<String> future = futureList.get(i);
try {
list.add(future.get());
} catch (CancellationException e) {
log.info("timeOut task：{}", i);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
Thread.sleep(1000);
callableList.clear();
IntStream.range(0, 3).forEach(index -> {
callableList.add(() -> request(index));
});
long start1 = System.currentTimeMillis();
// Task java.util.concurrent.FutureTask@5fdcaa40 rejected from java.util.concurrent.ThreadPoolExecutor@6dc17b83   
List<Future<String>> futureList   = executor.invokeAll(callableList, 1, TimeUnit.SECONDS); 
for (int i = 0; i < futureList.size(); i++) {
Future<String> future = futureList.get(i);
try {
list.add(future.get());
} catch (CancellationException e) {
log.info("timeOut Task：{}", i);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
public String request() throws InterruptedException { 
TimeUnit.MILLISECONDS.sleep(200000);
return "A";
}

Answer 1

我可以用以下简化代码重现您的错误：

import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class Main {
public static void main(String[] args) throws InterruptedException {
var pool = new ThreadPoolExecutor(
3, 3, 0L, TimeUnit.NANOSECONDS, new LinkedBlockingQueue<>(1));
try {
System.out.println("Executing first batch of tasks...");
submitTasks(pool);
System.out.println("Executing second batch of tasks...");
submitTasks(pool);
} finally {
pool.shutdown();
}
}
private static void submitTasks(ExecutorService executor) throws InterruptedException {
var tasks = new ArrayList<Callable<Void>>(3);
for (int i = 0; i < 3; i++) {
tasks.add(() -> {
Thread.sleep(2_000L);
return null;
});
}
executor.invokeAll(tasks);
}
}

哪个输出：

Executing first batch of tasks...
Executing second batch of tasks...
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@87aac27[Not completed, task = Main$$Lambda$1/0x0000000800c009f0@816f27d] rejected from java.util.concurrent.ThreadPoolExecutor@3e3abc88[Running, pool size = 3, active threads = 0, queued tasks = 1, completed tasks = 3]
at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2070)
at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:833)
at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1365)
at java.base/java.util.concurrent.AbstractExecutorService.invokeAll(AbstractExecutorService.java:247)
at Main.submitTasks(Main.java:32)
at Main.main(Main.java:18)

问题是由于队列太小造成的。创建的LinkedBlockingQueue的容量只有一个，但同时向池中提交三个任务。因此，问题变成了，为什么它只在第二次调用invokeAll时失败？

原因与ThreadPoolExecutor的实现方式有关。第一次创建实例时，不会启动任何核心线程。它们是在提交任务时缓慢启动的。当任务的提交导致线程启动时，该任务会立即提供给该线程。队列被绕过。因此，当第一次调用invokeAll时，三个核心线程中的每一个都会启动，并且没有任何任务进入队列。

但是第二次调用invokeAll时，核心线程已经启动。由于提交任务不会导致创建线程，因此任务会被放入队列中。但是队列太小，导致RejectedExecutionException。如果你想知道为什么核心线程在保持活动时间设置为零的情况下仍然活动，那是因为默认情况下不允许核心线程因超时而死亡(你必须明确配置池才能允许)。

通过稍微修改代码，您可以看到这种延迟启动的核心线程是问题的原因。简单添加：

pool.prestartAllCoreThreads();

就在创建池之后，对invokeAll的第一次调用现在失败，并返回RejectedExecutionException。

此外，如果将队列的容量从1更改为3，则RejectedExecutionException将不再出现。

---

以下是一些相关文档：

任何BlockingQueue都可以用于转移和保留提交的任务。此队列的使用与池大小交互：

• 如果运行的线程少于corePoolSize，则Executor总是倾向于添加新线程，而不是排队
• 如果corePoolSize或更多线程正在运行，则Executor总是倾向于对请求进行排队，而不是添加新线程
• 如果请求无法排队，则会创建一个新线程，除非超过maximumPoolSize，在这种情况下，任务将被拒绝