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接口类型的对象，该对象内部包含了任务执行结果信息。

<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<> submit(Runnable task)
复制代码

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

任务

实现Runnable或Callable都可被任务执行框架执行，它们之间的区别是实现Runnable的任务执行完成后不会返回结果，而实现Callable接口的任务能够返回结果。

返回结果为null的封装（没有什么意义）

public static Callable<Object> callable(Runnable task) 复制代码

拥有返回结果的封装

public static <T> Callable<T> callable(Runnable task, T result) 复制代码

ThreadPoolExecutor

上一篇讲过ThreadPoolExecutor构造函数中的参数含义，这里再也不赘述，若是忘记了或者还没阅读过，能够先去了解一下，两篇文章是衔接的：juejin.im/post/5e9c45…

主要参数有4个：

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

FixedThreadPool详解

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

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. corePoolSize和maximumPoolSize都设置为用户传入的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内存溢出示例（我的理解，若是有错，欢迎指出！）：

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。本质上是fixedThreadPool的corePoolSize和maximumPoolSize都设置为1，此外无其它区别。

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
复制代码

参数解释

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

execute()执行流程

1. 若是线程池尚未预热，第一个任务来的时候就建立Worker线程来执行任务
2. 没法建立新线程时放入阻塞队列中
3. 惟一的工做者线程不断地执行阻塞队列中的任务

CachedThreadPool详解

根据须要建立新线程的线程池

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有哪几种状态？有哪些核心方法？调用方法会致使状态如何变化？（灵魂三问）

若是这篇文章能给你带来一点点的小收获，你的一个小小的点赞我会开心一成天！感谢你的阅读！