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-loop 第一次运行时，execute 将创建新的工作人员，并为他们提供您的任务，而无需尝试将它们推入队列 (addWorker+return)，因此 2 个任务立即开始，3个进入可以容纳所有3个的队列。

第二次，submit-s 会以 workQueue.offer 结束，这可能会使队列饱和（取决于工作人员尝试消费新项目的速度），当它饱和时， last-effort addWorker 将运行并失败，导致reject，因为不允许创建新的工作人员。

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

Answer 2

我认为由于您的代码示例，您对线程池的工作方式有一些误解。我尝试运行它并从 5 个任务和无限数量的 RejectedExecutionException 中获取输出。发生这种情况是因为在出现异常Thread.currentThread().wait(10000); 的情况下没有调用，另外 5 个任务被添加到池中，并且这个逻辑一次又一次地重复产生新的异常。尝试包围 consumer.submit(task);使用 try-catch 块，您将看到只有两个线程按预期处理所有任务，因为 keepTimeAlive 比等待时间长。在第二个示例中，keepTimeAlive 比等待时间短，因此在每次等待后都会创建新的非核心线程，并且在每次循环调用后您会看到不同的 id。这是正确的，因为之前的非核心线程已停止，因为它的空闲时间超过了 keepTimeAlive。