同步队列在被线程池执行器提供任务时不会阻塞。

Question

我使用几乎默认的newCachedThreadPool，但我想限制线程创建，所以我像这样创建ExecutorService

new ThreadPoolExecutor(0, Runtime.getRuntime().availableProcessors() * 2,
            60L, TimeUnit.SECONDS,
            new SynchronousQueue<>());

阅读了javadocs后，我希望当我提交Runnable时，任务执行器会以这样的方式工作：当向同步队列提供新任务时，任务执行器会阻塞，直到有可用线程来执行它，然后进行交接。但不幸的是，在达到线程池容量且所有线程都繁忙时，任务执行器会抛出RejectedExecutionException异常。我知道我可以传递一个RejectedExecutionHandler来阻塞，但我很惊讶似乎我必须这样做。请问是否真的按照预期工作，或者我做错了什么？

以下是代码示例：

public static void main(String[] args) {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(0, Runtime.getRuntime().availableProcessors() * 2,
            60L, TimeUnit.SECONDS,
            new SynchronousQueue<>());

    while (true) {
        executor.submit(() -> System.out.println("bla"));
    }
}

- Łukasz Chorąży

---

如果您能附上能够重现您的情况的代码，那就太好了。 - nikita_pavlenko

Answer 1

这符合ThreadPoolExecutor API的规定：

• 如果一个请求无法排队，除非超过了maximumPoolSize，否则会创建一个新的线程，否则该任务将被拒绝。

至于为什么SynchronousQueue不会阻塞 - 因为ThreadPoolExecutor使用queue.offer()而不是put()

- Evgeniy Dorofeev

---

这种行为可能会让人感到困惑，而且在Javadoc中没有以任何方式表达出来。 - Łukasz Chorąży

@ŁukaszChorąży 它确实谈到了直接移交，使用“...一个同步队列将任务移交给线程而不会以其他方式保持它们”，这意味着如果没有等待或可用的线程，则任务将被拒绝。 - matt

Answer 2

SynchronousQueue不会阻塞，直到有其他线程取走等待的元素，因为它使用了offer方法。它只是无法将元素添加到队列中。阻塞队列的部分在于take方法，它会一直阻塞，直到有元素被添加进来。

SynchronousQueue<Integer> que = new SynchronousQueue<>();

    System.out.println(que.offer(1));
    Object lock = new Object();
    synchronized(lock){
        new Thread(()->{ 
            synchronized(lock){
                lock.notify();
            }
            try{
                que.take();
            }
            catch(Exception e){}
        }
    ).start();
        lock.wait();
    }
    System.out.println(que.offer(1));

这个例子将输出false，然后可能（稍微有竞争条件）输出true。第一个add操作失败是因为没有人在等待接收提供的元素。

- matt