entityMap|blocks|key|610k3|text|我认为这很简单，只需使用ArrayBlockingQueue而不是aLinkedBlockingQueue..。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|7o8a6|无视我..。这是完全错误的。ThreadPoolExecutor呼叫Queue#offer注释put会有你想要的效果。你可以扩展ThreadPoolExecutor|8a6o|并提供了execute(Runnable)它调用put代替offer..。恐怕这不是一个完全令人满意的答案。^0|C|I|Y|J|0|E|I|Y|B|1B|3|1S|I|0|4|H|O|3|T|5^^$0|$]|1|@$2|3|4|5|6|7|8|K|9|@$A|L|B|M|C|D]|$A|N|B|O|C|D]]|E|@]|F|$]]|$2|G|4|H|6|7|8|P|9|@$A|Q|B|R|C|D]|$A|S|B|T|C|D]|$A|U|B|V|C|D]|$A|W|B|X|C|D]]|E|@]|F|$]]|$2|I|4|J|6|7|8|Y|9|@$A|Z|B|10|C|D]|$A|11|B|12|C|D]|$A|13|B|14|C|D]]|E|@]|F|$]]]]

I think it is as simple as using a <code>ArrayBlockingQueue</code> instead of a a <code>LinkedBlockingQueue</code>.

Ignore me... that's totally wrong. <code>ThreadPoolExecutor</code> calls <code>Queue#offer</code> not <code>put</code> which would have the effect you require.

You could extend <code>ThreadPoolExecutor</code> and provide an implementation of <code>execute(Runnable)</code> that calls <code>put</code> in place of <code>offer</code>. 

That doesn't seem like a completely satisfactory answer I'm afraid.

entityMap|blocks|key|1fbpl|text|我也做过同样的事情。诀窍是创建一个BlockingQueue，其中的offer()方法实际上是一个put()。(您可以使用所需的任何基本BlockingQueue实现)。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|doqcf|public+class+LimitedQueue+extends+LinkedBlockingQueue+
{
++++public+LimitedQueue(int+maxSize)
++++{
++++++++super(maxSize);
++++}

++++@Override
++++public+boolean+offer(E+e)
++++{
++++++++//+turn+offer()+and+add()+into+a+blocking+calls+(unless+interrupted)
++++++++try+{
++++++++++++put(e);
++++++++++++return+true;
++++++++}+catch(InterruptedException+ie)+{
++++++++++++Thread.currentThread().interrupt();
++++++++}
++++++++return+false;
++++}

}|code-block|syntax|javascript|60oe5|请注意，这只适用于线程池，其中corePoolSize==maxPoolSize所以在这里要小心(见注释)。|offset|length|style|CODE^0|0|0|F|P^^$0|$]|1|@$2|3|4|5|6|7|8|N|9|@]|A|@]|B|$]]|$2|C|4|D|6|E|8|O|9|@]|A|@]|B|$F|G]]|$2|H|4|I|6|7|8|P|9|@$J|Q|K|R|L|M]]|A|@]|B|$]]]]

I have done this same thing. The trick is to create a BlockingQueue where the offer() method is really a put(). (you can use whatever base BlockingQueue impl you want).

<pre><code>public class LimitedQueue&lt;E&gt; extends LinkedBlockingQueue&lt;E&gt; 
{
 public LimitedQueue(int maxSize)
 {
 super(maxSize);
 }

 @Override
 public boolean offer(E e)
 {
 // turn offer() and add() into a blocking calls (unless interrupted)
 try {
 put(e);
 return true;
 } catch(InterruptedException ie) {
 Thread.currentThread().interrupt();
 }
 return false;
 }

}
</code></pre>

Note that this only works for thread pool where <code>corePoolSize==maxPoolSize</code> so be careful there (see comments).

entityMap|blocks|key|4v3u1|text|我遇到了类似的问题，我通过使用beforeExecute/afterExecute钩子来自ThreadPoolExecutor|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|ck1mm|import+java.util.concurrent.BlockingQueue;
import+java.util.concurrent.ThreadPoolExecutor;
import+java.util.concurrent.TimeUnit;
import+java.util.concurrent.locks.Condition;
import+java.util.concurrent.locks.ReentrantLock;

/**
+*+Blocks+current+task+execution+if+there+is+not+enough+resources+for+it.
+*+Maximum+task+count+usage+controlled+by+maxTaskCount+property.
+*/
public+class+BlockingThreadPoolExecutor+extends+ThreadPoolExecutor+{

++++private+final+ReentrantLock+taskLock+=+new+ReentrantLock();
++++private+final+Condition+unpaused+=+taskLock.newCondition();
++++private+final+int+maxTaskCount;

++++private+volatile+int+currentTaskCount;

++++public+BlockingThreadPoolExecutor(int+corePoolSize,+int+maximumPoolSize,
++++++++++++long+keepAliveTime,+TimeUnit+unit,
++++++++++++BlockingQueue+workQueue,+int+maxTaskCount)+{
++++++++super(corePoolSize,+maximumPoolSize,+keepAliveTime,+unit,+workQueue);
++++++++this.maxTaskCount+=+maxTaskCount;
++++}

++++/**
+++++*+Executes+task+if+there+is+enough+system+resources+for+it.+Otherwise
+++++*+waits.
+++++*/
++++@Override
++++protected+void+beforeExecute(Thread+t,+Runnable+r)+{
++++++++super.beforeExecute(t,+r);
++++++++taskLock.lock();
++++++++try+{
++++++++++++//+Spin+while+we+will+not+have+enough+capacity+for+this+job
++++++++++++while+(maxTaskCount+<+currentTaskCount)+{
++++++++++++++++try+{
++++++++++++++++++++unpaused.await();
++++++++++++++++}+catch+(InterruptedException+e)+{
++++++++++++++++++++t.interrupt();
++++++++++++++++}
++++++++++++}
++++++++++++currentTaskCount%2B%2B;
++++++++}+finally+{
++++++++++++taskLock.unlock();
++++++++}
++++}

++++/**
+++++*+Signalling+that+one+more+task+is+welcome
+++++*/
++++@Override
++++protected+void+afterExecute(Runnable+r,+Throwable+t)+{
++++++++super.afterExecute(r,+t);
++++++++taskLock.lock();
++++++++try+{
++++++++++++currentTaskCount--;
++++++++++++unpaused.signalAll();
++++++++}+finally+{
++++++++++++taskLock.unlock();
++++++++}
++++}
}|code-block|syntax|javascript|cd59q|这对你来说应该足够好了。顺便说一句，最初的实现是基于任务大小的，因为一个任务可能比另一个任务大100倍，提交两个巨大的任务是致命的，但运行一个大的和足够小的任务是可以的。如果您的I/O密集型任务的大小大致相同，您可以使用这个类，否则请让我知道，我将发布基于大小的实现。|79g42|附注:你会想要检查ThreadPoolExecutorjavadoc。这是来自Doug+Lea的非常好的用户指南，它可以很容易地进行定制。^0|F|Q|19|I|0|0|0|9|I^^$0|$]|1|@$2|3|4|5|6|7|8|P|9|@$A|Q|B|R|C|D]|$A|S|B|T|C|D]]|E|@]|F|$]]|$2|G|4|H|6|I|8|U|9|@]|E|@]|F|$J|K]]|$2|L|4|M|6|7|8|V|9|@]|E|@]|F|$]]|$2|N|4|O|6|7|8|W|9|@$A|X|B|Y|C|D]]|E|@]|F|$]]]]

I had the similar problem and I implemented that by using <code>beforeExecute/afterExecute</code> hooks from <code>ThreadPoolExecutor</code>:

<pre><code>import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Blocks current task execution if there is not enough resources for it.
 * Maximum task count usage controlled by maxTaskCount property.
 */
public class BlockingThreadPoolExecutor extends ThreadPoolExecutor {

 private final ReentrantLock taskLock = new ReentrantLock();
 private final Condition unpaused = taskLock.newCondition();
 private final int maxTaskCount;

 private volatile int currentTaskCount;

 public BlockingThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
 long keepAliveTime, TimeUnit unit,
 BlockingQueue&lt;Runnable&gt; workQueue, int maxTaskCount) {
 super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
 this.maxTaskCount = maxTaskCount;
 }

 /**
 * Executes task if there is enough system resources for it. Otherwise
 * waits.
 */
 @Override
 protected void beforeExecute(Thread t, Runnable r) {
 super.beforeExecute(t, r);
 taskLock.lock();
 try {
 // Spin while we will not have enough capacity for this job
 while (maxTaskCount &lt; currentTaskCount) {
 try {
 unpaused.await();
 } catch (InterruptedException e) {
 t.interrupt();
 }
 }
 currentTaskCount++;
 } finally {
 taskLock.unlock();
 }
 }

 /**
 * Signalling that one more task is welcome
 */
 @Override
 protected void afterExecute(Runnable r, Throwable t) {
 super.afterExecute(r, t);
 taskLock.lock();
 try {
 currentTaskCount--;
 unpaused.signalAll();
 } finally {
 taskLock.unlock();
 }
 }
}
</code></pre>

This should be good enough for you. Btw, original implementation was task size based because one task could be larger 100 time than another and submitting two huge tasks was killing the box, but running one big and plenty of small was Okay. If your I/O-intensive tasks are roughly the same size you could use this class, otherwise just let me know and I'll post size based implementation.

P.S. You would want to check <code>ThreadPoolExecutor</code> javadoc. It's really nice user guide from Doug Lea about how it could be easily customized.

entityMap|blocks|key|a7obs|text|我知道这是一个老问题，但也有一个类似的问题，那就是创建新任务的速度非常快，如果因为现有任务没有足够快地完成而发生了太多的OutOfMemoryError。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|42u4t|在我的情况下Callables，并且我需要结果，因此我需要存储所有Futures返回者executor.submit()..。我的解决方案是将Futures转换成一个BlockingQueue最大尺寸。一旦队列满了，就不会生成更多的任务，直到一些任务完成(从队列中删除元素)。在伪代码中：|offset|length|style|CODE|29qv3|final+ExecutorService+executor+=+Executors.newFixedThreadPool(numWorkerThreads);
final+LinkedBlockingQueue+futures+=+new+LinkedBlockingQueue<>(maxQueueSize);
try+{+++
++++Thread+taskGenerator+=+new+Thread()+{
++++++++@Override
++++++++public+void+run()+{
++++++++++++while+(reader.hasNext)+{
++++++++++++++++Callable+task+=+generateTask(reader.next());
++++++++++++++++Future+future+=+executor.submit(task);
++++++++++++++++try+{
++++++++++++++++++++//+if+queue+is+full+blocks+until+a+task
++++++++++++++++++++//+is+completed+and+hence+no+future+tasks+are+submitted.
++++++++++++++++++++futures.put(future);
++++++++++++++++}+catch+(InterruptedException+ex)+{
++++++++++++++++++++Thread.currentThread().interrupt();+++++++++
++++++++++++++++}
++++++++++++}
++++++++executor.shutdown();
++++++++}
++++}
++++taskGenerator.start();
++++
++++//+read+from+queue+as+long+as+task+are+being+generated
++++//+or+while+Queue+has+elements+in+it
++++while+(taskGenerator.isAlive()
++++++++++++++++++++%7C%7C+!futures.isEmpty())+{
++++++++Future+future+=+futures.take();
++++++++//+do+something
++++}
}+catch+(InterruptedException+ex)+{
++++Thread.currentThread().interrupt();+++++
}+catch+(ExecutionException+ex)+{
++++throw+new+MyException(ex);
}+finally+{
++++executor.shutdownNow();
}|code-block|syntax|javascript^0|0|6|9|X|7|17|H|1Z|7|2B|D|0^^$0|$]|1|@$2|3|4|5|6|7|8|N|9|@]|A|@]|B|$]]|$2|C|4|D|6|7|8|O|9|@$E|P|F|Q|G|H]|$E|R|F|S|G|H]|$E|T|F|U|G|H]|$E|V|F|W|G|H]|$E|X|F|Y|G|H]]|A|@]|B|$]]|$2|I|4|J|6|K|8|Z|9|@]|A|@]|B|$L|M]]]]

I know this is an old question but had a similar issue that creating new tasks was very fast and if there were too many an OutOfMemoryError occur because existing task were not completed fast enough.
In my case <code>Callables</code> are submitted and I need the result hence I need to store all the <code>Futures</code> returned by <code>executor.submit()</code>. My solution was to put the <code>Futures</code> into a <code>BlockingQueue</code> with a maximum size. Once that queue is full, no more tasks are generated until some are completed (elements removed from queue). In pseudo-code:
<pre><code>final ExecutorService executor = Executors.newFixedThreadPool(numWorkerThreads);
final LinkedBlockingQueue&lt;Future&gt; futures = new LinkedBlockingQueue&lt;&gt;(maxQueueSize);
try { 
 Thread taskGenerator = new Thread() {
 @Override
 public void run() {
 while (reader.hasNext) {
 Callable task = generateTask(reader.next());
 Future future = executor.submit(task);
 try {
 // if queue is full blocks until a task
 // is completed and hence no future tasks are submitted.
 futures.put(future);
 } catch (InterruptedException ex) {
 Thread.currentThread().interrupt(); 
 }
 }
 executor.shutdown();
 }
 }
 taskGenerator.start();
 
 // read from queue as long as task are being generated
 // or while Queue has elements in it
 while (taskGenerator.isAlive()
 || !futures.isEmpty()) {
 Future future = futures.take();
 // do something
 }
} catch (InterruptedException ex) {
 Thread.currentThread().interrupt(); 
} catch (ExecutionException ex) {
 throw new MyException(ex);
} finally {
 executor.shutdownNow();
}
</code></pre>

entityMap|blocks|key|2o40r|text|下面是我是如何解决这个问题的：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|7a3m|(注意:此解决方案确实会阻塞提交可调用对象的线程，因此它可以防止抛出RejectedExecutionException+)|3smmj|public+class+BoundedExecutor+extends+ThreadPoolExecutor{

++++private+final+Semaphore+semaphore;

++++public+BoundedExecutor(int+bound)+{
++++++++super(bound,+Integer.MAX_VALUE,+60L,+TimeUnit.SECONDS,+new+SynchronousQueue());
++++++++semaphore+=+new+Semaphore(bound);
++++}

++++/**Submits+task+to+execution+pool,+but+blocks+while+number+of+running+threads+
+++++*+has+reached+the+bound+limit
+++++*/
++++public++Future+submitButBlockIfFull(final+Callable+task)+throws+InterruptedException{

++++++++semaphore.acquire();++++++++++++
++++++++return+submit(task);++++++++++++++++++++
++++}


++++@Override
++++protected+void+afterExecute(Runnable+r,+Throwable+t)+{
++++++++super.afterExecute(r,+t);

++++++++semaphore.release();
++++}
}|code-block|syntax|javascript^0|0|0^^$0|$]|1|@$2|3|4|5|6|7|8|J|9|@]|A|@]|B|$]]|$2|C|4|D|6|7|8|K|9|@]|A|@]|B|$]]|$2|E|4|F|6|G|8|L|9|@]|A|@]|B|$H|I]]]]

Here is how I solved this on my end:

(note: this solution does block the thread that submits the Callable, so it prevents RejectedExecutionException from being thrown )

<pre><code>public class BoundedExecutor extends ThreadPoolExecutor{

 private final Semaphore semaphore;

 public BoundedExecutor(int bound) {
 super(bound, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue&lt;Runnable&gt;());
 semaphore = new Semaphore(bound);
 }

 /**Submits task to execution pool, but blocks while number of running threads 
 * has reached the bound limit
 */
 public &lt;T&gt; Future&lt;T&gt; submitButBlockIfFull(final Callable&lt;T&gt; task) throws InterruptedException{

 semaphore.acquire(); 
 return submit(task); 
 }


 @Override
 protected void afterExecute(Runnable r, Throwable t) {
 super.afterExecute(r, t);

 semaphore.release();
 }
}
</code></pre>

entityMap|0|type|LINK|mutability|MUTABLE|data|url|http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html|1|http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/RejectedExecutionHandler.html|2|https://stackoverflow.com/a/3518588/3080094|blocks|key|6b6c0|text|当前接受的答案有一个潜在的重大问题-它改变了ThreadPoolExecutor.execute的行为，这样如果您有一个|unstyled|depth|inlineStyleRanges|entityRanges|enkk4|corePoolSize+<+maxPoolSize，ThreadPoolExecutor逻辑永远不会在核心之外添加额外的工作进程。|offset|length|style|CODE|atgha|来自ThreadPoolExecutor.execute(可运行)：|amqbl|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);|code-block|syntax|javascript|o7ol|具体地说，最后一个'else‘块永远不会被命中。|dc8i1|一个更好的替代方法是做一些类似于OP已经在做的事情--使用RejectedExecutionHandler做同样的事情put逻辑：|4bbcp|public+void+rejectedExecution(Runnable+r,+ThreadPoolExecutor+executor)+{
++++try+{
++++++++if+(!executor.isShutdown())+{
++++++++++++executor.getQueue().put(r);
++++++++}
++++}+catch+(InterruptedException+e)+{
++++++++Thread.currentThread().interrupt();
++++++++throw+new+RejectedExecutionException("Executor+was+interrupted+while+the+task+was+waiting+to+put+on+work+queue",+e);
++++}
}|fho3f|正如评论中指出的那样，使用这种方法需要注意一些事情(请参阅此answer)：|5opsc|如果corePoolSize==0，则会出现争用情况，此时池中的所有线程都可能在任务可见之前终止。|ordered-list-item|6uk6u|使用包装队列任务的实现(不适用于ThreadPoolExecutor)将导致问题，除非处理程序也以相同的方式包装它。|4e6ou|记住这些问题，这个解决方案将适用于大多数典型的ThreadPoolExecutors，并且可以正确地处理以下情况corePoolSize+<+maxPoolSize..。^0|0|0|Q|0|2|I|0|0|0|0|1N|3|T|O|1|0|0|T|7|2|0|2|F|0|G|I|0|1K|Q^^$0|$1|$2|3|4|5|6|$7|8]]|9|$2|3|4|5|6|$7|A]]|B|$2|3|4|5|6|$7|C]]]|D|@$E|F|G|H|2|I|J|1E|K|@]|L|@]|6|$]]|$E|M|G|N|2|I|J|1F|K|@$O|1G|P|1H|Q|R]]|L|@]|6|$]]|$E|S|G|T|2|I|J|1I|K|@]|L|@$O|1J|P|1K|E|1L]]|6|$]]|$E|U|G|V|2|W|J|1M|K|@]|L|@]|6|$X|Y]]|$E|Z|G|10|2|I|J|1N|K|@]|L|@]|6|$]]|$E|11|G|12|2|I|J|1O|K|@$O|1P|P|1Q|Q|R]]|L|@$O|1R|P|1S|E|1T]]|6|$]]|$E|13|G|14|2|W|J|1U|K|@]|L|@]|6|$X|Y]]|$E|15|G|16|2|I|J|1V|K|@]|L|@$O|1W|P|1X|E|1Y]]|6|$]]|$E|17|G|18|2|19|J|1Z|K|@$O|20|P|21|Q|R]]|L|@]|6|$]]|$E|1A|G|1B|2|19|J|22|K|@$O|23|P|24|Q|R]]|L|@]|6|$]]|$E|1C|G|1D|2|I|J|25|K|@$O|26|P|27|Q|R]]|L|@]|6|$]]]]

The currently accepted answer has a potentially significant problem - it changes the behavior of ThreadPoolExecutor.execute such that if you have a <code>corePoolSize &lt; maxPoolSize</code>, the ThreadPoolExecutor logic will never add additional workers beyond the core. 

From <a href="http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html" rel="noreferrer">ThreadPoolExecutor</a>.execute(Runnable):

<pre><code> if (isRunning(c) &amp;&amp; workQueue.offer(command)) {
 int recheck = ctl.get();
 if (! isRunning(recheck) &amp;&amp; remove(command))
 reject(command);
 else if (workerCountOf(recheck) == 0)
 addWorker(null, false);
 }
 else if (!addWorker(command, false))
 reject(command);
</code></pre>

Specifically, that last 'else' block willl never be hit.

A better alternative is to do something similar to what OP is already doing - use a <a href="http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/RejectedExecutionHandler.html" rel="noreferrer">RejectedExecutionHandler</a> to do the same <code>put</code> logic:

<pre><code>public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
 try {
 if (!executor.isShutdown()) {
 executor.getQueue().put(r);
 }
 } catch (InterruptedException e) {
 Thread.currentThread().interrupt();
 throw new RejectedExecutionException("Executor was interrupted while the task was waiting to put on work queue", e);
 }
}
</code></pre>

There are some things to watch out for with this approach, as pointed out in the comments (referring to <a href="https://stackoverflow.com/a/3518588/3080094">this answer</a>):

<ol>
<li>If <code>corePoolSize==0</code>, then there is a race condition where all threads in the pool may die before the task is visible</li>
<li>Using an implementation that wraps the queue tasks (not applicable to <code>ThreadPoolExecutor</code>) will result in issues unless the handler also wraps it the same way.</li>
</ol>

Keeping those gotchas in mind, this solution will work for most typical ThreadPoolExecutors, and will properly handle the case where <code>corePoolSize &lt; maxPoolSize</code>.

entityMap|blocks|key|82uia|text|我已经实现了一个遵循装饰器模式的解决方案，并使用信号量来控制执行任务的数量。您可以将它与任何Executor和：|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|9mq71|指定正在进行的任务的最大值|unordered-list-item|fa17q|指定等待任务执行许可的最大超时时间(如果超时时间已过但未获取许可，则RejectedExecutionException被抛出)|75gqp|import+static+java.util.concurrent.TimeUnit.MILLISECONDS;

import+java.time.Duration;
import+java.util.Objects;
import+java.util.concurrent.Executor;
import+java.util.concurrent.RejectedExecutionException;
import+java.util.concurrent.Semaphore;

import+javax.annotation.Nonnull;

public+class+BlockingOnFullQueueExecutorDecorator+implements+Executor+{

++++private+static+final+class+PermitReleasingDecorator+implements+Runnable+{

++++++++@Nonnull
++++++++private+final+Runnable+delegate;

++++++++@Nonnull
++++++++private+final+Semaphore+semaphore;

++++++++private+PermitReleasingDecorator(@Nonnull+final+Runnable+task,+@Nonnull+final+Semaphore+semaphoreToRelease)+{
++++++++++++this.delegate+=+task;
++++++++++++this.semaphore+=+semaphoreToRelease;
++++++++}

++++++++@Override
++++++++public+void+run()+{
++++++++++++try+{
++++++++++++++++this.delegate.run();
++++++++++++}
++++++++++++finally+{
++++++++++++++++//+however+execution+goes,+release+permit+for+next+task
++++++++++++++++this.semaphore.release();
++++++++++++}
++++++++}

++++++++@Override
++++++++public+final+String+toString()+{
++++++++++++return+String.format("%25s[delegate='%25s']",+getClass().getSimpleName(),+this.delegate);
++++++++}
++++}

++++@Nonnull
++++private+final+Semaphore+taskLimit;

++++@Nonnull
++++private+final+Duration+timeout;

++++@Nonnull
++++private+final+Executor+delegate;

++++public+BlockingOnFullQueueExecutorDecorator(@Nonnull+final+Executor+executor,+final+int+maximumTaskNumber,+@Nonnull+final+Duration+maximumTimeout)+{
++++++++this.delegate+=+Objects.requireNonNull(executor,+"'executor'+must+not+be+null");
++++++++if+(maximumTaskNumber+<+1)+{
++++++++++++throw+new+IllegalArgumentException(String.format("At+least+one+task+must+be+permitted,+not+'%25d'",+maximumTaskNumber));
++++++++}
++++++++this.timeout+=+Objects.requireNonNull(maximumTimeout,+"'maximumTimeout'+must+not+be+null");
++++++++if+(this.timeout.isNegative())+{
++++++++++++throw+new+IllegalArgumentException("'maximumTimeout'+must+not+be+negative");
++++++++}
++++++++this.taskLimit+=+new+Semaphore(maximumTaskNumber);
++++}

++++@Override
++++public+final+void+execute(final+Runnable+command)+{
++++++++Objects.requireNonNull(command,+"'command'+must+not+be+null");
++++++++try+{
++++++++++++//+attempt+to+acquire+permit+for+task+execution
++++++++++++if+(!this.taskLimit.tryAcquire(this.timeout.toMillis(),+MILLISECONDS))+{
++++++++++++++++throw+new+RejectedExecutionException(String.format("Executor+'%25s'+busy",+this.delegate));
++++++++++++}
++++++++}
++++++++catch+(final+InterruptedException+e)+{
++++++++++++//+restore+interrupt+status
++++++++++++Thread.currentThread().interrupt();
++++++++++++throw+new+IllegalStateException(e);
++++++++}

++++++++this.delegate.execute(new+PermitReleasingDecorator(command,+this.taskLimit));
++++}

++++@Override
++++public+final+String+toString()+{
++++++++return+String.format("%25s[availablePermits='%25s',timeout='%25s',delegate='%25s']",+getClass().getSimpleName(),+this.taskLimit.availablePermits(),
++++++++++++++++this.timeout,+this.delegate);
++++}
}|code-block|syntax|javascript^0|1A|8|0|0|Y|Q|0^^$0|$]|1|@$2|3|4|5|6|7|8|Q|9|@$A|R|B|S|C|D]]|E|@]|F|$]]|$2|G|4|H|6|I|8|T|9|@]|E|@]|F|$]]|$2|J|4|K|6|I|8|U|9|@$A|V|B|W|C|D]]|E|@]|F|$]]|$2|L|4|M|6|N|8|X|9|@]|E|@]|F|$O|P]]]]

I have implemented a solution following the decorator pattern and using a semaphore to control the number of executed tasks. You can use it with any <code>Executor</code> and:

<ul>
<li>Specify the maximum of ongoing tasks</li>
<li>Specify the maximum timeout to wait for a task execution permit (if the timeout passes and no permit is acquired, a <code>RejectedExecutionException</code> is thrown)</li>
</ul>

<pre class="lang-java prettyprint-override"><code>import static java.util.concurrent.TimeUnit.MILLISECONDS;

import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;

import javax.annotation.Nonnull;

public class BlockingOnFullQueueExecutorDecorator implements Executor {

 private static final class PermitReleasingDecorator implements Runnable {

 @Nonnull
 private final Runnable delegate;

 @Nonnull
 private final Semaphore semaphore;

 private PermitReleasingDecorator(@Nonnull final Runnable task, @Nonnull final Semaphore semaphoreToRelease) {
 this.delegate = task;
 this.semaphore = semaphoreToRelease;
 }

 @Override
 public void run() {
 try {
 this.delegate.run();
 }
 finally {
 // however execution goes, release permit for next task
 this.semaphore.release();
 }
 }

 @Override
 public final String toString() {
 return String.format("%s[delegate='%s']", getClass().getSimpleName(), this.delegate);
 }
 }

 @Nonnull
 private final Semaphore taskLimit;

 @Nonnull
 private final Duration timeout;

 @Nonnull
 private final Executor delegate;

 public BlockingOnFullQueueExecutorDecorator(@Nonnull final Executor executor, final int maximumTaskNumber, @Nonnull final Duration maximumTimeout) {
 this.delegate = Objects.requireNonNull(executor, "'executor' must not be null");
 if (maximumTaskNumber &lt; 1) {
 throw new IllegalArgumentException(String.format("At least one task must be permitted, not '%d'", maximumTaskNumber));
 }
 this.timeout = Objects.requireNonNull(maximumTimeout, "'maximumTimeout' must not be null");
 if (this.timeout.isNegative()) {
 throw new IllegalArgumentException("'maximumTimeout' must not be negative");
 }
 this.taskLimit = new Semaphore(maximumTaskNumber);
 }

 @Override
 public final void execute(final Runnable command) {
 Objects.requireNonNull(command, "'command' must not be null");
 try {
 // attempt to acquire permit for task execution
 if (!this.taskLimit.tryAcquire(this.timeout.toMillis(), MILLISECONDS)) {
 throw new RejectedExecutionException(String.format("Executor '%s' busy", this.delegate));
 }
 }
 catch (final InterruptedException e) {
 // restore interrupt status
 Thread.currentThread().interrupt();
 throw new IllegalStateException(e);
 }

 this.delegate.execute(new PermitReleasingDecorator(command, this.taskLimit));
 }

 @Override
 public final String toString() {
 return String.format("%s[availablePermits='%s',timeout='%s',delegate='%s']", getClass().getSimpleName(), this.taskLimit.availablePermits(),
 this.timeout, this.delegate);
 }
}
</code></pre>

entityMap|0|type|LINK|mutability|MUTABLE|data|url|https://github.com/spring-projects/spring-integration/blob/master/spring-integration-core/src/main/java/org/springframework/integration/util/CallerBlocksPolicy.java|blocks|key|9jmga|text|使用CallerBlocksPolicy如果你使用的是spring集成的话？这个类实现了|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|b1ddc|RejectedExecutionHandler接口，该接口是一个处理程序，用于处理不能由ThreadPoolExecutor..。您可以像这样使用此策略。|ao45m|executor.setRejectedExecutionHandler(new+CallerBlocksPolicy());|code-block|syntax|js|bf9kg|它们之间的主要区别是CallerBlocksPolicy和CallerRunsPolicy是阻塞还是运行调用者线程中的任务。|eoejt|请参阅此code..。^0|2|I|0|0|O|19|I|0|0|A|I|T|G|0|3|5|0^^$0|$1|$2|3|4|5|6|$7|8]]]|9|@$A|B|C|D|2|E|F|X|G|@$H|Y|I|Z|J|K]]|L|@]|6|$]]|$A|M|C|N|2|E|F|10|G|@$H|11|I|12|J|K]|$H|13|I|14|J|K]]|L|@]|6|$]]|$A|O|C|P|2|Q|F|15|G|@]|L|@]|6|$R|S]]|$A|T|C|U|2|E|F|16|G|@$H|17|I|18|J|K]|$H|19|I|1A|J|K]]|L|@]|6|$]]|$A|V|C|W|2|E|F|1B|G|@]|L|@$H|1C|I|1D|A|1E]]|6|$]]]]

How about using the <code>CallerBlocksPolicy</code> class if you are using spring-integration?
This class implements the <code>RejectedExecutionHandler</code> interface, which is a handler for tasks that cannot be executed by a <code>ThreadPoolExecutor</code>.
You can use this policy like this.
<pre><code>executor.setRejectedExecutionHandler(new CallerBlocksPolicy());
</code></pre>
The main difference between <code>CallerBlocksPolicy</code> and <code>CallerRunsPolicy</code> is whether it blocks or runs the task in the caller thread.
Please refer to <a href="https://github.com/spring-projects/spring-integration/blob/master/spring-integration-core/src/main/java/org/springframework/integration/util/CallerBlocksPolicy.java" rel="noreferrer">this code</a>.

I am trying to code a solution in which a single thread produces I/O-intensive tasks that can be performed in parallel. Each task have significant in-memory data. So I want to be able limit the number of tasks that are pending at a moment. 

If I create ThreadPoolExecutor like this:

<pre><code> ThreadPoolExecutor executor = new ThreadPoolExecutor(numWorkerThreads, numWorkerThreads,
 0L, TimeUnit.MILLISECONDS,
 new LinkedBlockingQueue&lt;Runnable&gt;(maxQueue));
</code></pre>

Then the <code>executor.submit(callable)</code> throws <code>RejectedExecutionException</code> when the queue fills up and all the threads are already busy. 

What can I do to make <code>executor.submit(callable)</code> block when the queue is full and all threads are busy?

EDIT:
I tried <a href="https://stackoverflow.com/a/5656329/32453">this</a>:

<pre><code>executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
</code></pre>

And it somewhat achieves the effect that I want achieved but in an inelegant way (basically rejected threads are run in the calling thread, so this blocks the calling thread from submitting more).

EDIT: (5 years after asking the question)

To anyone reading this question and its answers, please don't take the accepted answer as one correct solution. Please read through all answers and comments.

Java: ExecutorService that blocks on submission after a certain queue size

 我正在尝试编写一个解决方案，其中单个线程产生可以并行执行的I/O密集型任务。每个任务都有大量的内存数据。因此，我希望能够限制当前挂起的任务的数量。 如果我像这样创建ThreadPoolExecutor： ThreadPoolExecutor executor = new ThreadPoolExecutor(numW...

问Java:在一定队列大小后阻塞提交的ExecutorService
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问Java:在一定队列大小后阻塞提交的ExecutorServiceEN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问Java:在一定队列大小后阻塞提交的ExecutorService
EN