java - 覆盖 ThreadPoolExecutor 中的 execute()

Question

我正在寻找一个 ThreadPoolExecutor，它会在任务队列已满时阻塞 - 如果底层队列已满，当前的 Java 实现将拒绝新任务 -

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    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);
}

会改变这一行:

 if (isRunning(c) && workQueue.offer(command)) {

至

 if (isRunning(c) && workQueue.put(command)) {

有戏吗？我错过了什么吗？

解决方案(可能会帮助下一个人):

public class BlockingThreadPoolExecutor extends ThreadPoolExecutor {

    private final Semaphore runLock;

    public BlockingThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, int maxTasks) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        runLock = new Semaphore(maxTasks);
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        runLock.acquireUninterruptibly();
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        runLock.release();
    }

}

Answer 1

取决于 ThreadPoolExecutor 状态和设置，因为并非所有任务提交都通过 BlockingQueue。通常，您只想将 ThreadPoolExecutor 的 RejectedExecutionHandler 更改为 ThreadPoolExecutor.CallerRunsPolicy，这将限制提交。如果您真的想在提交时阻止，那么您应该使用 CompletionService 并在您想要阻止时调用“take”方法。您还可以创建一个子类并使用 Semaphore 来阻止执行方法。参见 JDK-6648211 : Need for blocking ThreadPoolExecutor获取更多信息。