Java：ExecutorService在特定队列大小后阻止提交[重复]

Question

我正在尝试编写一个解决方案，其中单个线程产生可以并行执行的I/O密集型任务。每个任务都有重要的内存数据。所以我希望能够限制一次挂起的任务数量。

如果我像这样创建ThreadPoolExecator：

    ThreadPoolExecutor executor = new ThreadPoolExecutor(numWorkerThreads, numWorkerThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>(maxQueue));

然后是执行器。submit（可调用）在队列已满且所有线程都已忙时抛出RejectedExecutionException。

我该怎么做才能成为执行者。提交（可调用）在队列已满且所有线程都忙时阻止？

编辑：我尝试了以下操作：

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

它在某种程度上实现了我想要实现的效果，但以一种不优雅的方式（基本上被拒绝的线程在调用线程中运行，因此这会阻止调用线程提交更多）。

编辑：（提问后5年）

对于阅读此问题及其答案的任何人，请不要将已接受的答案视为一个正确答案。请通读所有答案和评论。

Answer 1

当前接受的答案有一个潜在的重大问题-它改变了ThreadPoolExecutor的行为。执行时，如果您有一个corePoolSize

来自ThreadPoolExecutor。执行（可运行）：

    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);

具体来说，最后的“else”块永远不会被击中。

一个更好的替代方案是做一些类似于OP已经在做的事情-使用Reject tedExecutionHandler执行相同的put逻辑：

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);
    }
}

正如评论中所指出的（参考此答案），这种方法需要注意一些事情：

1. 如果corePoolSize==0，则存在竞态条件，池中的所有线程都可能在任务可见之前死亡
2. 使用包装队列任务的实现（不适用于ThreadPoolExecutor）将导致问题，除非处理程序也以相同的方式包装它

记住这些问题，此解决方案将适用于大多数典型的ThreadPoolExecutors，并将正确处理corePoolSize

Answer 2

这是我如何解决这个问题的：

（注意：此解决方案会阻止提交可调用的线程，因此可以防止抛出RejectedExecutionException）

public class BoundedExecutor extends ThreadPoolExecutor{

    private final Semaphore semaphore;

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

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

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


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

        semaphore.release();
    }
}

Answer 3

我也做过同样的事。诀窍是创建一个BlockingQueue，其中offer（）方法实际上是put（）。（您可以使用所需的任何基本BlockingQueue impl）。

public class LimitedQueue<E> extends LinkedBlockingQueue<E> 
{
    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;
    }

}

注意，这只适用于corePoolSize==maxPoolSize的线程池，所以要小心（参见注释）。