JAVA线程池复用线程

Question

大家。我对使用线程池有误解。实际结果与该类的 API 描述不同。当我在线程池中使用 LinkedBlockedQueue 时，它不会重用线程，线程池等待在构造函数中设置的 KeepAliveTime，然后终止该线程并创建一个新线程。当我将 KeepAliveTime 设置得较小时，比如 1 秒或更短，它会删除线程并重新创建它，但是如果我设置一分钟，则不会创建新线程，因为 MaxPoolSize 不允许它并且队列已经满了所以所有任务都被拒绝，但是 keepAliveTime 设置分钟的线程这次什么都不做。我很新，不明白为什么它不重用这些线程。在 keepTimeAlive 过期后，它会杀死这些线程，如果队列已满，它会创建一个新线程。为什么会这样？据我从 API 了解，如果线程在 keepAliveTime 期间处于空闲状态，则必须重用它。当我使用 SynchronousQueue 而不是 LinkedBlockingQueue 时，它会重用线程。

public class Main {

    private volatile int remainingTasksCount;
    private volatile static ThreadPoolExecutor consumer = new ThreadPoolExecutor(1, 2, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(3));

    private static Runnable task = () -> {
        System.out.println(String.format("consumer %s, id %s, size %s, active count %s, queue %s",
                Thread.currentThread().getName(), Thread.currentThread().getId(),
                consumer.getPoolSize(), consumer.getActiveCount(), 3-consumer.getQueue().remainingCapacity()));
        String s = new String();
        synchronized (s) {
            try {
                s.wait(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    };


    public static void main(String[] args) throws IOException {
        try {
            new Thread(() -> {
                while (true) {
                    try {
                        for (int i = 0; i < 5; i++) {
                            consumer.submit(task);
                        }
                        System.out.println("PUSH TASKS");
                        synchronized (Thread.currentThread()) {
                            Thread.currentThread().wait(10000);
                        }
                    } catch (Throwable th) {
                        System.out.println(th);
                    }
                }
            }).start();
        } catch (Throwable th) {
            System.out.println(th);
        }
    }  

输出

PUSH TASKS
consumer pool-1-thread-1, id 15, size 2, active count 2, queue 3
consumer pool-1-thread-2, id 16, size 2, active count 2, queue 3
consumer pool-1-thread-2, id 16, size 2, active count 2, queue 1
consumer pool-1-thread-1, id 15, size 2, active count 1, queue 2
consumer pool-1-thread-1, id 15, size 2, active count 1, queue 0
Disconnected from the target VM, address: '127.0.0.1:64434', transport: 'socket'

Process finished with exit code 1

但下次生产者提交任务时，我得到 RejectedExecutionException

如果我将 keepAliveTime 更改为 1 Second。一切都运作良好，但创造新主题。

PUSH TASKS
consumer pool-1-thread-2, id 16, size 2, active count 2, queue 3
consumer pool-1-thread-1, id 15, size 2, active count 2, queue 3
consumer pool-1-thread-2, id 16, size 2, active count 2, queue 2
consumer pool-1-thread-1, id 15, size 2, active count 2, queue 1
consumer pool-1-thread-2, id 16, size 2, active count 1, queue 0
PUSH TASKS
consumer pool-1-thread-3, id 17, size 2, active count 2, queue 3
consumer pool-1-thread-2, id 16, size 2, active count 2, queue 2
consumer pool-1-thread-3, id 17, size 2, active count 2, queue 1
consumer pool-1-thread-2, id 16, size 2, active count 2, queue 1
consumer pool-1-thread-3, id 17, size 2, active count 1, queue 0
consumer pool-1-thread-3, id 17, size 1, active count 1, queue 2
PUSH TASKS
consumer pool-1-thread-4, id 18, size 2, active count 2, queue 3
consumer pool-1-thread-3, id 17, size 2, active count 2, queue 1
consumer pool-1-thread-4, id 18, size 2, active count 2, queue 1
consumer pool-1-thread-3, id 17, size 2, active count 1, queue 0
PUSH TASKS
consumer pool-1-thread-3, id 17, size 2, active count 2, queue 2
consumer pool-1-thread-5, id 19, size 2, active count 2, queue 3
consumer pool-1-thread-3, id 17, size 2, active count 2, queue 1
consumer pool-1-thread-5, id 19, size 2, active count 2, queue 1
consumer pool-1-thread-3, id 17, size 2, active count 1, queue 0

如果有人能解释我的错误，或者我错过的一些基本原则，我会很高兴

Answer 1

这是一个竞争条件。如果您遵循 submit() 足够长的时间(在源代码中)，您将到达 ThreadPoolExecutor.execucte():

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /* long comment block removed */
    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);
}

当您的 submit 循环第一次运行时，execute 将创建新的 worker，并向他们提供您的任务，而不是尝试将它们插入队列(addWorker+return), 所以 2 个任务立即开始，3 个进入可以容纳所有 3 个任务的队列。

第二次，submit-s 将以 workQueue.offer 结束，这可能会使队列饱和(取决于工作人员尝试的速度消费新元素)，当它这样做时，最后的努力 addWorker 将运行并失败，导致 reject，因为不允许创建新的 worker。

实际上，如果您开始在提交循环中做“事情”，它最终会开始工作。例如，我尝试 println(i)，但它的速度足够慢，可以消耗一些任务并使循环成功。当我尝试 print(i) 时已经太快了，它在第 4 次提交时就死了，所以没有任务很快被消耗掉。所以这是一个微妙的问题，竞争条件通常是什么。