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

如果我将保持AliveTime更改为1秒。一切正常，但会创建新线程。

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

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

Answer 2

这是一个竞争条件。如果您遵循的提交（）足够长的时间（在源代码中），您将到达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);
}

当您的提交循环第一次运行时，执行将创建新的辅助角色，并为他们提供您的任务，而无需尝试将它们推入队列（addWorker返回），因此2个任务立即开始，3个进入可以容纳所有3个任务的队列。

第二次，submit-s将以工作队列结束。提供东西，这可能会使队列饱和（取决于工作人员尝试使用新物品的速度），当它达到饱和时，最后一次尝试添加工作人员将运行并失败，导致拒绝，因为不允许创建新工作人员。

实际上，如果你开始在提交循环中做“事情”，它最终会开始工作。例如，我试着println（I），这足够慢，可以消耗一些任务，循环成功。当我尝试打印（I）时，它已经太快了，在第4次提交时就消失了，所以没有任务很快被消耗掉。所以这是一个微妙的问题，通常情况下的种族条件是什么。