一、概述
Spring通过ThreadPoolTaskExecutor实现线程池技术,它是使用jdk中的Java.util.concurrent.ThreadPoolExecutor进行实现。
1.1、xml方式配置
<!-- Spring线程池 --> <bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"> <!-- 核心线程数 --> <property name="corePoolSize" value="5" /> <!-- 线程池维护线程的最大数量 --> <property name="maxPoolSize" value="10" /> <!-- 允许的空闲时间, 默认60秒 --> <property name="keepAliveSeconds" value="60" /> <!-- 任务队列 --> <property name="queueCapacity" value="50" /> <!-- 线程超过空闲时间限制,均会退出直到线程数量为0 --> <property name="allowCoreThreadTimeOut" value="true"/> <!-- 对拒绝task的处理策略 --> <property name="rejectedExecutionHandler"> <bean class="java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy" /> </property> </bean>
java代码
//帮我们创建容器 @RunWith(SpringJUnit4ClassRunner.class) //指定创建容器时使用哪个配置文件 @ContextConfiguration("classpath:applicationContext.xml") public class SpringThreadPoolExecutorTest { @Autowired private ThreadPoolTaskExecutor taskExecutor; @Test public void newFixedThreadPoolTestSubmit() throws Exception { List<Future<?>> list=new ArrayList<>(); for (int i = 0; i < 20; i++) { Future<?> submit = taskExecutor.submit(new DataTask("ssss" + i)); list.add(submit); } for (Future<?> future : list) { future.get(); } } class DataTask implements Runnable { private String a; public DataTask(String a) { this.a = a; } @Override public void run() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("-------"+ LocalDateTime.now()+a); } } }
输出
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
-------2019-08-06T21:23:18.105ssss0 -------2019-08-06T21:23:18.105ssss3 -------2019-08-06T21:23:18.105ssss1 -------2019-08-06T21:23:18.105ssss2 -------2019-08-06T21:23:18.105ssss4 -------2019-08-06T21:23:19.110ssss8 -------2019-08-06T21:23:19.110ssss6 -------2019-08-06T21:23:19.110ssss5 -------2019-08-06T21:23:19.110ssss7 -------2019-08-06T21:23:19.110ssss9 -------2019-08-06T21:23:20.113ssss10 -------2019-08-06T21:23:20.113ssss12 -------2019-08-06T21:23:20.113ssss13 -------2019-08-06T21:23:20.113ssss11 -------2019-08-06T21:23:20.113ssss14 -------2019-08-06T21:23:21.117ssss18 -------2019-08-06T21:23:21.117ssss16 -------2019-08-06T21:23:21.117ssss17 -------2019-08-06T21:23:21.117ssss15 -------2019-08-06T21:23:21.117ssss19
可以看到每5个一组一组的运行,
过程:corePoolSize→queueCapacity→开启maxPoolSize
当一个新任务来临时: 1)如果此时线程池中的数量小于corePoolSize,即使线程池中的线程都处于空闲状态,也要创建新的线程来处理被添加的任务; 2)如果此时线程池中的数量等于corePoolSize,但是缓冲队列workQueue未满,那么任务被放入缓冲队列; 3)如果此时线程池中的数量大于corePoolSize,缓冲队列workQueue满,并且线程池中的数量小于maxPoolSize,建新的线程来处理被添加的任务; 4)如果此时线程池中的数量大于corePoolSize,缓冲队列workQueue满,并且线程池中的数量等于maxPoolSize,那么通过handler所指定的策略来处理此任务; 5)当线程池中的线程数量大于corePoolSize时,如果某线程空闲时间超过keepAliveTime,线程将被终止,如果allowCoreThreadTimeOut为false,则线程数量维持在corePoolSize, 如果为true,则线程数量可最低降至0;
1.2、使用task:executor方式
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:task="http://www.springframework.org/schema/task" xsi:schemaLocation= "http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd"> <!-- 包路径扫描 --> <context:component-scan base-package="spring.task"/> <!-- 定义线程池 --> <task:executor id="executor" pool-size="5" queue-capacity="10" rejection-policy="DISCARD_OLDEST"/> </beans>
java 测试
@Component public class AsyncTask { @Autowired private ThreadPoolTaskExecutor executor; @PostConstruct public void testAsycTask() { for (int i = 0; i < 10; i++) { executor.execute(new Runnable() { public void run() { asyncTask(); } }); } } private void asyncTask(){ System.out.println( "---" + Thread.currentThread().getName()); } }
二、池子如何配置参数
2.1、默认值
corePoolSize=1
queueCapacity=Integer.MAX_VALUE
maxPoolSize=Integer.MAX_VALUE
keepAliveTime=60s
allowCoreThreadTimeout=false
rejectedExecutionHandler=AbortPolicy()
2.2、如何设置
需要根据几个值来决定
- tasks :每秒的任务数,假设为100~1000
- taskcost:每个任务花费时间,假设为0.1s
- responsetime:系统允许容忍的最大响应时间,假设为2s
配置计算
- corePoolSize = 每秒需要多少个线程处理?
* threadcount = tasks/(1/taskcost) =tasks*taskcout = (100~1000)*0.1 = 10~100 个线程。corePoolSize设置应该大于10
* 根据8020原则,如果80%的每秒任务数小于200,那么corePoolSize设置为20即可
- queueCapacity = (coreSizePool/taskcost)*responsetime
* 计算可得 queueCapacity = 20/0.1*2 = 400。意思是队列里的线程可以等待4s,超过了的需要新开线程来执行
* 切记不能写:LinkedBlockingQueue queue = new LinkedBlockingQueue(); 实际是设置为Integer.MAX_VALUE,这样队列会很大,线程数只会保持在corePoolSize大小,当任务陡增时,不能新开线程来执行,响应时间会随之陡增。
- maxPoolSize = (max(tasks)- queueCapacity)/(1/taskcost)
* 计算可得 maxPoolSize = (1000-400)/10 = 60
* (最大任务数-队列容量)/每个线程每秒处理能力 = 最大线程数
- rejectedExecutionHandler:根据具体情况来决定,任务不重要可丢弃,任务重要则要利用一些缓冲机制来处理
- keepAliveTime和allowCoreThreadTimeout采用默认通常能满足
以上都是理想值,实际情况下要根据机器性能来决定。如果在未达到最大线程数的情况机器cpu load已经满了,则需要通过升级硬件和优化代码,降低taskcost来处理。
xml