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

Question

我创建了一个线程池来处理任务，处理完任务后，我发现我无法添加和启动其他任务？如何解决？如果我通过 executor = new ThreadPoolExecutor(3, 3, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory("timeOutThread")); 更改执行者,它会运行OK。但是如果任务因为超时而被取消，这样做会导致内存泄漏吗？

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。如果您想知道为什么尽管将保持 Activity 时间设置为零但核心线程仍然存在，那是因为默认情况下不允许核心线程因超时而死亡(您必须显式配置池以允许这样做)。

稍微修改一下代码就可以看出这个lazily-started-core-threads是问题的原因。只需添加:

pool.prestartAllCoreThreads();

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

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

---

这里有一些相关的 documentation :

Any BlockingQueue may be used to transfer and hold submitted tasks. The use of this queue interacts with pool sizing:

• If fewer than corePoolSize threads are running, the Executor always prefers adding a new thread rather than queuing.
• If corePoolSize or more threads are running, the Executor always prefers queuing a request rather than adding a new thread.
• If a request cannot be queued, a new thread is created unless this would exceed maximumPoolSize, in which case, the task will be rejected.