转载

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

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

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

任务执行部分： ThreadPoolExecutor 以及 ScheduledThreadPoolExecutor

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

任务执行工厂类： Executors

任务执行框架

ThreadPoolExecutor（核心）

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

FixedThreadPool ：含有 固定线程数 的线程池。
SingleThreadExecutor ： 单线程 的线程池，需要保证任务顺序执行时采用。
CachedThreadPool ： 大小无界 的线程池，只要需要线程就可以一直创建线程。

ScheduledThreadPoolExecutor

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

ScheduledThreadPoolExecutor ：含有 固定线程数 的定时任务线程池
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个：

corePolSize
maximumPoolSize
BlockingQueue
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);
}
复制代码

参数解读

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

execute()执行流程

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

如果 curThread < corPoolSize ，那么创建新线程执行任务
如果 curThread = corPoolSize ，那么将任务加入阻塞队列当中
线程池中的线程执行完任务后会空闲下来，然后会 一直从阻塞队列中获取任务执行

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

curThread = corPoolSize 后，任务会一直加入到阻塞队列， 严重时阻塞队列的任务数过多会造成GC内存溢出 。
使用无界队列，那么参数 maximumPoolSize 是一个无效参数。只有 当任务无法加入到阻塞队列时 满足 curThread < maximumPoolSiize 才会在线程池中创建新的线程执行该任务。
由于第1点和第2点导致 keepAliveTime 参数无效
使用无界队列后，任务都可以加入到阻塞队列中，所以不会出现 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>()));
}
复制代码

参数解释

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

execute()执行流程

Worker

CachedThreadPool详解

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

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}
复制代码