blocks|key|319502|text|您应该使用CallerRunsPolicy，它在调用线程中执行被拒绝的任务。这样，在任务完成之前，它不能向executor提交任何新任务，此时将有一些空闲的池线程，否则进程将重复。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|319503|http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/ThreadPoolExecutor.CallerRunsPolicy.html|319504|从文档中：|319505|319506|拒绝的任务|blockquote|319507|319508|在方法执行(java.lang.Runnable)中提交的新任务将在Executor关闭时被拒绝，当Executor对最大线程和工作队列容量使用有限的界限时也是如此，并且是饱和的。在这两种情况下，execute方法都会调用其RejectedExecutionHandler的RejectedExecutionHandler.rejectedExecution(java.lang.Runnable，java.util.concurrent.ThreadPoolExecutor方法。提供了四个预定义的处理程序策略：|319509|319510|319511|在默认ThreadPoolExecutor.AbortPolicy中，处理程序在被拒绝时抛出运行时RejectedExecutionException。|ordered-list-item|319512|在ThreadPoolExecutor.CallerRunsPolicy中，调用execute本身的线程运行任务。这提供了一种简单的反馈控制机制，可以减慢提交新任务的速度。|319513|在ThreadPoolExecutor.DiscardPolicy中，无法执行的任务将被简单地删除。|319514|在ThreadPoolExecutor.DiscardOldestPolicy中，如果未关闭执行器，则会删除位于工作队列头的任务，然后重试执行(这可能会再次失败，导致重复执行)|319515|。|319516|此外，在调用ThreadPoolExecutor构造函数时，请确保使用有界队列，如ArrayBlockingQueue。否则，什么都不会被拒绝。|319517|编辑:响应您的评论，将ArrayBlockingQueue的大小设置为等于线程池的最大大小，并使用AbortPolicy。|319518|编辑2:好的，我明白你的意思了。这样如何:重写beforeExecute()方法以检查getActiveCount()是否超过getMaximumPoolSize()，如果超过了，则休眠并重试？|319519|entityMap|0|LINK|mutability|MUTABLE|url^0|5|G|0|0|2T|0|0|0|0|0|0|0|0|0|0|0|0|0|0|6|I|0|0|N|F|17|G|1R|K|0^^$0|@$1|2|3|4|5|6|7|1G|8|@$9|1H|A|1I|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|1J|8|@]|D|@$9|1K|A|1L|1|1M]]|E|$]]|$1|H|3|I|5|6|7|1N|8|@]|D|@]|E|$]]|$1|J|3|-4|5|6|7|1O|8|@]|D|@]|E|$]]|$1|K|3|L|5|M|7|1P|8|@]|D|@]|E|$]]|$1|N|3|-4|5|6|7|1Q|8|@]|D|@]|E|$]]|$1|O|3|P|5|6|7|1R|8|@]|D|@]|E|$]]|$1|Q|3|-4|5|6|7|1S|8|@]|D|@]|E|$]]|$1|R|3|-4|5|6|7|1T|8|@]|D|@]|E|$]]|$1|S|3|T|5|U|7|1U|8|@]|D|@]|E|$]]|$1|V|3|W|5|U|7|1V|8|@]|D|@]|E|$]]|$1|X|3|Y|5|U|7|1W|8|@]|D|@]|E|$]]|$1|Z|3|10|5|U|7|1X|8|@]|D|@]|E|$]]|$1|11|3|12|5|6|7|1Y|8|@]|D|@]|E|$]]|$1|13|3|14|5|6|7|1Z|8|@$9|20|A|21|B|C]]|D|@]|E|$]]|$1|15|3|16|5|6|7|22|8|@]|D|@]|E|$]]|$1|17|3|18|5|6|7|23|8|@$9|24|A|25|B|C]|$9|26|A|27|B|C]|$9|28|A|29|B|C]]|D|@]|E|$]]|$1|19|3|-4|5|6|7|2A|8|@]|D|@]|E|$]]]|1A|$1B|$5|1C|1D|1E|E|$1F|G]]]]

You should use the <code>CallerRunsPolicy</code>, which executes the rejected task in the calling thread. This way, it can't submit any new tasks to the executor until that task is done, at which point there will be some free pool threads or the process will repeat.

<a href="http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/ThreadPoolExecutor.CallerRunsPolicy.html" rel="noreferrer">http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/ThreadPoolExecutor.CallerRunsPolicy.html</a>

From the docs:

<blockquote>
 Rejected tasks
 
 New tasks submitted in method execute(java.lang.Runnable) will be
 rejected when the Executor has been
 shut down, and also when the Executor
 uses finite bounds for both maximum
 threads and work queue capacity, and
 is saturated. In either case, the
 execute method invokes the
 RejectedExecutionHandler.rejectedExecution(java.lang.Runnable,
 java.util.concurrent.ThreadPoolExecutor)
 method of its
 RejectedExecutionHandler. Four
 predefined handler policies are
 provided:
 
 <ol>
 <li>In the default ThreadPoolExecutor.AbortPolicy, the
 handler throws a runtime
 RejectedExecutionException upon
 rejection.</li>
 <li>In ThreadPoolExecutor.CallerRunsPolicy,
 the thread that invokes execute itself
 runs the task. This provides a simple
 feedback control mechanism that will
 slow down the rate that new tasks are
 submitted.</li>
 <li>In ThreadPoolExecutor.DiscardPolicy, a
 task that cannot be executed is simply
 dropped.</li>
 <li>In ThreadPoolExecutor.DiscardOldestPolicy,
 if the executor is not shut down, the
 task at the head of the work queue is
 dropped, and then execution is retried
 (which can fail again, causing this to
 be repeated.)</li>
 </ol>
</blockquote>

Also, make sure to use a bounded queue, such as ArrayBlockingQueue, when calling the <code>ThreadPoolExecutor</code> constructor. Otherwise, nothing will get rejected.

Edit: in response to your comment, set the size of the ArrayBlockingQueue to be equal to the max size of the thread pool and use the AbortPolicy.

Edit 2: Ok, I see what you're getting at. What about this: override the <code>beforeExecute()</code> method to check that <code>getActiveCount()</code> doesn't exceed <code>getMaximumPoolSize()</code>, and if it does, sleep and try again?

blocks|key|322101|text|我刚刚找到的一个可能的解决方案是：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|322102|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;
++++++++}
++++}
}|code-block|syntax|javascript|322103|还有其他的解决方案吗？我更喜欢基于RejectedExecutionHandler的东西，因为它似乎是处理这种情况的标准方法。|offset|length|style|CODE|322104|entityMap^0|0|0|H|O|0^^$0|@$1|2|3|4|5|6|7|O|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|P|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|Q|8|@$I|R|J|S|K|L]]|9|@]|A|$]]|$1|M|3|-4|5|6|7|T|8|@]|9|@]|A|$]]]|N|$]]

One of the possible solutions I've just found:

<pre><code>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;
 }
 }
}
</code></pre>

Are there any other solutions? I'd prefer something based on <code>RejectedExecutionHandler</code> since it seems like a standard way to handle such situations.

blocks|key|3071429|text|创建您自己的阻塞队列以供Executor使用，并使用您正在寻找的阻塞行为，同时始终返回可用的剩余容量(确保executor不会尝试创建超过其核心池的线程，或触发拒绝处理程序)。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|3071430|我相信这会给你带来你想要的阻塞行为。拒绝处理程序永远不会符合要求，因为这表明执行器不能执行任务。我可以想象的是，你会在处理程序中得到某种形式的“忙碌等待”。这不是您想要的，您希望为阻止调用者的executor创建一个队列...|3071431|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|F|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|G|8|@]|9|@]|A|$]]|$1|D|3|-4|5|6|7|H|8|@]|9|@]|A|$]]]|E|$]]

Create your own blocking queue to be used by the Executor, with the blocking behavior you are looking for, while always returning available remaining capacity (ensuring the executor will not try to create more threads than its core pool, or trigger the rejection handler).

I believe this will get you the blocking behavior you are looking for. A rejection handler will never fit the bill, since that indicates the executor can not perform the task. What I could envision there is that you get some form of 'busy waiting' in the handler. That is not what you want, you want a queue for the executor that blocks the caller...

blocks|key|3284826|text|Hibernate有一个简单的BlockPolicy，可以做你想做的事情：|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|3284827|请参阅：Executors.java|3284828|/**
+*+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+);
++++++++}
++++}
}|code-block|syntax|javascript|3284829|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/hibernate/hibernate-search/blob/master/hibernate-search/src/main/java/org/hibernate/search/batchindexing/Executors.java^0|F|B|0|4|E|0|0|0^^$0|@$1|2|3|4|5|6|7|U|8|@$9|V|A|W|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|X|8|@]|D|@$9|Y|A|Z|1|10]]|E|$]]|$1|H|3|I|5|J|7|11|8|@]|D|@]|E|$K|L]]|$1|M|3|-4|5|6|7|12|8|@]|D|@]|E|$]]]|N|$O|$5|P|Q|R|E|$S|T]]]]

Hibernate has a <code>BlockPolicy</code> that is simple and may do what you want:

See: <a href="https://github.com/hibernate/hibernate-search/blob/master/hibernate-search/src/main/java/org/hibernate/search/batchindexing/Executors.java" rel="noreferrer">Executors.java</a>

<pre><code>/**
 * A handler for rejected tasks that will have the caller block until
 * space is available.
 */
public static class BlockPolicy implements RejectedExecutionHandler {

 /**
 * Creates a &lt;tt&gt;BlockPolicy&lt;/tt&gt;.
 */
 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 );
 }
 }
}
</code></pre>

blocks|key|3284833|text|您可以使用ThreadPoolExecutor和blockingQueue：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|3284834|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());
++++}
}|code-block|syntax|javascript|3284835|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

You can use ThreadPoolExecutor and a blockingQueue:

<pre><code>public class ImageManager {
 BlockingQueue&lt;Runnable&gt; blockingQueue = new ArrayBlockingQueue&lt;Runnable&gt;(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());
 }
}
</code></pre>

blocks|key|319578|text|在实践中，上面从Java+Concurrency引述的BoundedExecutor答案只有在您为Executor使用无界队列，或者信号量界限不大于队列大小时才能正确工作。信号量在提交线程和池中的线程之间共享状态，这使得即使队列大小<绑定<=+(队列大小%2B池大小)也可能使执行器饱和。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|319579|只有当您的任务不会永远运行时，使用CallerRunsPolicy才是有效的，在这种情况下，提交线程将永远保留在rejectedExecution中，如果您的任务需要很长时间才能运行，这不是一个好主意，因为如果提交线程本身正在运行任务，则它不能提交任何新任务或做任何其他事情。|319580|如果这是不可接受的，那么我建议在提交任务之前检查executor的有界队列的大小。如果队列已满，则在尝试再次提交之前等待一小段时间。吞吐量将受到影响，但我建议它是一个比许多其他建议的解决方案更简单的解决方案，并且您可以保证不会拒绝任何任务。|319581|entityMap^0|R|F|0|H|G|1K|H|0|0^^$0|@$1|2|3|4|5|6|7|L|8|@$9|M|A|N|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|O|8|@$9|P|A|Q|B|C]|$9|R|A|S|B|C]]|D|@]|E|$]]|$1|H|3|I|5|6|7|T|8|@]|D|@]|E|$]]|$1|J|3|-4|5|6|7|U|8|@]|D|@]|E|$]]]|K|$]]

The <code>BoundedExecutor</code> answer quoted above from Java Concurrency in Practice only works correctly if you use an unbounded queue for the Executor, or the semaphore bound is no greater than the queue size. The semaphore is state shared between the submitting thread and the threads in the pool, making it possible to saturate the executor even if queue size &lt; bound &lt;= (queue size + pool size).

Using <code>CallerRunsPolicy</code> is only valid if your tasks don't run forever, in which case your submitting thread will remain in <code>rejectedExecution</code> forever, and a bad idea if your tasks take a long time to run, because the submitting thread can't submit any new tasks or do anything else if it's running a task itself.

If that's not acceptable then I suggest checking the size of the executor's bounded queue before submitting a task. If the queue is full, then wait a short time before trying to submit again. The throughput will suffer, but I suggest it's a simpler solution than many of the other proposed solutions and you're guaranteed no tasks will get rejected.

blocks|key|319583|text|避免@FixPoint解决方案出现问题。可以使用ListeningExecutorService并在FutureCallback中释放信号量onSuccess和onFailure。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|319584|entityMap^0|0^^$0|@$1|2|3|4|5|6|7|D|8|@]|9|@]|A|$]]|$1|B|3|-4|5|6|7|E|8|@]|9|@]|A|$]]]|C|$]]

To avoid issues with @FixPoint solution. One could use ListeningExecutorService and release the semaphore onSuccess and onFailure inside FutureCallback.

blocks|key|321407|text|最近我发现这个问题也有同样的问题。OP没有明确说明这一点，但是我们不希望使用在提交者的线程上执行任务的RejectedExecutionHandler，因为如果该任务是一个长时间运行的任务，这将不会充分利用工作线程。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|321408|阅读所有的答案和评论，特别是使用信号量或使用afterExecute的有缺陷的解决方案，我仔细查看了ThreadPoolExecutor的代码，看看是否有解决办法。我很惊讶地看到有超过2000行(注释)的代码，其中一些让我有dizzy的感觉。考虑到我实际上有一个相当简单的需求-一个生产者，几个消费者，当没有消费者可以接受工作时，让生产者阻止-我决定推出我自己的解决方案。它不是一个ExecutorService，而仅仅是一个Executor。而且它不会根据工作负载调整线程的数量，而是只保留固定数量的线程，这也符合我的要求。下面是代码。请尽情抱怨:-)|321409|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;

++/*%2B**********************************************************************/
++/**
+++*+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%2B%2B)+{
++++++//+could+reuse+the+same+Runner+all+over,+but+keep+it+simple
++++++Thread+t+=+new+Thread(new+Runner());
++++++this.threads[i]+=+t;
++++++t.start();
++++}
++}
++/*%2B**********************************************************************/
++/**
+++*+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+"
++++++++++%2B+"worker.",+e);
++++}
++}
++/*%2B**********************************************************************/
++/**
+++*+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();
++++}
++}
++/*%2B**********************************************************************/
++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();
++++++++}
++++++}
++++}
++}
}|code-block|syntax|javascript|321410|entityMap|0|LINK|mutability|MUTABLE|url|http://grepcode.com/file/repository.grepcode.com/java/root/jdk/openjdk/8-b132/java/util/concurrent/ThreadPoolExecutor.java/|1|http://grepcode.com/file/repository.grepcode.com/java/root/jdk/openjdk/8-b132/java/util/concurrent/ThreadPoolExecutor.java/#1355^0|1F|O|0|M|C|5B|F|5X|8|1E|I|0|34|5|1|0|0^^$0|@$1|2|3|4|5|6|7|W|8|@$9|X|A|Y|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|Z|8|@$9|10|A|11|B|C]|$9|12|A|13|B|C]|$9|14|A|15|B|C]]|D|@$9|16|A|17|1|18]|$9|19|A|1A|1|1B]]|E|$]]|$1|H|3|I|5|J|7|1C|8|@]|D|@]|E|$K|L]]|$1|M|3|-4|5|6|7|1D|8|@]|D|@]|E|$]]]|N|$O|$5|P|Q|R|E|$S|T]]|U|$5|P|Q|R|E|$S|V]]]]

Recently I found this question having the same problem. The OP does not say so explicitly, but we do not want to use the <code>RejectedExecutionHandler</code> which executes a task on the submitter's thread, because this will under-utilize the worker threads if this task is a long running one.

Reading all the answers and comments, in particular the flawed solution with the semaphore or using <code>afterExecute</code> I had a closer look at the code of the <a href="http://grepcode.com/file/repository.grepcode.com/java/root/jdk/openjdk/8-b132/java/util/concurrent/ThreadPoolExecutor.java/" rel="nofollow">ThreadPoolExecutor</a> to see if there is some way out. I was amazed to see that there are more than 2000 lines of (commented) code, some of which make me feel <a href="http://grepcode.com/file/repository.grepcode.com/java/root/jdk/openjdk/8-b132/java/util/concurrent/ThreadPoolExecutor.java/#1355" rel="nofollow">dizzy</a>. Given the rather simple requirement I actually have --- one producer, several consumers, let the producer block when no consumers can take work --- I decided to roll my own solution. It is not an <code>ExecutorService</code> but just an <code>Executor</code>. And it does not adapt the number of threads to the work load, but holds a fixed number of threads only, which also fits my requirements. Here is the code. Feel free to rant about it :-)

<pre><code>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&lt;Runnable&gt; workQueue =
 new SynchronousQueue&lt;Runnable&gt;();
 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&lt;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();
 }
 }
 }
 }
}
</code></pre>

blocks|key|3285036|text|我相信通过使用java.util.concurrent.Semaphore和Executor.newFixedThreadPool的委托行为可以很好地解决这个问题。新的executor服务将仅在有线程执行新任务时执行新任务。阻塞由信号量管理，许可的数量等于线程的数量。当一个任务完成时，它会返回一个许可。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|3285037|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();
++++++++}
++++});
}|code-block|syntax|javascript|3285038|entityMap^0|7|U|12|R|0|0^^$0|@$1|2|3|4|5|6|7|M|8|@$9|N|A|O|B|C]|$9|P|A|Q|B|C]]|D|@]|E|$]]|$1|F|3|G|5|H|7|R|8|@]|D|@]|E|$I|J]]|$1|K|3|-4|5|6|7|S|8|@]|D|@]|E|$]]]|L|$]]

I believe there is quite elegant way to solve this problem by using <code>java.util.concurrent.Semaphore</code> and delegating behavior of <code>Executor.newFixedThreadPool</code>.
The new executor service will only execute new task when there is a thread to do so. Blocking is managed by Semaphore with number of permits equal to number of threads. When a task is finished it returns a permit.

<pre><code>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&lt;Runnable&gt; 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(() -&gt; {
 try {
 command.run();
 } finally {
 blockExecution.release();
 }
 });
}
</code></pre>

blocks|key|319636|text|我知道，这是一种黑客行为，但在我看来，这是这里提供的最干净的黑客行为;-)|type|unstyled|depth|inlineStyleRanges|entityRanges|data|319637|因为ThreadPoolExecutor使用阻塞队列"offer“而不是"put"，所以让我们覆盖阻塞队列的"offer”行为：|319638|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));|code-block|syntax|javascript|319639|我对它进行了测试，它似乎可以工作。实现一些超时策略留给读者练习。|319640|entityMap^0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|M|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|N|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|O|8|@]|9|@]|A|$G|H]]|$1|I|3|J|5|6|7|P|8|@]|9|@]|A|$]]|$1|K|3|-4|5|6|7|Q|8|@]|9|@]|A|$]]]|L|$]]

I know, it is a hack, but in my opinion most clean hack between those offered here ;-)

Because ThreadPoolExecutor uses blocking queue "offer" instead of "put", lets override behaviour of "offer" of the blocking queue:

<pre><code>class BlockingQueueHack&lt;T&gt; extends ArrayBlockingQueue&lt;T&gt; {

 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));
</code></pre>

I tested it and it seems to work.
Implementing some timeout policy is left as a reader's exercise.

blocks|key|3285059|text|我在过去也有同样的需求:一种由共享线程池支持的每个客户端具有固定大小的阻塞队列。我最终编写了我自己的ThreadPoolExecutor：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|3285060|UserThreadPoolExecutor+(阻塞队列(每个客户端)%2B线程池(在所有客户端之间共享))|3285061|请参阅：https://github.com/d4rxh4wx/UserThreadPoolExecutor|offset|length|3285062|为每个UserThreadPoolExecutor分配来自共享ThreadPoolExecutor的最大线程数|3285063|每个UserThreadPoolExecutor可以：|3285064|如果未达到共享线程池执行器的配额，则|3285065|将任务提交给共享线程池执行器。如果达到其配额，该作业将排队(非消耗阻塞等待CPU)。一旦完成其中一个已提交的任务，配额就会减少，从而允许另一个任务等待提交到ThreadPoolExecutor|unordered-list-item|3285066|wait，以便其余任务完成|3285067|3285068|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/d4rxh4wx/UserThreadPoolExecutor^0|0|0|4|1E|0|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|11|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|12|8|@]|9|@]|A|$]]|$1|D|3|E|5|6|7|13|8|@]|9|@$F|14|G|15|1|16]]|A|$]]|$1|H|3|I|5|6|7|17|8|@]|9|@]|A|$]]|$1|J|3|K|5|6|7|18|8|@]|9|@]|A|$]]|$1|L|3|M|5|6|7|19|8|@]|9|@]|A|$]]|$1|N|3|O|5|P|7|1A|8|@]|9|@]|A|$]]|$1|Q|3|R|5|P|7|1B|8|@]|9|@]|A|$]]|$1|S|3|-4|5|6|7|1C|8|@]|9|@]|A|$]]|$1|T|3|-4|5|6|7|1D|8|@]|9|@]|A|$]]]|U|$V|$5|W|X|Y|A|$Z|10]]]]

I had the same need in the past: a kind of blocking queue with a fixed size for each client backed by a shared thread pool. I ended up writing my own kind of ThreadPoolExecutor:

UserThreadPoolExecutor
 (blocking queue (per client) + threadpool (shared amongst all clients))

See: <a href="https://github.com/d4rxh4wx/UserThreadPoolExecutor" rel="nofollow">https://github.com/d4rxh4wx/UserThreadPoolExecutor</a>

Each UserThreadPoolExecutor is given a maximum number of threads from a shared ThreadPoolExecutor

Each UserThreadPoolExecutor can:

<ul>
<li>submit a task to the shared thread pool executor if its quota is not reached. If its quota is reached, the job is queued (non-consumptive blocking waiting for CPU). Once one of its submitted task is completed, the quota is decremented, allowing another task waiting to be submitted to the ThreadPoolExecutor</li>
<li>wait for the remaining tasks to complete</li>
</ul>

blocks|key|324232|text|下面的类包装了一个ThreadPoolExecutor并使用信号量阻塞，然后工作队列就满了：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|324233|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+%2B+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);
++++}
}|code-block|syntax|javascript|324234|此包装器类基于Brian+Goetz在《Java+Concurrency+in+Practice》一书中给出的解决方案。书中的解决方案只有两个构造函数参数:一个Executor和一个用于信号量的绑定。Fixpoint给出的答案中显示了这一点。这种方法有一个问题:它可能会进入池线程繁忙、队列已满，但信号量刚刚释放了许可的状态。(+finally块中的semaphore.release())。在这种状态下，新的任务可以抢占刚刚释放的权限，但会因为任务队列已满而被拒绝。当然，这不是您想要的；在这种情况下，您希望阻止。|offset|length|style|CODE|324235|为了解决这个问题，我们必须使用无界队列，正如JCiP明确提到的那样。信号量充当守卫，提供虚拟队列大小的效果。这有一个副作用，那就是该单元可能包含maxPoolSize+%2B+virtualQueueSize+%2B+maxPoolSize任务。为什么会这样呢？这是因为finally块中的semaphore.release()。如果所有池线程同时调用此语句，则释放maxPoolSize许可，允许相同数量的任务进入单元。如果我们使用的是有界队列，它仍然是满的，从而导致被拒绝的任务。现在，因为我们知道这只在池线程即将完成时才会发生，所以这不是问题。我们知道池线程不会阻塞，所以很快就会从队列中取出一个任务。|BOLD|324236|不过，您可以使用有界队列。只要确保它的大小等于virtualQueueSize+%2B+maxPoolSize即可。更大的尺寸是无用的，信号量将阻止更多的项目进入。较小的大小将导致拒绝的任务。任务被拒绝的机会随着规模的减小而增加。例如，假设您想要一个具有maxPoolSize=2和virtualQueueSize=5的有界执行器。然后获取一个具有5%2B2=7许可和实际队列大小为5%2B2=7的信号量。然后，可以在单元中的实际任务数为2%2B5%2B2=9。当执行器已满(队列中有5个任务，线程池中有2个任务，因此0个许可可用)并且所有池线程都释放了它们的许可时，进入的任务恰好可以获得2个许可。|324237|现在，来自JCiP的解决方案使用起来有些麻烦，因为它没有强制执行所有这些约束(无界队列，或有界的数学约束等)。我认为这只是一个很好的例子来演示如何基于已有的部分构建新的线程安全类，而不是作为一个完全成长的、可重用的类。我不认为后者是作者的本意。|324238|entityMap^0|0|0|28|8|4V|J|0|F|2|20|18|3X|J|4Z|B|0|N|U|0|0^^$0|@$1|2|3|4|5|6|7|V|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|W|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|X|8|@$I|Y|J|Z|K|L]|$I|10|J|11|K|L]]|9|@]|A|$]]|$1|M|3|N|5|6|7|12|8|@$I|13|J|14|K|O]|$I|15|J|16|K|L]|$I|17|J|18|K|L]|$I|19|J|1A|K|L]]|9|@]|A|$]]|$1|P|3|Q|5|6|7|1B|8|@$I|1C|J|1D|K|L]]|9|@]|A|$]]|$1|R|3|S|5|6|7|1E|8|@]|9|@]|A|$]]|$1|T|3|-4|5|6|7|1F|8|@]|9|@]|A|$]]]|U|$]]

The following class wraps around a ThreadPoolExecutor and uses a Semaphore to block then the work queue is full:

<pre><code>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&lt;Runnable&gt; queue = new LinkedBlockingQueue&lt;Runnable&gt;();
 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);
 }
}
</code></pre>

This wrapper class is based on a solution given in the book Java Concurrency in Practice by Brian Goetz. The solution in the book only takes two constructor parameters: an <code>Executor</code> and a bound used for the semaphore. This is shown in the answer given by Fixpoint. There is a problem with that approach: it can get in a state where the pool threads are busy, the queue is full, but the semaphore has just released a permit. (<code>semaphore.release()</code> in the finally block). In this state, a new task can grab the just released permit, but is rejected because the task queue is full. Of course this is not something you want; you want to block in this case.

To solve this, we must use an unbounded queue, as JCiP clearly mentions. The semaphore acts as a guard, giving the effect of a virtual queue size. This has the side effect that it is possible that the unit can contain <code>maxPoolSize + virtualQueueSize + maxPoolSize</code> tasks. Why is that? Because of the 
<code>semaphore.release()</code> in the finally block. If all pool threads call this statement at the same time, then <code>maxPoolSize</code> permits are released, allowing the same number of tasks to enter the unit. If we were using a bounded queue, it would still be full, resulting in a rejected task. Now, because we know that this only occurs when a pool thread is almost done, this is not a problem. We know that the pool thread will not block, so a task will soon be taken from the queue. 

You are able to use a bounded queue though. Just make sure that its size equals <code>virtualQueueSize + maxPoolSize</code>. Greater sizes are useless, the semaphore will prevent to let more items in. Smaller sizes will result in rejected tasks. The chance of tasks getting rejected increases as the size decreases. For example, say you want a bounded executor with maxPoolSize=2 and virtualQueueSize=5. Then take a semaphore with 5+2=7 permits and an actual queue size of 5+2=7. The real number of tasks that can be in the unit is then 2+5+2=9. When the executor is full (5 tasks in queue, 2 in thread pool, so 0 permits available) and ALL pool threads release their permits, then exactly 2 permits can be taken by tasks coming in.

Now the solution from JCiP is somewhat cumbersome to use as it doesn't enforce all these constraints (unbounded queue, or bounded with those math restrictions, etc.). I think that this only serves as a good example to demonstrate how you can build new thread safe classes based on the parts that are already available, but not as a full-grown, reusable class. I don't think that the latter was the author's intention.

blocks|key|324240|text|您可以像这样使用自定义RejectedExecutionHandler|type|unstyled|depth|inlineStyleRanges|entityRanges|data|324241|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());
++++++++++++++++++++++++}

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

you can use a custom RejectedExecutionHandler like this

<pre><code>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());
 }

 }
 });
</code></pre>

blocks|key|3285091|text|我在elastic+search+client中发现了这个拒绝策略。它在阻塞队列上阻塞调用者线程。下面的代码-|type|unstyled|depth|inlineStyleRanges|entityRanges|data|3285092|+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;
++++++++}
}|code-block|syntax|javascript|3285093|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

I found this rejection policy in elastic search client. It blocks caller thread on blocking queue. Code below- 

<pre><code> 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;
 }
}
</code></pre>

blocks|key|321434|text|我最近有一个类似的需求，但在ScheduledExecutorService上。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|321435|我还必须确保处理传递给方法的延迟，并确保任务要么像调用者期望的那样提交执行，要么就会失败，从而抛出RejectedExecutionException。|321436|来自ScheduledThreadPoolExecutor的用于在内部执行或提交任务的其他方法将调用#schedule，而后者仍将调用被覆盖的方法。|321437|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+%7C+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();
++++}
}|code-block|syntax|javascript|321438|我有代码在这里，将感谢任何反馈。https://github.com/AmitabhAwasthi/BlockingScheduler|321439|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/AmitabhAwasthi/BlockingScheduler^0|E|O|0|1D|Q|0|2|R|1E|9|0|0|G|1F|0|0^^$0|@$1|2|3|4|5|6|7|Y|8|@$9|Z|A|10|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|11|8|@$9|12|A|13|B|C]]|D|@]|E|$]]|$1|H|3|I|5|6|7|14|8|@$9|15|A|16|B|C]|$9|17|A|18|B|C]]|D|@]|E|$]]|$1|J|3|K|5|L|7|19|8|@]|D|@]|E|$M|N]]|$1|O|3|P|5|6|7|1A|8|@]|D|@$9|1B|A|1C|1|1D]]|E|$]]|$1|Q|3|-4|5|6|7|1E|8|@]|D|@]|E|$]]]|R|$S|$5|T|U|V|E|$W|X]]]]

I recently had a need to achieve something similar, but on a <code>ScheduledExecutorService</code>. 

I had to also ensure that I handle the delay being passed on the method and ensure that either the task is submitted to execute at the time as the caller expects or just fails thus throwing a <code>RejectedExecutionException</code>.

Other methods from <code>ScheduledThreadPoolExecutor</code> to execute or submit a task internally call <code>#schedule</code> which will still in turn invoke the methods overridden. 

<pre><code>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&lt;?&gt; schedule(Runnable command,
 long delay,
 TimeUnit unit) {
 final long newDelayInMs = beforeSchedule(command, unit.toMillis(delay));
 return super.schedule(command, newDelayInMs, TimeUnit.MILLISECONDS);
 }

 @Override
 public &lt;V&gt; ScheduledFuture&lt;V&gt; schedule(Callable&lt;V&gt; callable,
 long delay,
 TimeUnit unit) {
 final long newDelayInMs = beforeSchedule(callable, unit.toMillis(delay));
 return super.schedule(callable, newDelayInMs, TimeUnit.MILLISECONDS);
 }

 @Override
 public ScheduledFuture&lt;?&gt; 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&lt;?&gt; 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 &amp;&amp; runnable instanceof Future&lt;?&gt;) {
 try {
 ((Future&lt;?&gt;) 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();
 }
}
</code></pre>

I have the code here, will appreciate any feedback.
<a href="https://github.com/AmitabhAwasthi/BlockingScheduler" rel="nofollow">https://github.com/AmitabhAwasthi/BlockingScheduler</a>

blocks|key|322220|text|这是一个看起来非常有效的解决方案。它被称为NotifyingBlockingThreadPoolExecutor。|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|322221|Demo+program.|322222|编辑:这段代码有一个issue，await()方法有错误。调用shutdown()+%2B+awaitTermination()似乎工作得很好。|322223|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/USGS-CIDA/glri-afinch/blob/master/glri-afinch-util/src/main/java/org/java/util/concurrent/NotifyingBlockingThreadPoolExecutor.java|1|http://tinybrain.de/1011719|2|https://github.com/USGS-CIDA/glri-afinch/issues/125^0|L|Z|0|0|0|D|1|0|A|5|2|0^^$0|@$1|2|3|4|5|6|7|T|8|@]|9|@$A|U|B|V|1|W]]|C|$]]|$1|D|3|E|5|6|7|X|8|@]|9|@$A|Y|B|Z|1|10]]|C|$]]|$1|F|3|G|5|6|7|11|8|@]|9|@$A|12|B|13|1|14]]|C|$]]|$1|H|3|-4|5|6|7|15|8|@]|9|@]|C|$]]]|I|$J|$5|K|L|M|C|$N|O]]|P|$5|K|L|M|C|$N|Q]]|R|$5|K|L|M|C|$N|S]]]]

Here's the solution that seems to work really well. It's called <a href="https://github.com/USGS-CIDA/glri-afinch/blob/master/glri-afinch-util/src/main/java/org/java/util/concurrent/NotifyingBlockingThreadPoolExecutor.java" rel="nofollow noreferrer">NotifyingBlockingThreadPoolExecutor</a>.

<a href="http://tinybrain.de/1011719" rel="nofollow noreferrer">Demo program.</a>

Edit: There is an <a href="https://github.com/USGS-CIDA/glri-afinch/issues/125" rel="nofollow noreferrer">issue</a> with this code, the await() method is buggy. Calling shutdown() + awaitTermination() seems to work fine.

blocks|key|3285109|text|我并不总是喜欢CallerRunsPolicy，特别是因为它允许被拒绝的任务“跳过队列”，并在较早提交的任务之前执行。此外，在调用线程上执行任务可能比等待第一个插槽变为可用所需的时间要长得多。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|3285110|我使用一个自定义的RejectedExecutionHandler解决了这个问题，它只是简单地阻塞了调用线程一段时间，然后再次尝试提交任务：|3285111|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);
++++}
}|code-block|syntax|javascript|3285112|这个类可以像其他类一样在线程池执行器中作为RejectedExecutinHandler使用，例如：|3285113|executorPool+=+new+ThreadPoolExecutor(1,+1,+10,
++++++++++++++++++++++++++++++++++++++TimeUnit.SECONDS,+new+SynchronousQueue<Runnable>(),
++++++++++++++++++++++++++++++++++++++new+BlockWhenQueueFull());|3285114|我看到的唯一缺点是，调用线程可能会被锁定的时间略长于严格必要的时间(长达250ms)。此外，由于此executor实际上是以递归方式调用的，因此等待线程变得可用的时间过长(小时)可能会导致堆栈溢出。|3285115|尽管如此，我个人还是喜欢这种方法。它结构紧凑，易于理解，并且运行良好。|3285116|entityMap^0|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|S|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|T|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|U|8|@]|9|@]|A|$G|H]]|$1|I|3|J|5|6|7|V|8|@]|9|@]|A|$]]|$1|K|3|L|5|F|7|W|8|@]|9|@]|A|$G|H]]|$1|M|3|N|5|6|7|X|8|@]|9|@]|A|$]]|$1|O|3|P|5|6|7|Y|8|@]|9|@]|A|$]]|$1|Q|3|-4|5|6|7|Z|8|@]|9|@]|A|$]]]|R|$]]

I don't always like the CallerRunsPolicy, especially since it allows the rejected task to 'skip the queue' and get executed before tasks that were submitted earlier. Moreover, executing the task on the calling thread might take much longer than waiting for the first slot to become available.

I solved this problem using a custom RejectedExecutionHandler, which simply blocks the calling thread for a little while and then tries to submit the task again:

<pre><code>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);
 }
}
</code></pre>

This class can just be used in the thread-pool executor as a RejectedExecutinHandler like any other, for example:

<pre><code>executorPool = new ThreadPoolExecutor(1, 1, 10,
 TimeUnit.SECONDS, new SynchronousQueue&lt;Runnable&gt;(),
 new BlockWhenQueueFull());
</code></pre>

The only downside I see is that the calling thread might get locked slightly longer than strictly necessary (up to 250ms). Moreover, since this executor is effectively being called recursively, very long waits for a thread to become available (hours) might result in a stack overflow.

Nevertheless, I personally like this method. It's compact, easy to understand, and works well.

I want to create a <code>ThreadPoolExecutor</code> such that when it has reached its maximum size and the queue is full, the <code>submit()</code> method blocks when trying to add new tasks. Do I need to implement a custom <code>RejectedExecutionHandler</code> for that or is there an existing way to do this using a standard Java library?

How to make ThreadPoolExecutor's submit() method block if it is saturated?

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

我想创建一个ThreadPoolExecutor，这样当它达到最大大小并且队列已满时，submit()方法会在尝试添加新任务时阻塞。我是否需要为此实现一个自定义的RejectedExecutionHandler，或者是否有一种现有的方法可以使用标准的Java库来实现此目的？

问如果ThreadPoolExecutor的submit()方法饱和，该如何使其阻塞？
EN

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如果ThreadPoolExecutor的submit()方法饱和，该如何使其阻塞？EN