ThreadPoolExecutor的无界队列没有创建新线程

Question

我的ThreadPoolExecutor无法创建新线程。事实上，我写了一个有点蹩脚的LinkedBlockingQueue，它将接受任何任务(即它是无界的)，但调用一个额外的处理程序——在我的应用程序中，它会发出池落后的警告跟踪——这给了我非常明确的信息，即TPE拒绝创建新线程，即使队列中有数千个条目。我的构造函数如下:

private final ExecutorService s3UploadPool = 
new ThreadPoolExecutor(1, 40, 1, TimeUnit.HOURS, unboundedLoggingQueue);

为什么不创建新的线程?

Answer 1

这篇博文讨论了这个问题:

这个线程池的构造不会像预期的那样工作。这是由于ThreadPoolExecutor中的逻辑，如果无法向队列提供任务，则会添加新线程。在我们的例子中，我们使用了一个无界的LinkedBlockingQueue，在这里我们总是可以向队列提供一个任务。这实际上意味着我们永远不会超过核心池大小并达到最大池大小。

如果您还需要将最小池大小与最大池大小解耦，则必须进行一些扩展编码。我不知道存在于Java库或Apache Commons中的解决方案。解决方案是创建一个耦合的BlockingQueue，它可以感知TPE，并且如果它知道TPE没有可用的线程，那么它将以自己的方式拒绝任务，然后手动请求。在链接的帖子中有更详细的介绍。最终你的结构看起来像:

public static ExecutorService newScalingThreadPool(int min, int max, long keepAliveTime) {
   ScalingQueue queue = new ScalingQueue();
   ThreadPoolExecutor executor =
      new ScalingThreadPoolExecutor(min, max, keepAliveTime, TimeUnit.MILLISECONDS, queue);
   executor.setRejectedExecutionHandler(new ForceQueuePolicy());
   queue.setThreadPoolExecutor(executor);
   return executor;
}

然而，更简单地设置corePoolSize为maxPoolSize，不用担心这些废话。

Answer 2

有一个解决这个问题的方法。考虑以下实现:

int corePoolSize = 40;
int maximumPoolSize = 40;
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 
    60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
threadPoolExecutor.allowCoreThreadTimeOut(true);

通过设置allowCoreThreadTimeOut()为true，允许线程池中的线程在指定的超时时间(本例中为60秒)后终止。在这个解决方案中，实际上是corePoolSize构造函数参数决定了最大池大小，因为线程池将增长到corePoolSize，然后开始向队列添加作业。池可能永远不会超过这个值，因为池在队列满之前不会产生新的线程(考虑到LinkedBlockingQueue具有Integer.MAX_VALUE的容量，这可能永远不会发生)。因此，将maximumPoolSize设置为比corePoolSize大的值没有什么意义。

注意事项:线程池在超时后有0个空闲线程，这意味着在创建线程之前会有一些延迟(通常，您总是有corePoolSize线程可用)。

更多细节可以在ThreadPoolExecutor的JavaDoc中找到。

Answer 3

正如@ djhlin所提到的，这是ThreadPoolExecutor定义行为的一部分(让许多人感到惊讶)。我相信我已经找到了一个比较优雅的解决方案来解决这个问题，我在这里给出了答案:

如何让ThreadPoolExecutor在排队之前将线程增加到最大?

基本上，您扩展LinkedBlockingQueue，使queue.offer(...)始终返回false，这将在必要时向池中添加额外的线程。如果线程池已经达到最大线程数，并且它们都很忙，则会调用RejectedExecutionHandler。它是处理程序，然后将put(...)放入队列。

查看我的代码