电子商务网站规划与建设论文,建设工程网站教程,个人养老保险怎么查询,硬件开发方案JavaSE-线程池#xff08;5#xff09;- 建议使用的方式
虽然JDK Executors 工具类提供了默认的创建线程池的方法#xff0c;但一般建议自定义线程池参数#xff0c;下面是阿里巴巴开发手册给出的理由#xff1a; 另外Spring也提供了线程池的实现#xff0c;比如 Thread…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.ThreadPoolExecutor63e2203c[Running, pool size 4, active threads 4, queued tasks 2, completed tasks 0]] did not accept task: com.hs.example.base.multithread.threadpool.ThreadPoolTaskExecutorTest$MyTask3224f60bat 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$MyTask3224f60b rejected from java.util.concurrent.ThreadPoolExecutor63e2203c[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.ThreadPoolExecutor63e2203c[Running, pool size 4, active threads 4, queued tasks 2, completed tasks 0]] did not accept task: com.hs.example.base.multithread.threadpool.ThreadPoolTaskExecutorTest$MyTask4bbfb90aat 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$MyTask4bbfb90a rejected from java.util.concurrent.ThreadPoolExecutor63e2203c[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] 4ThreadPoolTaskExecutor 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) {BlockingQueueRunnable 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 ObjectProviderTaskExecutorCustomizer taskExecutorCustomizers;private final ObjectProviderTaskDecorator taskDecorator;public TaskExecutionAutoConfiguration(TaskExecutionProperties properties, ObjectProviderTaskExecutorCustomizer taskExecutorCustomizers, ObjectProviderTaskDecorator taskDecorator) {this.properties properties;this.taskExecutorCustomizers taskExecutorCustomizers;this.taskDecorator taskDecorator;}BeanConditionalOnMissingBeanpublic 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 LazyBean(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;}}
}
