Executors和ThreadPoolExecutor详解

概述

在《阿里巴巴java开发手册》中指出了线程资源必须通过线程池提供，不允许在应用中自行显示的创建线程，这样一方面是线程的创建更加规范，可以合理控制开辟线程的数量；另一方面线程的细节管理交给线程池处理，优化了资源的开销。而线程池不允许使用Executors去创建，而要通过ThreadPoolExecutor方式，这一方面是由于jdk中Executor框架虽然提供了如newFixedThreadPool()、newSingleThreadExecutor()、newCachedThreadPool()等创建线程池的方法，但都有其局限性，不够灵活；另外由于前面几种方法内部也是通过ThreadPoolExecutor方式实现，使用ThreadPoolExecutor有助于大家明确线程池的运行规则，创建符合自己的业务场景需要的线程池，避免资源耗尽的风险

1、线程池的七种创建方式

Executors 创建的六种

Executors.newFixedThreadPool：创建一个固定大小的线程池，可控制并发的线程数，超出的线程会在队列中等待；
Executors.newCachedThreadPool：创建一个可缓存的线程池，若线程数超过处理所需，缓存一段时间后会回收，若线程数不够，则新建线程；
Executors.newSingleThreadExecutor：创建单个线程数的线程池，它可以保证先进先出的执行顺序；
Executors.newScheduledThreadPool：创建一个可以执行延迟任务的线程池；
Executors.newSingleThreadScheduledExecutor：创建一个单线程的可以执行延迟任务的线程池；
Executors.newWorkStealingPool：创建一个抢占式执行的线程池（任务执行顺序不确定）【JDK 1.8 添加】

示例：

ExecutorService executorService = Executors.newFixedThreadPool(3);

ThreadPoolExecutor 创建的一种（ThreadPoolExecutor：最原始的创建线程池的方式，它包含了 7 个参数可供设置）

ThreadPoolExecutor提供了四个构造方法：

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

参数详解：

    public ThreadPoolExecutor(int corePoolSize, //核心线程数，线程池中始终存活的线程数int maximumPoolSize, //最大线程数，线程池中允许的最大线程数long keepAliveTime, //最大线程数可以存活的时间TimeUnit unit, //时间单位BlockingQueue<Runnable> workQueue, //阻塞队列，workQueue包含七种ThreadFactory threadFactory, //线程工厂，主要用来创建线程RejectedExecutionHandler handler //拒绝策略) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);this.prestartAllCoreThreads();}

实例：

    private static ExecutorService executorService;public static void main( String[] args ) {executorService = new ThreadPoolExecutor(2, 4, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(5),new ThreadFactory() {public Thread newThread(Runnable runnable) {log.info("创建线程:{}",runnable.hashCode());Thread threadName = new Thread(runnable,"threadPool"+runnable.hashCode());return threadName;}}, new ThreadPoolExecutor.CallerRunsPolicy());// ThreadTask类实现Runnable接口，这里省略了executorService.execute(new ThreadTask());}

ThreadPoolExecutor执行原理概述

点击executorService.execute(new ThreadTask())的execute()方法

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();// 1、工作线程 < 核心线程 if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}// 2、运行态，并尝试将任务加入队列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);} // 3、使用尝试使用最大线程运行else if (!addWorker(command, false))reject(command);}

这三处if判断，还是比较泛的，整体大框框上的流程，可用下图表示：

上述代码中的三种传参方式：

addWorker(command, true)：创建核心线程执行任务；
addWorker(command, false)：创建非核心线程执行任务；
addWorker(null, false)：创建非核心线程，当前任务为空；

addWorker的工作可分为两个部分：

第一部分：原子操作，判断是否可以创建worker。通过自旋、CAS、ctl 等操作，判断继续创建还是返回false，自旋周期一般很短。
第二部分：同步创建workder，并启动线程。

ThreadPoolExecutor的执行主要围绕Worker，Worker 实现了 AbstractQueuedSynchronizer 并继承了 Runnable

ExecutorService转换为ThreadPoolExecutor获取当前活动线程数

    public static void main(String[] args) {ExecutorService executorService = Executors.newFixedThreadPool(3);// 获取当前活动线程数int activeCount = ((ThreadPoolExecutor) executorService).getActiveCount();}

ThreadPoolExecutor常用方法：

Modifier and Type	Method and Description
`protected void`	`afterExecute(Runnable r, Throwable t)` 完成指定Runnable的执行后调用方法。
`void`	`allowCoreThreadTimeOut(boolean value)` 设置策略是否核心线程可能会超时，如果任务没有在活着的时间内到达，则在新任务到达时被替换。
`boolean`	`allowsCoreThreadTimeOut()` 如果此池允许核心线程超时并终止，如果没有任务在keepAlive时间内到达，则返回true，如果新任务到达时需要更换。
`boolean`	`awaitTermination(long timeout, TimeUnit unit)` 阻止所有任务在关闭请求完成后执行，或发生超时，或当前线程中断，以先到者为准。
`protected void`	`beforeExecute(Thread t, Runnable r)` 在给定的线程中执行给定的Runnable之前调用方法。
`void`	`execute(Runnable command)` 在将来某个时候执行给定的任务。
`protected void`	`finalize()` 当这个执行器不再被引用并且没有线程时，调用 `shutdown` 。
`int`	`getActiveCount()` 返回正在执行任务的线程的大概数量。
`long`	`getCompletedTaskCount()` 返回完成执行的任务的大致总数。
`int`	`getCorePoolSize()` 返回核心线程数。
`long`	`getKeepAliveTime(TimeUnit unit)` 返回线程保持活动时间，这是超过核心池大小的线程在终止之前可能保持空闲的时间量。
`int`	`getLargestPoolSize()` 返回在池中同时进行的最大线程数。
`int`	`getMaximumPoolSize()` 返回允许的最大线程数。
`int`	`getPoolSize()` 返回池中当前的线程数。
`BlockingQueue<Runnable>`	`getQueue()` 返回此执行程序使用的任务队列。
`RejectedExecutionHandler`	`getRejectedExecutionHandler()` 返回不可执行任务的当前处理程序。
`long`	`getTaskCount()` 返回计划执行的任务的大概总数。
`ThreadFactory`	`getThreadFactory()` 返回用于创建新线程的线程工厂。
`boolean`	`isShutdown()` 如果此执行者已关闭，则返回 `true` 。
`boolean`	`isTerminated()` 如果所有任务在关闭后完成，则返回 `true` 。
`boolean`	`isTerminating()` 如果此执行者在 `shutdown()`或 `shutdownNow()`之后终止，但尚未完全终止，则返回true。
`int`	`prestartAllCoreThreads()` 启动所有核心线程，导致他们等待工作。
`boolean`	`prestartCoreThread()` 启动核心线程，使其无法等待工作。
`void`	`purge()` 尝试从工作队列中删除已取消的所有`Future`任务。
`boolean`	`remove(Runnable task)` 如果此任务存在，则从执行程序的内部队列中删除此任务，从而导致该任务尚未运行。
`void`	`setCorePoolSize(int corePoolSize)` 设置核心线程数。
`void`	`setKeepAliveTime(long time, TimeUnit unit)` 设置线程在终止之前可能保持空闲的时间限制。
`void`	`setMaximumPoolSize(int maximumPoolSize)` 设置允许的最大线程数。
`void`	`setRejectedExecutionHandler(RejectedExecutionHandler handler)` 为不可执行的任务设置一个新的处理程序。
`void`	`setThreadFactory(ThreadFactory threadFactory)` 设置用于创建新线程的线程工厂。
`void`	`shutdown()` 启动有序关闭，其中先前提交的任务将被执行，但不会接受任何新任务。
`List<Runnable>`	`shutdownNow()` 尝试停止所有主动执行的任务，停止等待任务的处理，并返回正在等待执行的任务列表。
`protected void`	`terminated()` 执行程序已终止时调用方法。
`String`	`toString()` 返回标识此池的字符串及其状态，包括运行状态和估计的工作人员和任务计数的指示。