JavaSE-线程池（5）- 建议使用的方式

JavaSE-线程池（5）- 建议使用的方式

虽然JDK Executors 工具类提供了默认的创建线程池的方法，但一般建议自定义线程池参数，下面是阿里巴巴开发手册给出的理由：
另外Spring也提供了线程池的实现，比如 ThreadPoolExecutor

ThreadPoolExecutor

如下，初始化一个核心线程数为 2 ，最大线程数为 4，队列长度为 2 的线程池，其中 initialize 为线程池具体初始化方法

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.ThreadPoolExecutor;public class ThreadPoolTaskExecutorTest {static class MyTask implements Runnable {private int i;public MyTask(int i) {this.i = i;}@Overridepublic void run() {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread() + " " + i);}}public static void main(String[] args) {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();//设置核心线程数executor.setCorePoolSize(2);//设置最大线程数executor.setMaxPoolSize(4);//设置线程被回收的空闲时长executor.setKeepAliveSeconds(6);//设置队列容量executor.setQueueCapacity(2);//设置线程前缀executor.setThreadNamePrefix("t-");//设置拒绝策略executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());//初始化线程池executor.initialize();for (int i = 1; i <= 8; i++) {try {executor.execute(new MyTask(i));} catch (Exception e) {e.printStackTrace();}}}
}

由打印结果可以看出7,8任务由于线程数达到 maxPoolSize，且队列也填充满，线程池执行了拒绝策略

18:47:12.134 [main] INFO org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Initializing ExecutorService
org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@63e2203c[Running, pool size = 4, active threads = 4, queued tasks = 2, completed tasks = 0]] did not accept task: com.hs.example.base.multithread.threadpool.ThreadPoolTaskExecutorTest$MyTask@3224f60bat org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:317)at com.hs.example.base.multithread.threadpool.ThreadPoolTaskExecutorTest.main(ThreadPoolTaskExecutorTest.java:39)
Caused by: java.util.concurrent.RejectedExecutionException: Task com.hs.example.base.multithread.threadpool.ThreadPoolTaskExecutorTest$MyTask@3224f60b rejected from java.util.concurrent.ThreadPoolExecutor@63e2203c[Running, pool size = 4, active threads = 4, queued tasks = 2, completed tasks = 0]at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:314)... 1 more
org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@63e2203c[Running, pool size = 4, active threads = 4, queued tasks = 2, completed tasks = 0]] did not accept task: com.hs.example.base.multithread.threadpool.ThreadPoolTaskExecutorTest$MyTask@4bbfb90aat org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:317)at com.hs.example.base.multithread.threadpool.ThreadPoolTaskExecutorTest.main(ThreadPoolTaskExecutorTest.java:39)
Caused by: java.util.concurrent.RejectedExecutionException: Task com.hs.example.base.multithread.threadpool.ThreadPoolTaskExecutorTest$MyTask@4bbfb90a rejected from java.util.concurrent.ThreadPoolExecutor@63e2203c[Running, pool size = 4, active threads = 4, queued tasks = 2, completed tasks = 0]at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:314)... 1 more
Thread[t-4,5,main] 6
Thread[t-1,5,main] 1
Thread[t-3,5,main] 5
Thread[t-2,5,main] 2
Thread[t-3,5,main] 3
Thread[t-4,5,main] 4

ThreadPoolTaskExecutor initialize 方法

在上述例子中，initialize 方法是线程池初始化的具体实现，源码如下：

public void initialize() {if (this.logger.isInfoEnabled()) {this.logger.info("Initializing ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));}if (!this.threadNamePrefixSet && this.beanName != null) {this.setThreadNamePrefix(this.beanName + "-");}。；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；；this.executor = this.initializeExecutor(this.threadFactory, this.rejectedExecutionHandler);
}

由以下代码可以看出 ThreadPoolExecutor 是对 ThreadPoolExecutor 的封装，其中 this.threadFactory即为 ThreadPoolExecutor 本身，由下面的类结构图可以看出 ThreadPoolExecutor 继承自 ExecutorConfigurationSupport，而ExecutorConfigurationSupport实现了 ThreadFactory 接口

protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {BlockingQueue<Runnable> queue = this.createQueue(this.queueCapacity);ThreadPoolExecutor executor;if (this.taskDecorator != null) {executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, (long)this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler) {public void execute(Runnable command) {Runnable decorated = ThreadPoolTaskExecutor.this.taskDecorator.decorate(command);if (decorated != command) {ThreadPoolTaskExecutor.this.decoratedTaskMap.put(decorated, command);}super.execute(decorated);}};} else {executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, (long)this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler);}if (this.allowCoreThreadTimeOut) {executor.allowCoreThreadTimeOut(true);}this.threadPoolExecutor = executor;return executor;
}

ThreadPoolTaskExecutor 类结构图：

直接注入ThreadPoolTaskExecutor

除了手动实例化 ThreadPoolTaskExecutor 外，也可以直接注入 ThreadPoolTaskExecutor ，如下例：

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.test.context.junit4.SpringRunner;import java.util.concurrent.CountDownLatch;@RunWith(SpringRunner.class)
@SpringBootTest
public class ThreadPoolTaskExecutorTests {static class MyTask implements Runnable {private int i;private CountDownLatch countDownLatch;public MyTask(int i, CountDownLatch countDownLatch) {this.i = i;this.countDownLatch = countDownLatch;}@Overridepublic void run() {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread() + " " + i);countDownLatch.countDown();}}@Autowiredprivate ThreadPoolTaskExecutor taskExecutor;@Testpublic void contextLoads() {CountDownLatch countDownLatch = new CountDownLatch(8);long start = System.currentTimeMillis();for (int i = 1; i <= 8; i++) {try {taskExecutor.execute(new MyTask(i, countDownLatch));} catch (Exception e) {e.printStackTrace();}}try {countDownLatch.await();} catch (InterruptedException e) {e.printStackTrace();}System.out.println("耗时：" + (System.currentTimeMillis() - start));}}

2023-02-19 21:25:15.219  INFO 123964 --- [           main] c.h.e.aop.ThreadPoolTaskExecutorTests    : Starting ThreadPoolTaskExecutorTests on 0IZV69K0AKR0ELX with PID 123964 (started by Administrator in D:\workspace\idea_workspace\idea-test\example\2-aop)
2023-02-19 21:25:15.220  INFO 123964 --- [           main] c.h.e.aop.ThreadPoolTaskExecutorTests    : No active profile set, falling back to default profiles: default
2023-02-19 21:25:16.349  INFO 123964 --- [           main] c.h.e.aop.ThreadPoolTaskExecutorTests    : Started ThreadPoolTaskExecutorTests in 1.571 seconds (JVM running for 3.269)
2023-02-19 21:25:16.382  INFO 123964 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
Thread[task-8,5,main] 8
Thread[task-6,5,main] 6
Thread[task-4,5,main] 4
Thread[task-2,5,main] 2
Thread[task-3,5,main] 3
Thread[task-5,5,main] 5
Thread[task-7,5,main] 7
Thread[task-1,5,main] 1
耗时：1005
2023-02-19 21:25:17.637  INFO 123964 --- [       Thread-2] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'applicationTaskExecutor'

之所以可以直接使用 ThreadPoolTaskExecutor ，是因为SpringBoot自动注入了此类，具体看
org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration：

package org.springframework.boot.autoconfigure.task;import java.util.concurrent.Executor;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.task.TaskExecutionProperties.Pool;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.task.TaskExecutorBuilder;
import org.springframework.boot.task.TaskExecutorCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.task.TaskDecorator;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;@ConditionalOnClass({ThreadPoolTaskExecutor.class})
@Configuration
@EnableConfigurationProperties({TaskExecutionProperties.class})
public class TaskExecutionAutoConfiguration {public static final String APPLICATION_TASK_EXECUTOR_BEAN_NAME = "applicationTaskExecutor";private final TaskExecutionProperties properties;private final ObjectProvider<TaskExecutorCustomizer> taskExecutorCustomizers;private final ObjectProvider<TaskDecorator> taskDecorator;public TaskExecutionAutoConfiguration(TaskExecutionProperties properties, ObjectProvider<TaskExecutorCustomizer> taskExecutorCustomizers, ObjectProvider<TaskDecorator> taskDecorator) {this.properties = properties;this.taskExecutorCustomizers = taskExecutorCustomizers;this.taskDecorator = taskDecorator;}@Bean@ConditionalOnMissingBeanpublic TaskExecutorBuilder taskExecutorBuilder() {Pool pool = this.properties.getPool();TaskExecutorBuilder builder = new TaskExecutorBuilder();builder = builder.queueCapacity(pool.getQueueCapacity());builder = builder.corePoolSize(pool.getCoreSize());builder = builder.maxPoolSize(pool.getMaxSize());builder = builder.allowCoreThreadTimeOut(pool.isAllowCoreThreadTimeout());builder = builder.keepAlive(pool.getKeepAlive());builder = builder.threadNamePrefix(this.properties.getThreadNamePrefix());builder = builder.customizers(this.taskExecutorCustomizers);builder = builder.taskDecorator((TaskDecorator)this.taskDecorator.getIfUnique());return builder;}//实例化 ThreadPoolTaskExecutor @Lazy@Bean(name = {"applicationTaskExecutor", "taskExecutor"})@ConditionalOnMissingBean({Executor.class})public ThreadPoolTaskExecutor applicationTaskExecutor(TaskExecutorBuilder builder) {return builder.build();}
}

默认配置 TaskExecutionProperties ：

@ConfigurationProperties("spring.task.execution")
public class TaskExecutionProperties {private final TaskExecutionProperties.Pool pool = new TaskExecutionProperties.Pool();private String threadNamePrefix = "task-";public TaskExecutionProperties() {}public TaskExecutionProperties.Pool getPool() {return this.pool;}public String getThreadNamePrefix() {return this.threadNamePrefix;}public void setThreadNamePrefix(String threadNamePrefix) {this.threadNamePrefix = threadNamePrefix;}public static class Pool {private int queueCapacity = 2147483647;private int coreSize = 8;private int maxSize = 2147483647;private boolean allowCoreThreadTimeout = true;private Duration keepAlive = Duration.ofSeconds(60L);public Pool() {}public int getQueueCapacity() {return this.queueCapacity;}public void setQueueCapacity(int queueCapacity) {this.queueCapacity = queueCapacity;}public int getCoreSize() {return this.coreSize;}public void setCoreSize(int coreSize) {this.coreSize = coreSize;}public int getMaxSize() {return this.maxSize;}public void setMaxSize(int maxSize) {this.maxSize = maxSize;}public boolean isAllowCoreThreadTimeout() {return this.allowCoreThreadTimeout;}public void setAllowCoreThreadTimeout(boolean allowCoreThreadTimeout) {this.allowCoreThreadTimeout = allowCoreThreadTimeout;}public Duration getKeepAlive() {return this.keepAlive;}public void setKeepAlive(Duration keepAlive) {this.keepAlive = keepAlive;}}
}