Java并发篇-Executor框架全面扫盲

Java并发篇-全面解析Executor框架

Executor大家族成员

成员分为四个部分:任务、任务执行、任务执行结果以及任务执行工具类

任务:实现Callable接口或Runnable接口

任务执行部分:ThreadPoolExecutor以及ScheduledThreadPoolExecutor

任务执行结果:Future接口以及FutureTask实现类

任务执行工厂类:Executors

任务执行框架

ThreadPoolExecutor(核心)

ThreadPoolExecutor通常使用Executors进行创建,其内部含有三种线程池:

  1. FixedThreadPool:含有固定线程数的线程池。

  2. SingleThreadExecutor单线程的线程池,需要保证任务顺序执行时采用。

  3. CachedThreadPool大小无界的线程池,只要需要线程就可以一直创建线程。

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor通常使用Executors进行创建,其内部含有两种线程池:

  1. ScheduledThreadPoolExecutor:含有固定线程数的定时任务线程池
  2. SingleThreadScheduledExecutor:只包含一个线程数的定时任务线程池,需要保证任务顺序执行时采用。

任务执行结果

提交任务给任务执行框架执行后,任务执行框架会返回一个Future接口类型的对象,该对象内部包含了任务执行结果信息。

1
2
3
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<> submit(Runnable task)

在JDK1.8返回的是FutureTask对象,但不一定返回的就是FutureTask对象,只保证返回Future接口。

任务

实现RunnableCallable均可被任务执行框架执行,它们之间的区别是实现Runnable的任务执行完成后不会返回结果,而实现Callable接口的任务可以返回结果。

返回结果为null的封装(没有什么意义)

1
public static Callable<Object> callable(Runnable task)

拥有返回结果的封装

1
public static <T> Callable<T> callable(Runnable task, T result)

ThreadPoolExecutor

上一篇讲过ThreadPoolExecutor构造函数中的参数含义,这里不再赘述,如果忘记了或者还没阅读过,可以先去了解一下,两篇文章是衔接的:https://juejin.im/post/5e9c45bb6fb9a03c854630a2

主要参数有4个:

  1. corePolSize:核心线程池大小
  2. maximumPoolSize:最大线程池大小
  3. BlockingQueue:存储未执行任务的阻塞队列
  4. RejectedExecutionHandler:任务无法被执行时的拒绝策略

FixedThreadPool详解

fixedThreadPool称为可重用固定线程数的线程池,下面是构造该线程池的方法源码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}

public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}

参数解读

  1. corePoolSizemaximumPoolSize都设置为用户传入的nThreads
  2. keepAliveTime设置为0L,代表只要空闲线程闲下来就马上被终止,如果设置为SECONDS > 0,则代表线程空闲下来后等待新任务的最长时间SECONDS
  3. TimeUnit代表keepAliveTime的单位
  4. workQueue设置阻塞队列

execute()执行流程

curThread代表核心线程池的当前线程数

  1. 如果 curThread < corPoolSize,那么创建新线程执行任务

  2. 如果 curThread = corPoolSize,那么将任务加入阻塞队列当中

  3. 线程池中的线程执行完任务后会空闲下来,然后会一直从阻塞队列中获取任务执行

使用无界阻塞队列LinkedBlockingQueue的后果:

  1. curThread = corPoolSize后,任务会一直加入到阻塞队列,严重时阻塞队列的任务数过多会造成GC内存溢出
  2. 使用无界队列,那么参数maximumPoolSize是一个无效参数。只有当任务无法加入到阻塞队列时满足curThread < maximumPoolSiize才会在线程池中创建新的线程执行该任务。
  3. 由于第1点和第2点导致keepAliveTime参数无效
  4. 使用无界队列后,任务都可以加入到阻塞队列中,所以不会出现curThread > maximumPoolSize,也就不会出现任务拒绝执行的情况,不会调用任务拒绝方法。

GC内存溢出示例(个人理解,如果有错,欢迎指出!):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class Task implements Runnable {    
private String name;
public Task(String name) {
this.name = name;
}
@Override
public void run() {
System.out.println(name);
}
}
public class FixedThreadPoolTest {
private static final ExecutorService fixThreadPool = Executors.newFixedThreadPool(10);
private static int i = 0;

public static void main(String[] args) {
// 不停地往线程池中提交任务
for (;i < 100000;i++) {
fixThreadPool.execute(new Task("任务" + i));
}
fixThreadPool.shutdown();
}
}

设置堆内存参数:-Xms2m -Xmx2m

抛出OOM异常,原因在于阻塞队列中的任务数过多。

SingleThreadPool详解

使用单个Worker线程的Executor。本质上是fixedThreadPoolcorePoolSizemaximumPoolSize都设置为1,此外无其它区别。

1
2
3
4
5
6
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}

参数解释

  1. corePoolSizeMaximumPoolSize的参数都设置为1,其它参数含义和默认值都和fixedThreadPool相同。
  2. 无界阻塞队列对线程池的影响和fixedThreadPool相同。

execute()执行流程

  1. 如果线程池还没有预热,第一个任务来的时候就创建Worker线程来执行任务
  2. 无法创建新线程时放入阻塞队列中
  3. 唯一的工作者线程不断地执行阻塞队列中的任务

CachedThreadPool详解

根据需要创建新线程的线程池

1
2
3
4
5
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}

参数解释

  1. corePoolSize设置为0,即corePool为空
  2. maximumPoolSize设置为Integer.MAX_VALUE,所以maximumPool是几乎无界的,可容纳足够多的线程
  3. keepAliveTime设置为60L,每个空闲线程可以等待新任务的时间是60秒
  4. TimeUnit单位设置为SECONDS代表以秒为单位计算
  5. 阻塞队列采用无界的SynchronousQueue,每一次的offer操作都要对应一个poll操作

execute()执行流程

  1. 先把任务offer进阻塞队列中
  2. 如果线程池有空闲的线程则从阻塞队列中poll出任务执行,否则执行第3步
  3. 创建新的线程,并从队列中poll出任务执行。

ScheduledThreadPoolExecutor

由于JDK1.6与JDK1.8中延时队列的实现不同,所以在阅读本小节时不要死记硬背其实现,掌握其思想。图的组件均会用中文表示,对思想的掌握有帮助,实现的思想是一致的。

该类用于指定特定的延迟时间后运行任务,或者定期执行任务Timer是单线程的,该类是多线程的,而且该类功能更加强大。

运行机制

ScheduledThreadPoolExecutor将任务提交到延时队列当中存储,当任务到达执行时间时,Worker工作线程会拉取任务进行执行,空闲的线程会一直阻塞,不断尝试获取延时队列中的任务。

ScheduledThreadPoolExecutor本质上是ThreadPoolExecutor的改造:

  1. 使用延时队列作为任务队列
  2. 获取任务的方式不同,空闲线程将一直阻塞获取任务
  3. 执行周期任务后,增加了额外的处理操作

实现方法

调用线程把待调度的任务(ScheduledFutureTask)放入一个延时队列当中,ScheduledFutureTask的主要成员变量有三个:

  1. time:该任务要被执行的具体时间
  2. sequenceNumber:该任务被添加到ScheduledThreadPoolExecutor的序号,标记任务添加到队列的时刻
  3. period:任务执行的间隔周期

延时队列中采用优先级队列进行排序。排序时,time越小排在越前面,time相同时,sequenceNumber越小排在越前面。

ScheduledThreadPoolExecutor执行任务的流程如下图:

  1. 获取已到期的任务,到期的条件是time >= 当前时间
  2. 线程1执行该任务
  3. 线程1修改该任务的time变量为下次要被执行的时间
  4. 把该任务放回到延时队列当中

FutureTask

表示异步计算的结果,实现了Runnable、Future接口。

FutureTask的几种状态

  1. 未启动。FutureTask没有执行run()方法之前处于未启动状态
  2. 已启动。FutureTask.run()方法被执行的过程中
  3. 已完成。FutureTask.run()方法执行完后正常结束、或被取消、或执行方法时抛出异常而异常结束

FutureTask的状态转换图如下所示

FutureTask的核心方法

FutureTask.run()

当任务处于未启动状态时,调用FutureTask.run()会使任务执行。

FutureTask.get()

FutureTask处于未启动或已启动状态时,执行FutureTask.get()会导致线程阻塞

当FutureTask处于已完成状态时,调用FutureTask.get()会立即返回结果或者抛出异常

FutureTask.cancel()

当任务处于未启动状态时,调用FutureTask.cancel()会导致此任务永远不会被执行

当任务处于已启动状态时,执行FutureTask.cancel(true)会中断任务,执行FutureTask.cancel(false)方法将不会对正在执行此任务的线程产生任何影响;

当FutureTask处于完成状态时,执行FutureTask.cancel(…)方法将返回false

get()和cancel()的状态转换图:

FutureTask的使用

当一个线程需要等待另一个线程把某个任务执行完后才能继续执行,可以使用FutureTask。这种用法并不是很常见,只要知道有这种用法就可以了。

总结

这一章主要和上一章的线程池有很大联系,阅读到这里非常不容易,收获也应该很多,留下几道面试题来检验自己是否真的掌握了这些内容吧。

  1. Executor框架主要有哪些成员?
  2. 任务执行框架主要有哪几种线程池?
  3. 线程池有哪些参数?它们的含义是什么?
  4. 每种线程池的工作流程是如何的?简单介绍一下(越详细越好,使用了什么队列、线程池大小······)
  5. 任务执行完成后返回的结果如何接收?
  6. FutureTask有哪几种状态?有哪些核心方法?调用方法会导致状态如何变化?(灵魂三问)

如果这篇文章能给你带来一点点的小收获,你的一个小小的点赞我会开心一整天!感谢你的阅读!

0%