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;
}
@Override
public 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@3224f60b
at 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@4bbfb90a
at 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;
}
@Override
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread() + " " + i);
countDownLatch.countDown();
}
}
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
@Test
public 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
@ConditionalOnMissingBean
public 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;
}
}
}