创建动态（增长/收缩）线程池

Question

我需要在Java（Java.util.concurrent）中实现一个线程池，它的线程数在空闲时处于某个最小值，当作业提交到线程池中时，线程数会增长到一个上限（但不会再增长），而当所有作业完成且不再提交作业时，线程数会收缩到下限。

您将如何实现这样的功能？我想这将是一个相当常见的使用场景，但显然是java。util。同时发生的Executorsfactory方法只能创建固定大小的池以及在提交多个作业时无限增长的池。ThreadPoolExecutor类提供了corePoolSize和maximumPoolSize参数，但其文档似乎暗示，同时拥有超过corePoolSize线程的唯一方法是使用有界作业队列，在这种情况下，如果你已经达到maximumPoolSizethreads，你会被拒绝工作，你必须自己处理吗？我想到了这个：

//pool creation
ExecutorService pool = new ThreadPoolExecutor(minSize, maxSize, 500, TimeUnit.MILLISECONDS,
    new ArrayBlockingQueue<Runnable>(minSize));
...

//submitting jobs
for (Runnable job : ...) {
    while (true) {
        try {
            pool.submit(job);
            System.out.println("Job " + job + ": submitted");
            break;
        } catch (RejectedExecutionException e) {
            // maxSize jobs executing concurrently atm.; re-submit new job after short wait
            System.out.println("Job " + job + ": rejected...");
            try {
                Thread.sleep(300);
            } catch (InterruptedException e1) {
            }
        }
    }
}

我是不是忽略了什么？有更好的方法吗？此外，根据用户的需求，在至少（我认为）（作业总数）-maxSize作业完成之前，上述代码可能无法完成。因此，如果您希望能够将任意数量的作业提交到池中，并且在不等待任何作业完成的情况下立即继续，那么如果没有一个专用的“job sumitting”线程来管理所需的无限队列以容纳所有提交的作业，我看不出您如何做到这一点。AFAICS，如果您对ThreadPoolExecutor本身使用一个无界队列，那么它的线程计数永远不会超过corePoolSize。

Answer 1

设置最大池大小为整数。MAX_VALUE。如果你有超过20亿线程...祝你好运。

无论如何，ThreadPoolExecutor的Javadoc声明：

通过将maximumPoolSize设置为本质上无界的值，如Integer。最大值时，允许池容纳任意数量的并发任务。最典型的情况是，核心池和最大池大小仅在构建时设置，但也可以使用setCorePoolSize（int）和setMaximumPoolSize（int）动态更改。

对于类似于LinkedBlockingQueue这样的无界任务队列，其容量应该是任意大的。

Answer 2

一个可能对您有帮助的技巧是分配一个Reject tedExektionHandler，它使用相同的线程将作业提交到阻塞队列中。这将阻塞当前线程，并消除对某种循环的需求。

在这里看到我的答案：

如果ThreadPoolExecutor命令需要处理的数据太多，如何使其等待？

这是从该答案复制的拒绝处理程序。

final BlockingQueue queue = new ArrayBlockingQueue<Runnable>(200);
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(nThreads, nThreads,
       0L, TimeUnit.MILLISECONDS, queue);
// by default (unfortunately) the ThreadPoolExecutor will call the rejected
// handler when you submit the 201st job, to have it block you do:
threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
   public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
      // this will block if the queue is full
      executor.getQueue().put(r);
   }
});

然后，只要您意识到在核心线程之上创建任何线程之前，您首先使用的有界阻塞队列会被填满，您就可以使用核心/最大线程计数。因此，如果您有10个核心线程，并且您希望第11个作业启动第11个线程，那么您将需要一个大小为0的阻塞队列（可能是一个SynChronousQueue）。我觉得这是一个真正的限制在其他伟大的ExecutorService类。

Answer 3

随着线程的增长和收缩，我想到的只有一个名字：来自java的CachedThreadPool。util。并发包。

ExecutorService executor = Executors.newCachedThreadPool();

CachedThreadPool（）可以重用线程，并在需要时创建新线程。是的，如果一个线程空闲60秒，CachedThreadPool将杀死它。因此，这是相当轻量级的——用你的话来说，增长和收缩！