ThreadPoolExecutor配置

Question

我有一种情况，我希望添加到线程池的速度比处理快。我认为无界队列不是一个好主意，因为如果不进行检查，队列可能会增长到吞噬所有内存的足够数据。鉴于此，我正在尝试确定ThreadPoolExecutor的正确设置。

我的第一个想法是使用直接切换和调用者运行故障策略的固定线程池。但我想知道这是否会损害吞吐量(因为每次调用调用者运行策略时，线程池任务可能会完成并闲置一段时间)。

另一个想法是用ArrayBlockingQueue修复线程池，但我实际上不确定它的行为。我希望这意味着执行器更喜欢创建线程，如果小于coreThread大小，则创建队列，如果队列已满，则阻塞等待队列获得空间。但在阅读这里的文档时：

http://docs.oracle.com/javase/1.5.0/docs/api/java/util/concurrent/ThreadPoolExecutor.html

它似乎更喜欢创建直到corePoolSize的线程，然后添加到队列中，如果队列已满，它将尝试创建直到maxThreads的线程(与本例中的coreThreads相同)，如果失败，它将运行失败策略。

有没有人能澄清上面这个案例的行为？并建议对于这种特定情况的最佳设置是什么(我建议的想法之一或其他可能更好的想法)？

Answer 1

我想我给出了另一个答案，因为它对同一个问题有不同的解决方案。

您可以只使用ThreadPoolExecutor和信号量。信号量将以您希望队列中允许的最大数量创建，并且在每个线程完成执行后，您将调用release (beforeExecute，当项目被从队列中拉出时)

Semaphore semaphore = new Semaphore(1000);
ThreadPoolExecutor executor = new ThreadPoolExecutor(5,10,60,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>()){
  protected void beforeExecute(Runnable r, Throwable t) { 
     semaphore.release();
  }
}

public void doSubmit(Runnable r){
  sempahore.acquire();
  executor.submit(r);      
}

所以在这里，所有线程都将挂起，直到有可用的许可(队列中的条目)。

Answer 2

当当前正在使用所有线程时，ThreadPoolExecutor将创建更多的线程。这意味着队列可以是空的，但如果所有线程都在运行以前的任务，则新任务将创建一个新线程，直到达到最大值。

如果队列已满，且所有线程都已饱和，则ThreadPoolExecutor实际上将拒绝该任务并抛出RejectedExecutionException。因此，使用BlockingQueue实际上不会有预期的结果。

如果想要限制当前队列中的任务数量，可以使用ExecutorCompletionService和后备队列。

//core 5 max 10 with 60 second idle time
ThreadPoolExecutor executor = new ThreadPoolExecutor(5,10,60,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
ExecutorCompletionService completionService = new ExecutorCompletionService(executor);
private final static int MAX_IN_QUEUE = 1000; 

public void doSubmit(Runnable r){
    while(executor.getQueue().size() >= MAX_IN_QUEUE) 
       completionService.poll(100,TimeUnit.MILLISECONDS);
    completionService.submit(r);
}

这有一个明显的副作用，那就是必须不断地等待一个元素完成。我在条件上循环，因为可能的竞争条件，而在进入块后实际上是真正的非实质性。

当然，多个提交的竞争条件，但它应该足够节流，以防止过度拥挤的队列。这可以通过简单地同步doSubmit方法来解决。

Answer 3

如果一个线程发出所有请求，那么一个无界队列将按您的喜好阻塞，而不会受到队列增长的负面影响。

如果多个线程发出请求，则具有调用者运行的固定池策略应按您所希望的那样工作。池中的其余线程将由其他请求线程保持活动状态。