Java 并发实践 : race condition in BoundedExecutor?

Question

在《Java 并发实践》一书中，BoundedExecutor 的实现有些奇怪。

当有足够多的线程在执行器中排队或运行时，它应该通过阻塞提交线程来限制向执行器提交任务。

这是实现(在 catch 子句中添加缺少的重新抛出之后):

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() {
                @Override public void run() {
                    try {
                        command.run();
                    } finally {
                        semaphore.release();
                    }
                }
            });
        } catch (RejectedExecutionException e) {
            semaphore.release();
            throw e;
        }
    }

当我使用 Executors.newCachedThreadPool() 和 4 的边界实例化 BoundedExecutor 时，我希望缓存线程池实例化的线程数永远不会超过 4。然而，在实践中，确实如此。我得到了这个小测试程序来创建多达 11 个线程:

public static void main(String[] args) throws Exception {
    class CountingThreadFactory implements ThreadFactory {
        int count;

        @Override public Thread newThread(Runnable r) {
            ++count;
            return new Thread(r);
        }           
    }

    List<Integer> counts = new ArrayList<Integer>();

    for (int n = 0; n < 100; ++n) {
        CountingThreadFactory countingThreadFactory = new CountingThreadFactory();
        ExecutorService exec = Executors.newCachedThreadPool(countingThreadFactory);

        try {
            BoundedExecutor be = new BoundedExecutor(exec, 4);

            for (int i = 0; i < 20000; ++i) {
                be.submitTask(new Runnable() {
                    @Override public void run() {}
                });
            }
        } finally {
            exec.shutdown();
        }

        counts.add(countingThreadFactory.count);
    }

    System.out.println(Collections.max(counts));
}

我认为在释放信号量和任务结束之间有一个很小的时间范围，另一个线程可以在释放线程尚未完成时获得许可并提交任务。换句话说，它具有竞争条件。

有人可以确认吗？

Answer 1

BoundedExecutor 确实旨在说明如何限制任务提交，而不是限制线程池大小。至少有一条评论指出，有更直接的方法可以实现后者。

但是其他答案没有提到书中的文字说使用无界队列和

set the bound on the semaphore to be equal to the pool size plus the number of queued tasks you want to allow, since the semaphore is bounding the number of tasks both currently executing and awaiting execution. [JCiP, end of section 8.3.3]

通过提及无界队列和池大小，我们暗示(显然不是很清楚)使用有界大小的线程池。

然而，关于 BoundedExecutor 一直困扰我的是它没有实现 ExecutorService 接口(interface)。一种实现类似功能并仍然实现标准接口(interface)的现代方法是使用 Guava 的 listeningDecorator。方法和ForwardingListeningExecutorService类。