如何使ThreadPoolExecutor的submit()方法块如果饱和?

我想创build一个ThreadPoolExecutor ,以便当它达到其最大值并且队列已满时, sumbit()方法会在尝试添加新任务时阻塞。 我是否需要实现一个自定义的RejectedExecutionHandler ,或者是否有一个现有的方法来使用标准的Java库做到这一点?

我刚刚find的可能解决scheme之一:

 public class BoundedExecutor { private final Executor exec; private final Semaphore semaphore; public BoundedExecutor(Executor exec, int bound) { this.exec = exec; this.semaphore = new Semaphore(bound); } public void submitTask(final Runnable command) throws InterruptedException, RejectedExecutionException { semaphore.acquire(); try { exec.execute(new Runnable() { public void run() { try { command.run(); } finally { semaphore.release(); } } }); } catch (RejectedExecutionException e) { semaphore.release(); throw e; } } } 

还有其他解决scheme吗? 我更喜欢基于RejectedExecutionHandler东西,因为它似乎是处理这种情况的标准方法。

你可以使用ThreadPoolExecutor和一个blockingQueue:

 public class ImageManager { BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(blockQueueSize); RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy(); private ExecutorService executorService = new ThreadPoolExecutor(numOfThread, numOfThread, 0L, TimeUnit.MILLISECONDS, blockingQueue, rejectedExecutionHandler); private int downloadThumbnail(String fileListPath){ executorService.submit(new yourRunnable()); } } 

检查出这样做的四个select : 创build一个NotifyingBlockingThreadPoolExecutor

您应该使用CallerRunsPolicy ,它在调用线程中执行被拒绝的任务。 这样,在任务完成之前,它不能向执行器提交任何新的任务,此时将会有一些空闲的线程池或重复该过程。

http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/ThreadPoolExecutor.CallerRunsPolicy.html

从文档:

拒绝的任务

执行方法execute(java.lang.Runnable)中提交的新任务将在执行程序closures时被拒绝,并且执行程序对最大线程和工作队列的容量使用有限边界且饱和。 无论哪种情况,execute方法都会调用RejectedExecutionHandler的RejectedExecutionHandler.rejectedExecution(java.lang.Runnable,java.util.concurrent.ThreadPoolExecutor)方法。 提供四个预定义的处理程序策略

  1. 在默认的ThreadPoolExecutor.AbortPolicy中,处理程序在拒绝时抛出运行时RejectedExecutionException。
  2. 在ThreadPoolExecutor.CallerRunsPolicy中,调用执行本身的线程运行该任务。 这提供了一个简单的反馈控制机制,将减慢提交新任务的速度。
  3. 在ThreadPoolExecutor.DiscardPolicy中,无法执行的任务将被简单地删除。
  4. 在ThreadPoolExecutor.DiscardOldestPolicy中,如果执行程序没有closures,工作队列头部的任务将被删除,然后重试执行(可能会再次失败,导致重复执行)。

此外,确保在调用ThreadPoolExecutor构造函数时使用有界的队列,如ArrayBlockingQueue。 否则,什么都不会被拒绝。

编辑:为了响应您的评论,请将ArrayBlockingQueue的大小设置为等于线程池的最大大小,并使用AbortPolicy。

编辑2:好吧,我看到你在什么。 那么怎么样:重写beforeExecute()方法来检查getActiveCount()是否不超过getMaximumPoolSize() ,如果是的话,睡觉,然后再试一次?

Hibernate有一个BlockPolicy ,很简单,可以做你想做的事情:

请参阅: Executors.java

 /** * A handler for rejected tasks that will have the caller block until * space is available. */ public static class BlockPolicy implements RejectedExecutionHandler { /** * Creates a <tt>BlockPolicy</tt>. */ public BlockPolicy() { } /** * Puts the Runnable to the blocking queue, effectively blocking * the delegating thread until space is available. * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { try { e.getQueue().put( r ); } catch (InterruptedException e1) { log.error( "Work discarded, thread was interrupted while waiting for space to schedule: {}", r ); } } } 

上面引用的来自“ Java并发实践 ”中的BoundedExecutor答案只有在对Executor使用无限队列或信号量限制不大于队列大小时才能正常工作。 信号量在提交线程和池中的线程之间共享状态,即使队列大小<bound <=(队列大小+池大小),也可以使执行程序饱和。

使用CallerRunsPolicy只有在你的任务不能永久运行的情况下才有效,在这种情况下,你的提交线程将永远保持在rejectedExecution ,如果你的任务需要很长时间才能运行,这是一个坏主意,因为提交线程不能提交任何新的任务或任何其他任何事情,如果它正在运行一个任务本身。

如果这是不可接受的,那么我build议在提交任务之前检查执行程序的有界队列的大小。 如果队列已满,请稍等片刻再尝试提交。 吞吐量将会受到影响,但是我build议这个解决scheme比许多其他解决scheme更简单,并保证没有任何任务会被拒绝。

我知道,这是一个黑客,但在我看来,这里提供的最干净的黑客;-)

由于ThreadPoolExecutor使用阻塞队列“offer”而不是“put”,因此可以覆盖阻塞队列的“offer”行为:

 class BlockingQueueHack<T> extends ArrayBlockingQueue<T> { BlockingQueueHack(int size) { super(size); } public boolean offer(T task) { try { this.put(task); } catch (InterruptedException e) { throw new RuntimeException(e); } return true; } } ThreadPoolExecutor tp = new ThreadPoolExecutor(1, 2, 1, TimeUnit.MINUTES, new BlockingQueueHack(5)); 

我testing了它,它似乎工作。 实施一些超时政策是留给读者的练习。

下面的类包装一个ThreadPoolExecutor并使用信号量来阻塞工作队列已满:

 public final class BlockingExecutor { private final Executor executor; private final Semaphore semaphore; public BlockingExecutor(int queueSize, int corePoolSize, int maxPoolSize, int keepAliveTime, TimeUnit unit, ThreadFactory factory) { BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(); this.executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, unit, queue, factory); this.semaphore = new Semaphore(queueSize + maxPoolSize); } private void execImpl (final Runnable command) throws InterruptedException { semaphore.acquire(); try { executor.execute(new Runnable() { @Override public void run() { try { command.run(); } finally { semaphore.release(); } } }); } catch (RejectedExecutionException e) { // will never be thrown with an unbounded buffer (LinkedBlockingQueue) semaphore.release(); throw e; } } public void execute (Runnable command) throws InterruptedException { execImpl(command); } } 

这个包装类是基于Brian Goetz的“Java并发实践”(Java Concurrency in Practice)一书中给出的解决scheme。 本书中的解决scheme只有两个构造函数参数:一个Executor和一个用于信号量的边界。 这在Fixpoint给出的答案中显示。 这种方法存在一个问题:它可能处于线程繁忙,队列已满,但信号量刚刚释放许可的状态。 (在finally块中的semaphore.release() )。 在这种状态下,一个新的任务可以抓取刚刚释放的许可证,但是由于任务队列已满而被拒绝。 当然这不是你想要的; 你想在这种情况下阻止。

为了解决这个问题,我们必须使用一个无限的队列,正如JCiP明确提到的那样。 信号量充当警卫,给出虚拟队列大小的效果。 这有副作用,有可能单位可以包含maxPoolSize + virtualQueueSize + maxPoolSize任务。 这是为什么? 由于finally块中的semaphore.release() 。 如果所有的池线程同时调用这个语句,那么maxPoolSize允许被释放,允许相同数目的任务进入单元。 如果我们使用有界的队列,它仍然是满的,导致被拒绝的任务。 现在,因为我们知道这只在池线程几乎完成时才会发生,所以这不是问题。 我们知道池线程不会阻塞,所以很快就会从队列中取出一个任务。

你可以使用有界的队列。 只要确保其大小等于virtualQueueSize + maxPoolSize 。 更大的尺寸是没用的,信号量会阻止让更多的物品进来。尺寸更小会导致被拒绝的任务。 任务被拒绝的机会随着规模的减小而增加。 例如,假设你想要一个maxPoolSize = 2和virtualQueueSize = 5的有界执行器。 然后用5 + 2 = 7个许可证和一个实际队列大小为5 + 2 = 7的信号量。 那么可以在单位中的实际任务数是2 + 5 + 2 = 9。 当执行程序已满时(队列中有5个任务,线程池中有2个任务,因此允许有0个),并且所有池线程都释放它们的许可证,则进入的任务只能执行2个许可证。

现在,来自JCiP的解决scheme使用起来有些麻烦,因为它没有强制执行所有这些约束(无界队列,或者受这些math约束限制等)。 我认为这只是一个很好的例子来演示如何基于已经可用的部分构build新的线程安全类,而不是一个完整的,可重用的类。 我不认为后者是作者的意图。

你可以像这样使用一个自定义的RejectedExecutionHandler

 ThreadPoolExecutor tp= new ThreadPoolExecutor(core_size, // core size max_handlers, // max size timeout_in_seconds, // idle timeout TimeUnit.SECONDS, queue, new RejectedExecutionHandler() { public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // This will block if the queue is full try { executor.getQueue().put(r); } catch (InterruptedException e) { System.err.println(e.getMessage()); } } }); 

创build你自己的阻塞队列,由Executor使用,阻止你正在寻找的行为,同时总是返回可用的剩余容量(确保执行者不会尝试创build比核心池更多的线程,或者触发拒绝处理程序)。

我相信这会让你find你正在寻找的阻塞行为。 拒绝处理程序将永远不符合法案,因为这表明执行者不能执行任务。 我可以想象的是,你在处理程序中得到某种“忙等待”的forms。 这不是你想要的,你想要一个阻止调用者的执行者队列…

你应该看看这个总结了几个解决scheme的链接(通知阻塞线程池) ,最后给出一个优雅的通知。

避免@FixPoint解决scheme的问题。 可以使用ListeningExecutorService并在FutureCallback中释放onSuccess和onFailure信号量。

最近我发现这个问题有同样的问题。 OP没有这么明确地表示,但是我们不想使用在提交者线程上执行一个任务的RejectedExecutionHandler ,因为如果这个任务是长时间运行的,那么这将不利用工作者线程。

阅读所有的答案和评论,特别是信号量的错误解决scheme或使用afterExecute我仔细看看ThreadPoolExecutor的代码,看看是否有某种出路。 我惊奇地发现有2000多行(注释)代码,其中有些代码让我感到头晕目眩 。 鉴于相当简单的要求,我实际上有一个生产者,几个消费者,当没有消费者可以工作时让生产者阻止 – 我决定推出我自己的解决scheme。 这不是一个ExecutorService而只是一个Executor 。 而且它不会将线程的数量调整到工作负载,而是只保留固定数量的线程,这也符合我的要求。 这是代码。 随意咆哮:-)

 package x; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executor; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.SynchronousQueue; /** * distributes {@code Runnable}s to a fixed number of threads. To keep the * code lean, this is not an {@code ExecutorService}. In particular there is * only very simple support to shut this executor down. */ public class ParallelExecutor implements Executor { // other bounded queues work as well and are useful to buffer peak loads private final BlockingQueue<Runnable> workQueue = new SynchronousQueue<Runnable>(); private final Thread[] threads; /*+**********************************************************************/ /** * creates the requested number of threads and starts them to wait for * incoming work */ public ParallelExecutor(int numThreads) { this.threads = new Thread[numThreads]; for(int i=0; i<numThreads; i++) { // could reuse the same Runner all over, but keep it simple Thread t = new Thread(new Runner()); this.threads[i] = t; t.start(); } } /*+**********************************************************************/ /** * returns immediately without waiting for the task to be finished, but may * block if all worker threads are busy. * * @throws RejectedExecutionException if we got interrupted while waiting * for a free worker */ @Override public void execute(Runnable task) { try { workQueue.put(task); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RejectedExecutionException("interrupt while waiting for a free " + "worker.", e); } } /*+**********************************************************************/ /** * Interrupts all workers and joins them. Tasks susceptible to an interrupt * will preempt their work. Blocks until the last thread surrendered. */ public void interruptAndJoinAll() throws InterruptedException { for(Thread t : threads) { t.interrupt(); } for(Thread t : threads) { t.join(); } } /*+**********************************************************************/ private final class Runner implements Runnable { @Override public void run() { while (!Thread.currentThread().isInterrupted()) { Runnable task; try { task = workQueue.take(); } catch (InterruptedException e) { // canonical handling despite exiting right away Thread.currentThread().interrupt(); return; } try { task.run(); } catch (RuntimeException e) { // production code to use a logging framework e.printStackTrace(); } } } } } 

我相信通过使用java.util.concurrent.Semaphore和委托Executor.newFixedThreadPool行为来解决这个问题是非常好的方法。 新的执行者服务只有在有线程的情况下才会执行新的任务。 阻塞由Semaphorepipe理,许可数量等于线程数量。 当任务完成时,它返回一个许可证。

 public class FixedThreadBlockingExecutorService extends AbstractExecutorService { private final ExecutorService executor; private final Semaphore blockExecution; public FixedThreadBlockingExecutorService(int nTreads) { this.executor = Executors.newFixedThreadPool(nTreads); blockExecution = new Semaphore(nTreads); } @Override public void shutdown() { executor.shutdown(); } @Override public List<Runnable> shutdownNow() { return executor.shutdownNow(); } @Override public boolean isShutdown() { return executor.isShutdown(); } @Override public boolean isTerminated() { return executor.isTerminated(); } @Override public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { return executor.awaitTermination(timeout, unit); } @Override public void execute(Runnable command) { blockExecution.acquireUninterruptibly(); executor.execute(() -> { try { command.run(); } finally { blockExecution.release(); } }); } 

过去我也有同样的需求:每个客户端都有一个固定大小的阻塞队列,由共享线程池支持。 我结束了写我自己的一种ThreadPoolExecutor:

UserThreadPoolExecutor(阻塞队列(每个客户端)+线程池(在所有客户端之间共享))

请参阅: https : //github.com/d4rxh4wx/UserThreadPoolExecutor

每个UserThreadPoolExecutor都有一个来自共享ThreadPoolExecutor的最大数量的线程

每个UserThreadPoolExecutor可以:

  • 如果未达到配额,则将任务提交给共享线程池执行程序。 如果达到配额,则作业排队(非消耗性阻塞等待CPU)。 一旦其提交的任务之一完成,配额递减,允许另一个任务等待提交给ThreadPoolExecutor
  • 等待剩下的任务完成

我在弹性search客户端中发现了这个拒绝策略。 它阻止阻塞队列上的调用者线程。 下面的代码 –

  static class ForceQueuePolicy implements XRejectedExecutionHandler { public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { try { executor.getQueue().put(r); } catch (InterruptedException e) { //should never happen since we never wait throw new EsRejectedExecutionException(e); } } @Override public long rejected() { return 0; } } 

我最近有一个需要实现类似的东西,但在ScheduledExecutorService

我还必须确保处理在方法上传递的延迟,并确保当调用者期望的时候提交任务或执行失败,从而抛出RejectedExecutionExceptionexception。

ScheduledThreadPoolExecutor在内部执行或提交一个任务的其他方法调用#schedule ,它仍然会调用覆盖的方法。

 import java.util.concurrent.*; public class BlockingScheduler extends ScheduledThreadPoolExecutor { private final Semaphore maxQueueSize; public BlockingScheduler(int corePoolSize, ThreadFactory threadFactory, int maxQueueSize) { super(corePoolSize, threadFactory, new AbortPolicy()); this.maxQueueSize = new Semaphore(maxQueueSize); } @Override public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) { final long newDelayInMs = beforeSchedule(command, unit.toMillis(delay)); return super.schedule(command, newDelayInMs, TimeUnit.MILLISECONDS); } @Override public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) { final long newDelayInMs = beforeSchedule(callable, unit.toMillis(delay)); return super.schedule(callable, newDelayInMs, TimeUnit.MILLISECONDS); } @Override public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) { final long newDelayInMs = beforeSchedule(command, unit.toMillis(initialDelay)); return super.scheduleAtFixedRate(command, newDelayInMs, unit.toMillis(period), TimeUnit.MILLISECONDS); } @Override public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long period, TimeUnit unit) { final long newDelayInMs = beforeSchedule(command, unit.toMillis(initialDelay)); return super.scheduleWithFixedDelay(command, newDelayInMs, unit.toMillis(period), TimeUnit.MILLISECONDS); } @Override protected void afterExecute(Runnable runnable, Throwable t) { super.afterExecute(runnable, t); try { if (t == null && runnable instanceof Future<?>) { try { ((Future<?>) runnable).get(); } catch (CancellationException | ExecutionException e) { t = e; } catch (InterruptedException ie) { Thread.currentThread().interrupt(); // ignore/reset } } if (t != null) { System.err.println(t); } } finally { releaseQueueUsage(); } } private long beforeSchedule(Runnable runnable, long delay) { try { return getQueuePermitAndModifiedDelay(delay); } catch (InterruptedException e) { getRejectedExecutionHandler().rejectedExecution(runnable, this); return 0; } } private long beforeSchedule(Callable callable, long delay) { try { return getQueuePermitAndModifiedDelay(delay); } catch (InterruptedException e) { getRejectedExecutionHandler().rejectedExecution(new FutureTask(callable), this); return 0; } } private long getQueuePermitAndModifiedDelay(long delay) throws InterruptedException { final long beforeAcquireTimeStamp = System.currentTimeMillis(); maxQueueSize.tryAcquire(delay, TimeUnit.MILLISECONDS); final long afterAcquireTimeStamp = System.currentTimeMillis(); return afterAcquireTimeStamp - beforeAcquireTimeStamp; } private void releaseQueueUsage() { maxQueueSize.release(); } } 

我有这里的代码,将不胜感激任何反馈。 https://github.com/AmitabhAwasthi/BlockingScheduler

这里的解决scheme似乎工作得很好。 它被称为NotifyingBlockingThreadPoolExecutor 。

演示程序。

编辑:有这个代码的问题 ,await()方法是越野车。 调用shutdown()+ awaitTermination()似乎工作正常。

我并不总是喜欢CallerRunsPolicy,尤其是因为它允许被拒绝的任务“跳过队列”并在之前提交的任务之前执行。 而且,在调用线程上执行任务可能需要比等待第一个时隙更长的时间。

我使用自定义的RejectedExecutionHandler解决了这个问题,它只是简单地阻塞调用线程一段时间,然后尝试再次提交任务:

 public class BlockWhenQueueFull implements RejectedExecutionHandler { public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // The pool is full. Wait, then try again. try { long waitMs = 250; Thread.sleep(waitMs); } catch (InterruptedException interruptedException) {} executor.execute(r); } } 

这个类只能在线程池执行器中像任何其他一样被用作RejectedExecutinHandler,例如:

 executorPool = new ThreadPoolExecutor(1, 1, 10, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new BlockWhenQueueFull()); 

我看到唯一的缺点是调用线程可能会被locking的时间稍长于绝对必要的时间(高达250ms)。 而且,由于这个执行器被有效地recursion调用,很长时间等待一个线程变为可用(小时)可能导致堆栈溢出。

不过,我个人喜欢这个方法。 它结构紧凑,易于理解,并且运行良好。