线程池:ThreadPoolExecutor和ThreadPoolTaskExecutor简述

JDK一般通过ThreadPoolExecutor方式创建线程池而不用Executors,原因是其构造方法参数限制较多,运行规则十分清晰,可以根据业务需求创建合适的线程池,而Executors返回的线程池对象的弊端,网上搜到比如FixedThreadPool和SingleThreadPool,允许的请求队列长度为 Integer.MAX_VALUE,可能会堆积大量的请求,从而导致 OOM;CachedThreadPool和ScheduledThreadPool,允许的创建线程数量为Integer.MAX_VALUE,可能会创建大量的线程,从而导致OOM

整个ThreadPoolExecutor最关键的就是构造方法的几个参数的理解

 * @param corePoolSize 核心线程池中最大线程数
* @param maximumPoolSize 线程池中的最大线程数
* @param keepAliveTime 空闲线程的存活时间
* @param unit 存活时间的单位
* @param workQueue 任务队列,保存已经提交但是还没有执行的任务
* @param threadFactory 线程工厂
* @param handler 拒绝策略
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue}
* or {@code threadFactory} or {@code handler} is null
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)

至于execute的执行流程,网上到处都能搜到,也十分详细,一开始以为queue的size有max约束,导致理解有点误解

我这里按自己的理解也画了一份完整的

NewImage

 

简单总结下重点

1、当前执行的线程数小于core上限,直接新建工作线程执行任务
2、当前执行的线程数大于等于core上限,workQueue没满,任务入队,如果队列没满,等待core线程结束上一轮,复用core线程执行队列任务
3、当workQueue满了,当前执行的线程数小于max上限,会在max-core的数量区间中继续创建线程执行任务
4、当前执行线程大于等于max上限,任务被拒绝
5、至于workQueue里任务的执行,复用当前线程池中运行但是没有执行任务的线程

下面以一个还算合适的例子说明下这个流程

任务类

public class Task implements Runnable {

private int no;
private int size;

public Task(int no, int size) {
this.no = no;
this.size = size;
}

@Override
public void run() {
try {
System.out.println("执行中, Task: " + no + ", Thread: " + Thread.currentThread().getName() + ", queue.size: " + size);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

执行,core为2,max为5,队列长度8,重写了拒绝handler方法,直接打印一条记录;任务执行里也直接打印一条记录,然后sleep为了得到相互之间的顺序

理论上结果应该是Task0-1直接创建线程执行,Task2-9直接入队列,Task10-12创建线程执行,Task13-14被拒绝,最后复用线程池中的线程执行Task2-9;这种结果预期的原因是加了sleep,不加sleep可能都不会有拒绝信息,因为Task很快就执行完了

public class Main {

private static ExecutorService pool;

public static void main(String[] args) {
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(8);

pool = new ThreadPoolExecutor(
2,
5,
1000,
TimeUnit.MILLISECONDS,
queue,
Executors.defaultThreadFactory(),
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println("拒绝中, " + r.toString() + ", queue.size: " + queue.size());
}
});

for(int i = 0; i < 15; i++) {
pool.execute(new Task(i, queue.size()));
}
pool.shutdown();
}
}

执行结果如下

执行中, Task: 0, Thread: pool-1-thread-1, queue.size: 0
执行中, Task: 1, Thread: pool-1-thread-2, queue.size: 0
执行中, Task: 10, Thread: pool-1-thread-3, queue.size: 8
执行中, Task: 11, Thread: pool-1-thread-4, queue.size: 8
执行中, Task: 12, Thread: pool-1-thread-5, queue.size: 8
拒绝中, com.lihuia.code.main.Task@6e8cf4c6, queue.size: 8
拒绝中, com.lihuia.code.main.Task@12edcd21, queue.size: 8
执行中, Task: 2, Thread: pool-1-thread-3, queue.size: 0
执行中, Task: 6, Thread: pool-1-thread-2, queue.size: 4
执行中, Task: 5, Thread: pool-1-thread-5, queue.size: 3
执行中, Task: 4, Thread: pool-1-thread-1, queue.size: 2
执行中, Task: 3, Thread: pool-1-thread-4, queue.size: 1
执行中, Task: 7, Thread: pool-1-thread-3, queue.size: 5
执行中, Task: 9, Thread: pool-1-thread-4, queue.size: 7
执行中, Task: 8, Thread: pool-1-thread-1, queue.size: 6

这个结果其实有很多可以注意的点

1、Task0-1最先起了两个线程,因为queue都是空
2、Task2-9入队列,因为虽然程序里它们是最后执行打印,但是queue.size当参数传进去的,任务开始执行的时候计算,而且从队列长度++可以看出,Task2-9来一个,进队列一个
3、队列塞满了,maxPool为5,因此还可以新建3个工作线程执行任务,因此Task10-12也开始执行,这样maxPool线程池就占满了
4、maxPool满了,因此Task13-14就会被拒绝,打印,可以看到队列一直是满的
5、由于sleep的作用,此时线程池里的线程都已经执行完了上面的Task,最后队列里的Task2-9就直接复用当前运行的线程来完成执行,可以根据pool-1-thread-x这个字段看出来

除了这些内容,还有线程池生命周期管理,阻塞队列,拒绝策略,任务调度管理,worker管理等,网上都能搜到比较详细的内容

 

Spring里的ThreadPoolTaskExecutor实际上就是ThreadPoolExecutor封装了一层,将它和ThreadPoolExecutor和相关属性封装起来,然后通过暴露这些属性,通过BEAN注入到容器当中,依赖注入管理起来

可以先看下ThreadPoolTaskExecutor的定义

public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport implements AsyncListenableTaskExecutor, SchedulingTaskExecutor {
private final Object poolSizeMonitor = new Object();
private int corePoolSize = 1;
private int maxPoolSize = 2147483647;
private int keepAliveSeconds = 60;
private int queueCapacity = 2147483647;
private boolean allowCoreThreadTimeOut = false;
@Nullable
private TaskDecorator taskDecorator;
@Nullable
private ThreadPoolExecutor threadPoolExecutor;
private final Map<Runnable, Object> decoratedTaskMap;

很明显,字段和ThreadPoolExecutor一模一样,因此很容易进行Bean注入

比如这里通过Java Config的方式,XML也是类似,配置类如下,和上面例子数据一致好了

@Configuration
public class TaskExecutorConfig {

@Bean
public ThreadPoolTaskExecutor taskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(2);
executor.setMaxPoolSize(5);
executor.setQueueCapacity(8);
executor.setKeepAliveSeconds(1);
executor.setThreadNamePrefix("task-thread-");
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(5);
executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println("拒绝中: " + r.toString());
}
});
executor.initialize();
return executor;
}
}

这里多注入了两个额外的属性setWaitForTasksToCompleteOnShutdown和setAwaitTerminationSeconds,这两个属性是关闭线程池的时候触发的,setWaitForTasksToCompleteOnShutdown会等待所有线程执行完,setAwaitTerminationSeconds指定具体等待的时间,可是ThreadPoolExecutor在shutdown的时候,并不需要指定这些,原因是ThreadPoolTaskExecutor在shutdown的时候,修改了逻辑,并不是直接调用ThreadPoolExecutor的shutdown,源码如下

路径如下:org.springframework.scheduling.concurrent.ExecutorConfigurationSupport#shutdown

public void shutdown() {
if (this.logger.isDebugEnabled()) {
this.logger.debug("Shutting down ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));
}

if (this.executor != null) {
if (this.waitForTasksToCompleteOnShutdown) {
this.executor.shutdown();
} else {
Iterator var1 = this.executor.shutdownNow().iterator();

while(var1.hasNext()) {
Runnable remainingTask = (Runnable)var1.next();
this.cancelRemainingTask(remainingTask);
}
}

this.awaitTerminationIfNecessary(this.executor);
}

}

这里的业务逻辑是:

如果waitForTasksToCompleteOnShutdown为true,才会调用ThreadPoolExecutor的shutdown()

如果waitForTasksToCompleteOnShutdown为false,调用的是ThreadPoolExecutor的shutdownNow(),结束为完成的任务

而我现在需要最终执行完阻塞队列里的任务,因此需要添加这两个属性

最后再看下executor.initialize()方法,最终调用的是org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor#initializeExecutor

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;
}

有了前面ThreadPoolExecutor的了解,这里实际上就是将各个属性来初始化ThreadPoolTaskExecutor里的ThreadPoolExecutor,这样就完成了这个线程池的依赖注入

测试类保持和前面例子一样的调用,执行15个任务,Task类和上面一样,只不过没打印queue的size

@ContextConfiguration(classes = {TaskExecutorConfig.class})
public class TaskTest extends AbstractTestNGSpringContextTests {

@Autowired
private ThreadPoolTaskExecutor taskExecutor;

@Test
public void testTask() {
for (int i = 0; i < 15; i++) {
taskExecutor.execute(new Task(i));
}
taskExecutor.shutdown();
}
}

这里调用的shutdown()方法,并不是JDK里的,而是Spring里的,上面已经说明了,由于加了两个属性,因此这里最终调用的还是JDK里的shutdown()方法

再来看一下执行结果

执行中, Task: 0, Thread: task-thread-1
执行中, Task: 1, Thread: task-thread-2
执行中, Task: 10, Thread: task-thread-3
执行中, Task: 11, Thread: task-thread-4
执行中, Task: 12, Thread: task-thread-5
拒绝中: com.lihuia.code.pool.Task@2d0399f4
拒绝中: com.lihuia.code.pool.Task@14dd7b39
执行中, Task: 2, Thread: task-thread-1
执行中, Task: 3, Thread: task-thread-2
执行中, Task: 4, Thread: task-thread-3
执行中, Task: 5, Thread: task-thread-4
执行中, Task: 6, Thread: task-thread-5
执行中, Task: 7, Thread: task-thread-1
执行中, Task: 8, Thread: task-thread-2
执行中, Task: 9, Thread: task-thread-4

和前面的分析也一致

 

最后看下Spring里的异步线程池,基于上面ThreadPoolExecutor配置的基础上,只需要开启异步操作,然后注解类或者方法,指定线程池就可以完成异步线程池操作

具体首先还是配置线程池,还是用的spring的ThreadPoolTaskExecutor,同时添加@EnableAsync注解开启异步功能,当然可以放在启动类上

@Slf4j
@EnableAsync
@Configuration
public class TaskExecutorConfig {

@Bean
public ThreadPoolTaskExecutor taskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(2);
executor.setMaxPoolSize(5);
executor.setQueueCapacity(8);
executor.setKeepAliveSeconds(1);
executor.setThreadNamePrefix("task-thread-");
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(5);
executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
log.info("拒绝中: " + r.toString());
}
});
executor.initialize();
return executor;
}
}

异步执行的方法就随便写一个了,这里是一个无返回的方法

这里要注意的是,@Async注解异步执行的方法, 指定了线程池,也就是配置的Bean name,因此通过依赖查找就找到了配置的线程池,当然这里bean name也可以不指定,因为后面会显式地指定线程池配置类信息

@Slf4j
public class Hello {

@Async("taskExecutor")
public void sayHello(String s) {
log.info("Hello {}", s);
}
}

最后就是测试类,这里通过@ContextConfiguration显式注入需要用到的bean,然后调用异步方法后,sleep一下避免主线程直接结束

@Slf4j
@ContextConfiguration(classes = {TaskExecutorConfig.class, Hello.class})
public class TaskTest extends AbstractTestNGSpringContextTests {

@Autowired
private Hello hello;

private void sleep() {
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

@Test
public void testTask() {
hello.sayHello("lihuia.com");
sleep();
}
}

执行的结果是

[task-thread-1] INFO com.lihuia.code.pool.Hello - Hello lihuia.com

从线程的name可以看出,异步线程是来自配置的ThreadPoolTaskExecutor类中,这里有一点,假如将注解里TaskExecutorConfig.class去掉

[main] INFO com.lihuia.code.pool.Hello - Hello lihuia.com

也就是这会就没有用自定义配置的线程池,要注意的是如果没有自定义,spring会搜索TaskExecutor类型的bean或者名称为taskExecutor的Executor类型的bean,如果也不存在,就会用SimpleAsyncTaskExecutor,但这玩意根本就不用线程池,每次都新起一个新线程,因此不能用,容易OOM

通常调用异步方法,都是无返回值的执行,假如要获取返回值,就只有借助Future接口,它可以对具体任务的执行结果进行处理和查询,大致有三种功能

1、判断任务是否完成
2、中断任务
3、获取任务执行结果

比如要获取异步线程池执行任务的结果

@Slf4j
@Component
public class Task{

private void sleep(int t) {
try {
TimeUnit.SECONDS.sleep(t);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

@Async("taskExecutor")
public Future<String> doTask(String s) {
sleep(5);
return new AsyncResult<>("Hello " + s);
}
}

异步方法同样指定线程池添加@Async注解,方法返回一个泛型String,添加一个sleep 5秒

测试类

@Test
public void testFuture() throws InterruptedException, ExecutionException {
Future<String> future = task.doTask("lihuia.com");
log.info("Waiting for result ...");
String result = future.get();
log.info(result);
}

获取异步方法结果之前加一行日志,目的是为了确认获取返回结果这个过程

21:04:40.800 [main] INFO com.lihuia.pool.TaskTest - Waiting for result ...
21:04:45.812 [main] INFO com.lihuia.pool.TaskTest - Hello lihuia.com

从时间戳可以看到,结果的打印一直阻塞了5S,而这5S正好是doTask里sleep的,可见get()方法会一直阻塞,等到任务执行完毕才会返回

同样可以试试get(long timeout, TimeUnit unit)方法,指定时间来取返回结果

@Test
public void testFuture() throws InterruptedException, ExecutionException, TimeoutException {
Future<String> future = task.doTask("lihuia.com");
log.info("Waiting for result ...");
String result = future.get(4, TimeUnit.SECONDS);
log.info(result);
}

比如4S后来返回结果,可是doTask方法4S还并没有执行完,因此会抛出异常

java.util.concurrent.TimeoutException
	at java.util.concurrent.FutureTask.get(FutureTask.java:205)
	at com.lihuia.pool.TaskTest.testFuture(TaskTest.java:47)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:124)
	at org.testng.internal.MethodInvocationHelper$1.runTestMethod(MethodInvocationHelper.java:230)
	at org.springframework.test.context.testng.AbstractTestNGSpringContextTests.run(AbstractTestNGSpringContextTests.java:181)
	at org.testng.internal.MethodInvocationHelper.invokeHookable(MethodInvocationHelper.java:242)
	at org.testng.internal.Invoker.invokeMethod(Invoker.java:579)
	at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:719)
	at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:989)
	at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:125)
	at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:109)
	at org.testng.TestRunner.privateRun(TestRunner.java:648)
	at org.testng.TestRunner.run(TestRunner.java:505)
	at org.testng.SuiteRunner.runTest(SuiteRunner.java:455)
	at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:450)
	at org.testng.SuiteRunner.privateRun(SuiteRunner.java:415)
	at org.testng.SuiteRunner.run(SuiteRunner.java:364)
	at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:52)
	at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:84)
	at org.testng.TestNG.runSuitesSequentially(TestNG.java:1208)
	at org.testng.TestNG.runSuitesLocally(TestNG.java:1137)
	at org.testng.TestNG.runSuites(TestNG.java:1049)
	at org.testng.TestNG.run(TestNG.java:1017)
	at com.intellij.rt.testng.IDEARemoteTestNG.run(IDEARemoteTestNG.java:66)
	at com.intellij.rt.testng.RemoteTestNGStarter.main(RemoteTestNGStarter.java:109)

这样就会因为执行超时,而释放回到线程池,而不会一直阻塞占用资源

 

总结一下,核心还是要掌握ThreadPoolExecutor所有参数的含义,包括每个参数的所有选项的区别,ThreadPoolTaskExecutor会进一步封装,然后通过Spring依赖注入更方便用户进行线程池属性配置,异步线程池在线程池基础上管理线程进行任务执行的异步调用

发表评论