Executor框架（线程池的实现）

一、为什么要使用线程池？

降低资源消耗：通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
提高响应速度：当任务到达时，任务可以不需要的等到线程创建就能立即执行。
提高线程的可管理性：线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

二、Executor框架（线程池的实现）

Java的线程既是工作单元，也是执行机制。从JDK 5开始，把工作单元与执行机制分离开来。工作单元包括Runnable和Callable，而执行机制由Executor框架提供

Executor框架的两级调度模型
在HotSpot VM的线程模型中，Java线程被一对一映射为本地操作系统线程。Java线程启动时会创建一个本地操作系统线程；当Java线程终止时，这个操作系统线程也会被回收。操作系统会调用所有线程并将他们分配给可用的CPU。可以将此种模式分为两层，在上层，Java多线程程序通常把应用程序分解为若干任务，然后使用用户级的调度器（Executor框架）将这些任务映射为固定数量的线程；在底层，操作系统内核将这些线程映射到硬件处理器上。
Executor 框架的类图

Executor接口：它是Executor框架的基础，它将任务的提交与任务的执行分离开来，无返回值；
ExecutorService接口：提供了 submit(Runnable)、submit(Callable) 两个方法，带返回值；
AbstractExecutorService：提供 ExecutorService 执行方法的默认实现。
ThreadPoolExecutor：线程池的核心类，根据参数设置的不同，可以创建出多种类型的线程池；
ScheduledExecutorService接口：一个特殊的 ExecutorService，提供了可安排在给定的延迟后运行或定期执行的命令。
ScheduledThreadPoolExecutor： ScheduledExecutorService接口的具体实现类；
Executor框架的结构
任务：也就是工作单元，包括被执行任务需要实现的接口：Runnable接口或者Callable接口；
任务的执行：也就是把任务分派给多个线程的执行机制，包括任务执行机制的核心接口Executor，以及继承自Executor的ExecutorService接口。Executor框架有两个关键类实现了ExecutorService接口：ThreadPoolExecutor 和 ScheduledThreadPoolExecutor、ForkJoinPool；
任务的异步计算结果：包括Future接口和实现Future接口的FutureTask类、ForkJoinTask类。

三、线程池的创建

1、使用ThreadPoolExecutor类来创建线程池

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,  long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

创建一个线程池的参数有：

1）corePoolSize（线程池核心线程）

线程池中会维护一个最小的线程数量，即使这些线程是空闲状态，他们也不会被销毁，除非设置了allowCoreThreadTimeOut。这里的最小线程数量即是corePoolSize。

2）maximumPoolSize（线程池最大线程数量）

一个任务被提交到线程池以后，首先会找有没有空闲存活线程，如果有则直接将任务交给这个空闲线程来执行，如果没有则会缓存到工作队列（后面会介绍）中，如果工作队列满了，才会创建一个新线程，然后从工作队列的头部取出一个任务交由新线程来处理，而将刚提交的任务放入工作队列尾部。线程池不会无限制的去创建新线程，它会有一个最大线程数量的限制，这个数量即由maximunPoolSize指定。

3）keepAliveTime 空闲线程存活时间

一个线程如果处于空闲状态，并且当前的线程数量大于corePoolSize，那么在指定时间后，这个空闲线程会被销毁，这里的指定时间由keepAliveTime来设定

4）unit 空闲线程存活时间单位

keepAliveTime的计量单位

5）workQueue 工作队列

新任务被提交后，会先进入到此工作队列中，任务调度时再从队列中取出任务。

jdk中提供了四种工作队列：
①ArrayBlockingQueue（基于数组的有界阻塞队列）
新任务进来后，会放到该队列的队尾，有界的数组可以防止资源耗尽问题。当线程池中线程数量达到corePoolSize后，再有新任务进来，则会将任务放入该队列的队尾，等待被调度。如果队列已经是满的，则创建一个新线程，如果线程数量已经达到maximumPoolSize，则会执行拒绝策略。
②LinkedBlockingQuene（基于链表的无界阻塞队列）
由于该队列的近似无界性，当线程池中线程数量达到corePoolSize后，再有新任务进来，会一直存入该队列，而不会去创建新线程直到maxPoolSize，因此使用该工作队列时，参数maximumPoolSize其实是不起作用的。
③SynchronousQuene（不缓存任务的阻塞队列）
生产者放入一个任务必须等到消费者取出这个任务。也就是说新任务进来时，不会缓存，而是直接被调度执行该任务，如果没有可用线程，则创建新线程，如果线程数量达到maximumPoolSize，则执行拒绝策略。
④PriorityBlockingQueue（具有优先级的无界阻塞队列）
优先级通过参数Comparator实现。

6）threadFactory 线程工厂

创建一个新线程时使用的工厂，可以用来设定线程名等。

7）handler 拒绝策略

当工作队列中的任务已到达最大限制，并且线程池中的线程数量也达到最大限制，这时如果有新任务提交进来，该如何处理呢。这里的拒绝策略，就是解决这个问题的。

jdk中提供了4中拒绝策略：
①CallerRunsPolicy
该策略下，在调用者线程中直接执行被拒绝任务的run方法，除非线程池已经shutdown，则直接抛弃任务。
②AbortPolicy
该策略下，直接丢弃任务，并抛出RejectedExecutionException异常。
③DiscardPolicy
该策略下，直接丢弃任务，什么都不做。
④DiscardOldestPolicy
该策略下，抛弃进入队列最早的那个任务，然后尝试把这次拒绝的任务放入队列

2、使用工厂类Executors来创建线程池

1）newFixedThreadPool 固定线程池

这种线程池的corePoolSize和maximumPoolSize都被设置为参数nThreads，表示线程池中所有线程均是核心线程，那么keepAliveTime参数就是无意义的，使用无界阻塞队列LinkedBlockingQueue，表明不会出现拒绝策略

public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
}public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory);
}

处理任务流程：
①当线程池中的线程数小于corePoolSize时，线程池会创建新线程去执行任务。
②在经过一段时间预热后，线程数达到了corePoolSize（因为maximumPoolSize与corePoolSize相同，此时也达到了最大线程数，以后不会再创建线程），开始将任务放入工作队列中。
③此后有新任务到达就向工作队列中放入（LinkedBlockingQueue无参构造方法，创建的队列容量是Integer.MAX_VALUE ,这种队列几乎不可能容量爆满，不会拒绝任务，拒绝策略不起作用），若有线程处于空闲状态则从工作队列中获取任务并执行。

2）newSingleThreadExecutor 单线程池

这类线程池的corePoolSize和maximumPoolSize都被设为1，线程池中最多有一个线程，同样地这里的keepAliveTime参数是无意义的，使用无界阻塞队列LinkedBlockingQueue，表明不会出现拒绝策略

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

处理任务流程：
①当线程池中的无任何线程时，线程池会创建一个线程去执行任务。
②当线程池中有一个线程时，开始将任务放入工作队列中。
③此后有新任务到达就向工作队列中放入（LinkedBlockingQueue无参构造方法，创建的队列容量是Integer.MAX_VALUE ,这种队列几乎不可能容量爆满，不会拒绝任务，拒绝策略不起作用），若线程池中的唯一线程处于空闲状态则从工作队列中取出任务并执行。

3）newCachedThreadPool 缓存线程池

这种线程池的corePoolSize为零、maximumPoolSize为Integer.MAX_VALUE，表明线程池中没有核心线程、所有线程均是非核心线程，且可允许的最大线程数是近乎无限大。keepAliveTime参数设为60,表明空闲线程最多等待60秒就被终止。
使用不缓存任务的阻塞队列SynchronousQueue，也就是说新任务进来时，不会缓存，而是直接被调度执行该任务，如果没有可用线程，则创建新线程，但可允许创建的最大线程数是无限大的。这意味着主线程提交任务的速度大于任务处理的速度，线程池就会创不断建新线程，这样可能导致创建的线程过多，系统资源被耗尽、程序崩溃。

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

处理任务流程：
①因为核心线程数为0，所以线程池在启动时核心线程池就已经满了。在主线程提交第一个任务时，线程池就要将尝试此任务入队，由于SynchronousQueue的特殊性，只有当此时空闲线程也正在出队,入队与出队两者恰好匹配时，主线程会把任务交给空闲线程去执行。否则将进入下一步。
②当线程池中无任何线程或无空闲线程时，将没有线程执行出队操作。此时线程池会创建建一个新线程执行任务
③上一步中创建的新线程在执行完的任务后，会调用SynchronousQueue.poll等待任务出队。这个空闲线程最多等待60秒时间，若主线程在60秒内提交了一个新任务，此空闲线程将获取到这个任务并执行。若等待60秒后，还没等到新任务到达，这个线程将被终止。

四、线程池的分析

源码分析

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();// 如果当前线程池的工作线程数量小于corePoolSize，则调用addWoker()添加工作线程，执行任务if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}// 判断线程池是否运行，将任务加入任务队列if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();// 再次判断线程池是否运行，如果未运行则拒绝执行这一任务if (! isRunning(recheck) && remove(command))// 执行任务的拒绝策略reject(command);else if (workerCountOf(recheck) == 0)// 线程池中如果已经没有了工作线程，需要重新建立一个工作线程去执行仍旧在任务队列中等待执行的任务。addWorker(null, false);}else if (!addWorker(command, false))// 建立工作线程失败，则执行任务的拒绝策略reject(command);
}

五、ScheduledThreadPoolExecutor（定时任务执行器）

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，并实现了表示定时执行器的接口ScheduledExecutorService。它主要用来在给定的延迟之后运行任务，或者定期执行任务.

1、构造方法创建

直接调用父类ThreadPoolExecutor的实现

public ScheduledThreadPoolExecutor(int corePoolSize) {super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,new DelayedWorkQueue());
}
public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory) {super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,new DelayedWorkQueue(), threadFactory);}
public ScheduledThreadPoolExecutor(int corePoolSize,RejectedExecutionHandler handler) {super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,new DelayedWorkQueue(), handler);
}
public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory,RejectedExecutionHandler handler) {super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,new DelayedWorkQueue(), threadFactory, handler);
}

2、工厂类Executors的静态方法创建

1）newSingleThreadScheduledExecutor
用于创建单个线程的定时任务执行器。它适用于按照固定顺序执行周期性的定时任务，且最多同时执行一个任务的情况。

 public static ScheduledExecutorService newSingleThreadScheduledExecutor() {return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1));
}
public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1, threadFactory));
}

2）newScheduledThreadPool
用于创建给定个数线程的定时任务执行器。它适用于需要多个线程执行周期任务，同时又要限制线程数、防止创建线程过多耗费资源的情况。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {return new ScheduledThreadPoolExecutor(corePoolSize);
}
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory) {return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

DelayedWorkQueue是ScheduledThreadPoolExecutor的一个静态内部类，它是一个无界队列，所以父类ThreadPoolExecutor中的maximumPoolSize、keepAliveTime这两个参数无意义、没有任何效果。

处理任务流程：
①当调用ScheduledThreadPoolExecutor的scheduleAtFixedRate()方法或者scheduleWithFixedDelay()方法时，会向的DelayedWorkQueue添加一个实现了RunnableScheduledFuture接口的ScheduledFutureTask类型任务。
②某空闲线程从DelayedWorkQueue中获取已到期的ScheduledFutureTask任务(到期任务是指ScheduledFutureTask的time大于等于当前时间)。
③此线程执行这个ScheduledFutureTask任务，此线程修改ScheduledFutureTask的time变量为下次将要被执行的时间。
④此线程把这个time被修改后的ScheduledFutureTask重新放回DelayedWorkQueue中。

六、向线程池提交任务，和线程池关闭

1、向线程池提交任务

1）使用execute提交的任务，但是execute方法没有返回值，所以无法判断任务知否被线程池执行成功。execute方法输入的任务是一个Runnable类的实例。

// 调用的是Executor接口的方法
void execute(Runnable command);

2）使用submit 方法来提交任务，它会返回一个future,那么我们可以通过这个future来判断任务是否执行成功，通过future的get方法来获取返回值，get方法会阻塞住直到任务完成，而使用get(long timeout, TimeUnit unit)方法则会阻塞一段时间后立即返回，这时有可能任务没有执行完。

// 调用的是ExecutorService接口的方法
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);

2、线程池的关闭

1）线程池的shutdown方法

shutdown的原理是只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程

2）线程池的shutdownNow方法

shutdownNow的原理是遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无法响应中断的任务可能永远无法终止。shutdownNow会首先将线程池的状态设置成STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表。

3）总结

只要调用了这两个关闭方法的其中一个，isShutdown方法就会返回true。当所有的任务都已关闭后,才表示线程池关闭成功，这时调用isTerminaed方法会返回true。至于我们应该调用哪一种方法来关闭线程池，应该由提交到线程池的任务特性决定，通常调用shutdown来关闭线程池，如果任务不一定要执行完，则可以调用shutdownNow。

七、线程池监控

ThreadPoolExecutor类提供了以下几个方法可以监控线程池的使用情况：
1）getActiveCount() ：线程池中正在执行任务的线程数量
2）getCompletedTaskCount() ：线程池已完成的任务数量，该值小于等于taskCount
3）getCorePoolSize() ：线程池的核心线程数量
4）getLargestPoolSize() ：线程池曾经创建过的最大线程数量。通过这个数据可以知道线程池是否满过，也就是达到了maximumPoolSize
5）getMaximumPoolSize() ：线程池的最大线程数量
6）getPoolSize() ：线程池当前的线程数量
7）getTaskCount() ：线程池已经执行的和未执行的任务总数
自定义线程池监控类实现ThreadPoolExecutor类，重写beforeExecute 、afterExecute 、terminated 方法

public class ThreadPoolMonitor extends ThreadPoolExecutor {private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPoolMonitor.class);/*** 保存任务开始执行的时间，当任务结束时，用任务结束时间减去开始时间计算任务执行时间*/private ConcurrentHashMap<String, Date> startTimes;/*** 线程池名称，一般以业务名称命名，方便区分*/private String poolName;/*** 调用父类的构造方法，并初始化HashMap和线程池名称** @param corePoolSize    线程池核心线程数* @param maximumPoolSize 线程池最大线程数* @param keepAliveTime   线程的最大空闲时间* @param unit            空闲时间的单位* @param workQueue       保存被提交任务的队列* @param poolName        线程池名称*/public ThreadPoolMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime,TimeUnit unit, BlockingQueue<Runnable> workQueue, String poolName) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), poolName);}/*** 调用父类的构造方法，并初始化HashMap和线程池名称** @param corePoolSize    线程池核心线程数* @param maximumPoolSize 线程池最大线程数* @param keepAliveTime   线程的最大空闲时间* @param unit            空闲时间的单位* @param workQueue       保存被提交任务的队列* @param threadFactory   线程工厂* @param poolName        线程池名称*/public ThreadPoolMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime,TimeUnit unit, BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory, String poolName) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);this.startTimes = new ConcurrentHashMap<>();this.poolName = poolName;}/*** 线程池延迟关闭时（等待线程池里的任务都执行完毕），统计线程池情况*/@Overridepublic void shutdown() {// 统计已执行任务、正在执行任务、未执行任务数量LOGGER.info("{} Going to shutdown. Executed tasks: {}, Running tasks: {}, Pending tasks: {}",this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());super.shutdown();}/*** 线程池立即关闭时，统计线程池情况*/@Overridepublic List<Runnable> shutdownNow() {// 统计已执行任务、正在执行任务、未执行任务数量LOGGER.info("{} Going to immediately shutdown. Executed tasks: {}, Running tasks: {}, Pending tasks: {}",this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());return super.shutdownNow();}/*** 任务执行之前，记录任务开始时间*/@Overrideprotected void beforeExecute(Thread t, Runnable r) {startTimes.put(String.valueOf(r.hashCode()), new Date());}/*** 任务执行之后，计算任务结束时间*/@Overrideprotected void afterExecute(Runnable r, Throwable t) {Date startDate = startTimes.remove(String.valueOf(r.hashCode()));Date finishDate = new Date();long diff = finishDate.getTime() - startDate.getTime();// 统计任务耗时、初始线程数、核心线程数、正在执行的任务数量、// 已完成任务数量、任务总数、队列里缓存的任务数量、池中存在的最大线程数、// 最大允许的线程数、线程空闲时间、线程池是否关闭、线程池是否终止LOGGER.info("{}-pool-monitor: " +"Duration: {} ms, PoolSize: {}, CorePoolSize: {}, Active: {}, " +"Completed: {}, Task: {}, Queue: {}, LargestPoolSize: {}, " +"MaximumPoolSize: {},  KeepAliveTime: {}, isShutdown: {}, isTerminated: {}",this.poolName,diff, this.getPoolSize(), this.getCorePoolSize(), this.getActiveCount(),this.getCompletedTaskCount(), this.getTaskCount(), this.getQueue().size(), this.getLargestPoolSize(),this.getMaximumPoolSize(), this.getKeepAliveTime(TimeUnit.MILLISECONDS), this.isShutdown(), this.isTerminated());}/*** 创建固定线程池，代码源于Executors.newFixedThreadPool方法，这里增加了poolName** @param nThreads 线程数量* @param poolName 线程池名称* @return ExecutorService对象*/public static ExecutorService newFixedThreadPool(int nThreads, String poolName) {return new ThreadPoolMonitor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), poolName);}/*** 创建缓存型线程池，代码源于Executors.newCachedThreadPool方法，这里增加了poolName** @param poolName 线程池名称* @return ExecutorService对象*/public static ExecutorService newCachedThreadPool(String poolName) {return new ThreadPoolMonitor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(), poolName);}/*** 生成线程池所用的线程，只是改写了线程池默认的线程工厂，传入线程池名称，便于问题追踪*/static class EventThreadFactory implements ThreadFactory {private static final AtomicInteger poolNumber = new AtomicInteger(1);private final ThreadGroup group;private final AtomicInteger threadNumber = new AtomicInteger(1);private final String namePrefix;/*** 初始化线程工厂** @param poolName 线程池名称*/EventThreadFactory(String poolName) {SecurityManager s = System.getSecurityManager();group = Objects.nonNull(s) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();namePrefix = poolName + "-pool-" + poolNumber.getAndIncrement() + "-thread-";}@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);if (t.isDaemon())t.setDaemon(false);if (t.getPriority() != Thread.NORM_PRIORITY)t.setPriority(Thread.NORM_PRIORITY);return t;}}
}

1）shutdown() ：线程池延迟关闭时（等待线程池里的任务都执行完毕），统计已执行任务、正在执行任务、未执行任务数量
2）shutdownNow() ：线程池立即关闭时，统计已执行任务、正在执行任务、未执行任务数量
3）beforeExecute(Thread t, Runnable r) ：任务执行之前，记录任务开始时间，startTimes这个HashMap以任务的hashCode为key，开始时间为值
4）afterExecute(Runnable r, Throwable t) ：任务执行之后，计算任务结束时间。统计任务耗时、初始线程数、核心线程数、正在执行的任务数量、已完成任务数量、任务总数、队列里缓存的任务数量、池中存在的最大线程数、最大允许的线程数、线程空闲时间、线程池是否关闭、线程池是否终止信息

注意事项

在 afterExecute 方法中需要注意，需要调用 ConcurrentHashMap 的 remove 方法移除并返回任务的开始时间信息，而不是调用 get 方法，因为在高并发情况下，线程池里要执行的任务很多，如果只获取值不移除的话，会使 ConcurrentHashMap 越来越大，引发内存泄漏或溢出问题。该行代码如下：
Date startDate = startTimes.remove(String.valueOf(r.hashCode()));
有了ThreadPoolMonitor类之后，我们可以通过 newFixedThreadPool(int nThreads, String poolName) 和 newCachedThreadPool(String poolName) 方法创建两个日常我们使用最多的线程池，跟默认的 Executors 里的方法不同的是，这里需要传入 poolName 参数，该参数主要是用来给线程池定义一个与业务相关并有具体意义的线程池名字，方便我们排查线上问题。
在生产环境中，谨慎调用 shutdown() 和 shutdownNow() 方法，因为调用这两个方法之后，线程池会被关闭，不再接收新的任务，如果有新任务提交到一个被关闭的线程池，会抛出 RejectedExecutionException 异常。其实在使用Spring等框架来管理类的生命周期的条件下，也没有必要调用这两个方法来关闭线程池，线程池的生命周期完全由该线程池所属的Spring管理的类决定。