概述

Spring ThreadPoolTaskExecutor是一个JavaBean，它提供了一个基于java.util.concurrent.ThreadPoolExecutor和org.springframework.core.task.TaskExecutior的线程池实例。

可以通过corePoolSize、maxPoolSize、queueCapacity、allowCoreThreadTimeOut和keepAliveSeconds等属性进行细粒度配置。

本文重点示例corePoolSize和maxPoolSize之间的区别。

corePoolSize vs maxPoolSize

• corePoolSize：不超时的情况下保持生存的最小工作线程数。如果我们将allowCoreThreadTimeOut设置为true，则相当于将corePoolSize的值设置为零。
• maxPoolSize：可以创建的最大线程数。maxPoolSize取决于queueCapacity，因为ThreadPoolTaskExecutor只有在其队列中的任务数超过queueCapacity时才会创建新线程。

当我们向ThreadPoolTaskExecutor提交一个新任务时，如果运行的线程少于corePoolSize，即使池中有空闲线程，或者运行的线程低于maxPoolSize，并且queueCapacity定义的队列已满，都会创建一个新线程。

示例

首先，假设我们有一个实现ThreadPoolTaskExecutor执行新线程的方法startThreads：

public void startThreads(ThreadPoolTaskExecutor taskExecutor, CountDownLatch countDownLatch, 
  int numThreads) {
    for (int i = 0; i < numThreads; i++) {
        taskExecutor.execute(() -> {
            try {
                Thread.sleep(100L * ThreadLocalRandom.current().nextLong(1, 10));
                countDownLatch.countDown();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
    }
}

• 让我们测试ThreadPoolTaskExecutor的默认配置，它定义了只有一个线程的corePoolSize、无界的maxPoolSize和无界的queueCapacity。因此，无论我们启动多少任务，都只有一个线程在运行：

@Test
public void whenUsingDefaults_thenSingleThread() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.afterPropertiesSet();

    CountDownLatch countDownLatch = new CountDownLatch(10);
    this.startThreads(taskExecutor, countDownLatch, 10);

    while (countDownLatch.getCount() > 0) {
        Assert.assertEquals(1, taskExecutor.getPoolSize());
    }
}

• 如果将corePoolSize更改为最多5个线程，其他参数仍然按默认值，无论提交给ThreadPoolTaskExecutor的任务数量如何，都会启动5个线程：

@Test
public void whenCorePoolSizeFive_thenFiveThreads() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(5);
    taskExecutor.afterPropertiesSet();

    CountDownLatch countDownLatch = new CountDownLatch(10);
    this.startThreads(taskExecutor, countDownLatch, 10);

    while (countDownLatch.getCount() > 0) {
        Assert.assertEquals(5, taskExecutor.getPoolSize());
    }
}

• 类似地，可以将maxPoolSize增加到10，同时将corePoolSize保留为5。无论提交给ThreadPoolTaskExecutor的任务数量如何，也都只有5个线程启动，因为queueCapacity仍然是无限制的：

@Test
public void whenCorePoolSizeFiveAndMaxPoolSizeTen_thenFiveThreads() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(5);
    taskExecutor.setMaxPoolSize(10);
    taskExecutor.afterPropertiesSet();

    CountDownLatch countDownLatch = new CountDownLatch(10);
    this.startThreads(taskExecutor, countDownLatch, 10);

    while (countDownLatch.getCount() > 0) {
        Assert.assertEquals(5, taskExecutor.getPoolSize());
    }
}

• 我们现在重复前面的测试，但是将queueCapacity增加到10，并启动20个线程。因此总共会启动10个线程：

@Test
public void whenCorePoolSizeFiveAndMaxPoolSizeTenAndQueueCapacityTen_thenTenThreads() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(5);
    taskExecutor.setMaxPoolSize(10);
    taskExecutor.setQueueCapacity(10);
    taskExecutor.afterPropertiesSet();

    CountDownLatch countDownLatch = new CountDownLatch(20);
    this.startThreads(taskExecutor, countDownLatch, 20);

    while (countDownLatch.getCount() > 0) {
        Assert.assertEquals(10, taskExecutor.getPoolSize());
    }
}

同样，如果我们将queueCapacity设置为0，并且只启动10个任务，那么ThreadPoolTaskExecutor中也将会有10个线程。

结论

初始化线程池ThreadPoolTaskExecutor要显式指定corePoolSize、maxPoolSize、queueCapacity这三个参数的值，不能用默认值避免任务提交累积导致内存溢出OOM。

corePoolSize可以考虑与内核数保持一致。maxPoolSize和queueCapacity需要考虑到并发吞吐量来确定其值。