具有 corePoolSize 0 的 ThreadPoolExecutor

Question

我正在学习Java Concurrency In Practice并被困在8.3.1 线程创建和拆卸主题上。以下脚注警告保持corePoolSize零。开发人员有时会试图将核心大小设置为零，这样工作线程最终会被拆除，因此不会阻止 JVM 退出，但这可能会在不使用SynchronousQueue 用于他们的工作队列（就像 newCachedThreadPool 一样）。如果池已经达到核心大小，只有在工作队列已满时，ThreadPoolExecutor 才会创建新线程。因此，提交到具有任何容量和核心大小为零的工作队列的线程池的任务将在队列填满之前不会执行，这通常不是所期望的。所以为了验证这一点，我写了这个程序，它不像上面所说的那样工作。    final int corePoolSize = 0;    ThreadPoolExecutor tp = new ThreadPoolExecutor(corePoolSize, 1, 5, TimeUnit.SECONDS,            new LinkedBlockingQueue<>());    // If the pool is already at the core size    if (tp.getPoolSize() == corePoolSize) {        ExecutorService ex = tp;        // So tasks submitted to a thread pool with a work queue that has any capacity        // and a core size of zero will not execute until the queue fills up.        // So, this should not execute until queue fills up.        ex.execute(() -> System.out.println("Hello"));    }输出： 你好那么，程序的行为是否表明ThreadPoolExecutor如果提交了一个任务而不管corePoolSize=0. 如果是，那么教科书上的警告是什么。编辑：根据@SK 的建议测试了jdk1.5.0_22 中的代码，并进行了以下更改：ThreadPoolExecutor tp = new ThreadPoolExecutor(corePoolSize, 1, 5, TimeUnit.SECONDS,                new LinkedBlockingQueue<Runnable>(1));//Queue size is set to 1.但是随着这个变化，程序终止而不打印任何输出。那么我是否误解了书中的这些陈述？编辑（@sjlee）：很难在评论中添加代码，所以我将它作为编辑添加在这里...你能尝试这个修改并针对最新的 JDK 和 JDK 1.5 运行它吗？final int corePoolSize = 0;ThreadPoolExecutor tp = new ThreadPoolExecutor(corePoolSize, 1, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<>());// If the pool is already at the core sizeif (tp.getPoolSize() == corePoolSize) {    ExecutorService ex = tp;    // So tasks submitted to a thread pool with a work queue that has any capacity    // and a core size of zero will not execute until the queue fills up.    // So, this should not execute until queue fills up.    ex.execute(() -> System.out.println("Hello"));}tp.shutdown();if (tp.awaitTermination(1, TimeUnit.SECONDS)) {    System.out.println("thread pool shut down. exiting.");} else {    System.out.println("shutdown timed out. exiting.");}

Answer 1

在 jdk 1.5、1.6、1.7 和 1.8 中运行这个程序时，我发现ThreadPoolExecutor#execute(Runnable)了 1.5、1.6 和 1.7+ 中的不同实现。这是我发现的：

JDK 1.5 实现

 //Here poolSize is the number of core threads running.

 public void execute(Runnable command) {

    if (command == null)

        throw new NullPointerException();

    for (;;) {

        if (runState != RUNNING) {

            reject(command);

            return;

        }

        if (poolSize < corePoolSize && addIfUnderCorePoolSize(command))

            return;

        if (workQueue.offer(command))

            return;

        Runnable r = addIfUnderMaximumPoolSize(command);

        if (r == command)

            return;

        if (r == null) {

            reject(command);

            return;

        }

        // else retry

    }

}

当corePoolSize为 0 时，此实现不会创建线程，因此不会执行提供的任务。

JDK 1.6 实现

//Here poolSize is the number of core threads running.

  public void execute(Runnable command) {

    if (command == null)

        throw new NullPointerException();

    if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {

        if (runState == RUNNING && workQueue.offer(command)) {

            if (runState != RUNNING || poolSize == 0)

                ensureQueuedTaskHandled(command);

        }

        else if (!addIfUnderMaximumPoolSize(command))

            reject(command); // is shutdown or saturated

    }

}

即使corePoolSize为 0，JDK 1.6 也会创建一个新线程。

JDK 1.7+ 实现（类似于 JDK 1.6，但具有更好的锁和状态检查）

    public void execute(Runnable command) {

    if (command == null)

        throw new NullPointerException();

    /*

     * Proceed in 3 steps:

     *

     * 1. If fewer than corePoolSize threads are running, try to

     * start a new thread with the given command as its first

     * task.  The call to addWorker atomically checks runState and

     * workerCount, and so prevents false alarms that would add

     * threads when it shouldn't, by returning false.

     *

     * 2. If a task can be successfully queued, then we still need

     * to double-check whether we should have added a thread

     * (because existing ones died since last checking) or that

     * the pool shut down since entry into this method. So we

     * recheck state and if necessary roll back the enqueuing if

     * stopped, or start a new thread if there are none.

     *

     * 3. If we cannot queue task, then we try to add a new

     * thread.  If it fails, we know we are shut down or saturated

     * and so reject the task.

     */

    int c = ctl.get();

    if (workerCountOf(c) < corePoolSize) {

        if (addWorker(command, true))

            return;

        c = ctl.get();

    }

    if (isRunning(c) && workQueue.offer(command)) {

        int recheck = ctl.get();

        if (! isRunning(recheck) && remove(command))

            reject(command);

        else if (workerCountOf(recheck) == 0)

            addWorker(null, false);

    }

    else if (!addWorker(command, false))

        reject(command);

}

即使corePoolSize是 0，JDK 1.7 也会创建一个新线程。

因此，corePoolSize=0在 JDK 1.5 和 JDK 1.6+ 的每个版本中，这似乎都是一个特例。

但奇怪的是，书中的解释与任何程序结果都不符。

Answer 2

似乎这是旧 Java 版本的错误，但现在在 Java 1.8 中不存在。

根据来自的 Java 1.8 文档ThreadPoolExecutor.execute()：

     /*

     * Proceed in 3 steps:

     *

     * 1. If fewer than corePoolSize threads are running, try to

     * start a new thread with the given command as its first

     * task.  The call to addWorker atomically checks runState and

     * workerCount, and so prevents false alarms that would add

     * threads when it shouldn't, by returning false.

     *

     * 2. If a task can be successfully queued, then we still need

     * to double-check whether we should have added a thread

     * (because existing ones died since last checking) or that

     * the pool shut down since entry into this method. So we

     * recheck state and if necessary roll back the enqueuing if

     * stopped, or start a new thread if there are none.

     * ....

     */

第二点，在加入一个worker到队列后，再检查一下，如果不是排队任务，可以启动一个新线程，而不是回滚入队并启动一个新线程。

这就是正在发生的事情。在第一次检查期间，任务已排队，但在重新检查期间，将启动一个新线程来执行您的任务。