前言

创建一个新的线程可以通过继承Thread类或者实现Runnable接口来实现，这两种方式创建的线程在运行结束后会被虚拟机销毁，进行垃圾回收，如果线程数量过多，频繁的创建和销毁线程会浪费资源，降低效率。而线程池的引入就很好解决了上述问题，线程池可以更好的创建、维护、管理线程的生命周期，做到复用，提高资源的使用效率，也避免了开发人员滥用new关键字创建线程的不规范行为。

说明：阿里开发手册中明确指出，在实际生产中，线程资源必须通过线程池提供，不允许在应用中显式的创建线程。如果不使用线程池，有可能造成系统创建大量同类线程而导致消耗完内存或者“过度切换”的问题。

接下来主要对Java中线程池核心实现类ThreadPoolExecutor核心参数及工作原理、Executors工具类等，进行说明。

ThreadPoolExecutor

ThreadPoolExecutor是线程池的核心实现类，在JDK1.5引入，位于java.util.concurrent包，由Doug Lea完成。

Executor接口

Executor是线程池的顶层接口，JDK1.5开始引入了，位于java.util.concurrent 包。

public interface Executor {// 该接口中只定义了一个Runnable作为入参的execute方法void execute(Runnable command);
}

查看Executor接口的实现类图

  • Executor线程池相关顶级接口，它将任务的提交与任务的执行分离开来
  • ExecutorService继承并扩展了Executor接口，提供了Runnable、FutureTask等主要线程实现接口扩展
  • ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务
  • ScheduledExecutorService继承ExecutorService接口，并定义延迟或定期执行的方法
  • ScheduledThreadPoolExecutor继承ThreadPoolExecutor并实现了ScheduledExecutorService接口，是延时执行类任务的主要实现

生命周期

线程存在生命周期，同样线程池也有生命周期，源码中定义了五种状态。

private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

关于线程池状态间转换如下图所示：

构造方法

如何利用ThreadPoolExecutor创建一个线程池，查看其构造方法

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.acc = System.getSecurityManager() == null ?null :AccessController.getContext();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

ThreadPoolExecutor包含了7个核心参数，参数含义：

  • corePoolSize：核心线程池的大小
  • maximumPoolSize：最大线程池的大小
  • keepAliveTime：当线程池中线程数大于corePoolSize，并且没有可执行任务时大于corePoolSize那部分线程的存活时间
  • unit：keepAliveTime的时间单位
  • workQueue：用来暂时保存任务的工作队列
  • threadFactory：线程工厂提供线程的创建方式，默认使用Executors.defaultThreadFactory()
  • handler：当线程池所处理的任务数超过其承载容量或关闭后继续有任务提交时，所调用的拒绝策略

核心参数

ThreadPoolExecutor中包含了七大核心参数，如果需要对线程池进行定制化操作，需要对其中比较核心的参数进行一定程度的认识。

corePoolSize

ThreadPoolExecutor会根据corePoolSize和maximumPoolSize在构造方法中设置的边界值自动调整池大小，也可以使用setCorePoolSize和setMaximumPoolSize动态更改，关于线程数量的自动调整分为以下两种场景：

  • 线程数量小于corePoolSize

当在线程池中提交了一个新任务，并且运行的线程少于corePoolSize时，即使其他工作线程处于空闲状态，也会创建一个新线程来处理该请求。

  • 线程数量介于corePoolSize和maximumPoolSize之间

如果运行的线程数多于corePoolSize但少于maximumPoolSize，则仅当队列已满时才会创建新线程。

如果corePoolSize和maximumPoolSize相同，那么可以创建一个固定大小的线程池。如果maximumPoolSize被设置为无界值(Integer.MAX_VALUE)，在资源允许的前提下，意味着线程池允许容纳任意数量的并发任务。
默认情况下，即使是核心线程也会在新任务到达时开始创建和启动，如果使用非空队列创建线程池池，可以通过重写prestartCoreThread或prestartAllCoreThreads方法动态覆盖，进行线程预启动。

在实际开发中，如果需要自定义线程数量，可以参考以下公式：


其中参数含义如下：

  • _Ncpu_是处理器的核数目，可以通过Runtime.getRuntime().availableProcessors()获得
  • _Ucpu_是期望的CPU利用率，介于0-1之间
  • W/C是等待时间与计算时间的比率

keepAliveTime

keepAliveTime参数用来来设置空闲时间。如果池当前有多个corePoolSize线程，多余的线程如果空闲时间超过将会被终止，这种机制减少了在任务数量较少时线程池资源消耗。如果某个时间需要处理的任务数量增加，则将构造新线程。使用方法setKeepAliveTime可以动态更改参数值。
默认情况下，keep-alive策略仅适用于超过corePoolSize线程的情况，但是方法allowCoreThreadTimeOut也可用于将此超时策略应用于核心线程，只要 keepAliveTime值不为零即可。

workQueue

workQueue参数用来指定存放提交任务的队列，任何BlockingQueue都可以用来传输和保存提交的任务。关于队列大小与线程数量之间存在这样的关系：

  • 如果线程数少于corePoolSize，对于提交的新任务会创建一个新的线程处理，并不会把任务放入队列
  • 如果线程数介于corePoolSize和maximumPoolSize之间，新提交的任务会被放入阻塞队列中
  • 如果线程池处于饱和状态，即无法创建线程也无法存放在阻塞队列，那么新任务将交由拒绝策略来处理

线程池中的常用阻塞队列一般包含SynchronousQueue、LinkedBlockingQueue、ArrayBlockingQueue几种，它们都是BlockingQueue的实现类，下面进行简单介绍。

SynchronousQueue

SynchronousQueue并不能算得上一个真正的队列，虽然实现了BlockingQueue接口，但是并没有容量，不能存储任务。只是维护一组线程，在等待着把元素加入或移出队列，相当于直接交接任务给具体执行的线程。
如果没有立即可用的线程来运行任务，则尝试将任务排队失败，因此将构造一个新线程。在处理可能具有内部依赖关系的请求集时，此策略可避免锁定。这种队列方式通常需要无限的maximumPoolSizes以避免拒绝新提交的任务。当任务提交的平均到达速度快于线程处理速度时，线程存在无限增长的可能性，而CachedThreadPool正式采用这种形式。

LinkedBlockingQueue

LinkedBlockingQueue是采用链表实现的无界队列，如果使用没有预定义容量的LinkedBlockingQueue，当所有corePoolSize线程都在处理任务时，将导致新任务都会在队列中等待，不会创建超过corePoolSize个线程。这种场景下maximumPoolSize的值对于线程数量没有任何影响。
这种依托队列处理任务的方式恰与SynchronousQueue依托线程处理任务的方式相反。

ArrayBlockingQueue

ArrayBlockingQueue是通过数组实现的有界队列。有界队列在与有限的maximumPoolSizes一起使用时有助于防止资源耗尽，但可能更难以调整和控制。使用ArrayBlockingQueue可以根据应用场景，预先估计池和队列的容量，互相权衡队列大小和最大池大小：

  • 使用大队列和小池：减少线程数量，可以最大限度地减少CPU使用率、操作系统资源和上下文切换开销，但可能会导致吞吐量降低
  • 使用小队列大池：较大数量的线程，如果任务提交速度过快，会在短时间内提升CPU使用率，理论上可以提高系统的吞吐量。如果任务经常阻塞(如受到IO限制)，会使得CPU切换更加频繁，可能会遇到更大的调度开销，这也会降低吞吐量

threadFactory

该参数提供了线程池中线程的创建方式，这里使用了工厂模式ThreadFactory创建新线程，默认情况下，会使用 Executors.defaultThreadFactory，它创建的线程都在同一个ThreadGroup中，并具有相同的NORM_PRIORITY优先级和非守护进程状态。
也可以根据实际场景自定义ThreadFactory，可以更改线程的名称、线程组、优先级、守护程序状态等，在自定义情况下需要注意的是如果ThreadFactory在从newThread返回null时未能创建线程，则执行程序将继续，但可能无法执行任何任务。线程应该拥有“modifyThread”RuntimePermission。如果工作线程或其他使用该池的线程不具备此权限，则服务可能会降级：配置更改可能无法及时生效，关闭池可能会一直处于可以终止但未完成的状态。

handler

如果线程池处于饱和状态，没有足够的线程数或者队列空间来处理提交的任务，或者是线程池已经处于关闭状态但还在处理进行中的任务，那么继续提交的任务就会根据线程池的拒绝策略处理。
无论哪种情况，execute方法都会调用其RejectedExecutionHandler的rejectedExecution方法。线程池中提供了四个预定义的处理程序策略：

  • ThreadPoolExecutor.AbortPolicy (默认)
  • ThreadPoolExecutor.DiscardPolicy
  • ThreadPoolExecutor.DiscardOldestPolicy
  • ThreadPoolExecutor.CallerRunsPolicy

这些预定义策略都实现了RejectedExecutionHandler接口，也可以定义实现类重写拒绝策略。

AbortPolicy

查看AbortPolicy的源码，处理程序在拒绝时抛出运行时异常RejectedExecutionException 。

public static class AbortPolicy implements RejectedExecutionHandler {/*** Creates an {@code AbortPolicy}.*/public AbortPolicy() { }/*** Always throws RejectedExecutionException.** @param r the runnable task requested to be executed* @param e the executor attempting to execute this task* @throws RejectedExecutionException always*/public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {throw new RejectedExecutionException("Task " + r.toString() +" rejected from " +e.toString());}
}

DiscardPolicy

查看源码，无法执行的任务被简单地丢弃，不做任何处理。

public static class DiscardPolicy implements RejectedExecutionHandler {/*** Creates a {@code DiscardPolicy}.*/public DiscardPolicy() { }/*** Does nothing, which has the effect of discarding task r.** @param r the runnable task requested to be executed* @param e the executor attempting to execute this task*/public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {}
}

DiscardOldestPolicy

查看源码，如果executor没有关闭，工作队列头部的任务就会被丢弃，然后重试执行(可能会再次失败，导致这个重复。

public static class DiscardOldestPolicy implements RejectedExecutionHandler {/*** Creates a {@code DiscardOldestPolicy} for the given executor.*/public DiscardOldestPolicy() { }/*** Obtains and ignores the next task that the executor* would otherwise execute, if one is immediately available,* and then retries execution of task r, unless the executor* is shut down, in which case task r is instead discarded.** @param r the runnable task requested to be executed* @param e the executor attempting to execute this task*/public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {e.getQueue().poll();e.execute(r);}}
}

CallerRunsPolicy

查看源码，这种策略会调用执行自身的线程运行任务，这也提供了一个简单的反馈控制机制，可以减慢提交新任务的速度。

public static class CallerRunsPolicy implements RejectedExecutionHandler {/*** Creates a {@code CallerRunsPolicy}.*/public CallerRunsPolicy() { }/*** Executes task r in the caller's thread, unless the executor* has been shut down, in which case the task is discarded.** @param r the runnable task requested to be executed* @param e the executor attempting to execute this task*/public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {r.run();}}
}

钩子函数

ThreadPoolExecutor提供受保护的可重写的钩子函数，用于在线程池中线程在初始化或者执行完任务后做一些特殊处理，同样也提供了在线程池终止时可以覆写的terminated方法。

beforeExecute

线程中执行Runnable之前调用的方法。此方法由将执行任务r的线程t调用，可用于重新初始化 ThreadLocals，或执行日志记录。这实现什么都不做，但可以在子类中定制。需要注意的事是，要正确嵌套多个覆盖，子类通常应在此方法的末尾调用 super.beforeExecute。查看源码：

protected void beforeExecute(Thread t, Runnable r) { }

afterExecute

在完成给定的Runnable任务时调用的方法，此方法由执行任务的线程调用。需要注意，要正确嵌套多个覆盖，子类通常应在此方法的开头调用 super.afterExecute。查看源码：

protected void afterExecute(Runnable r, Throwable t) { }

terminated

Executor终止时调用的方法。需要注意的是子类通常应在此方法中调用 super.terminated。查看源码：

protected void terminated() { }

核心源码分析

线程池的整体工作流程可以参考如下图所示：

ThreadPoolExecutor执行execute方法分下面4种情况。

  1. 如果当前运行的线程少于corePoolSize,则创建新线程来执行任务(注意,执行这一步骤需要获取全局锁)。
  2. 如果运行的线程等于或多于corePoolSize,则将任务加入BlockingQueue。
  3. 如果无法将任务加入BlockingQueue(队列已满),则创建新的线程来处理任务(注意,执行这一步骤需要获取全局锁)。
  4. 如果创建新线程将使当前运行的线程超出maximumPoolSize,任务将被拒绝,并调用RejectedExecutionHandler.rejectedExecution()方法。

ThreadPoolExecutor采取上述步骤的总体设计思路,是为了在执行execute()方法时,尽可能地避免获取全局锁(那将会是一个严重的可伸缩瓶颈)。在ThreadPoolExecutor完成预热之后(当前运行的线程数大于等于corePoolSize),几乎所有的execute()方法调用都是执行步骤2,而步骤2不需要获取全局锁。

任务单元Worker

ThreadPoolExecutor中核心任务单元是由一个Worker内部类来实现，Worker类中定义了两个重要方法runWorker方法和addWorker方法。

private final class Workerextends AbstractQueuedSynchronizerimplements Runnable
{/*** This class will never be serialized, but we provide a* serialVersionUID to suppress a javac warning.*/private static final long serialVersionUID = 6138294804551838833L;/** Thread this worker is running in.  Null if factory fails. */final Thread thread;/** Initial task to run.  Possibly null. */Runnable firstTask;/** Per-thread task counter */volatile long completedTasks;/*** Creates with given first task and thread from ThreadFactory.* @param firstTask the first task (null if none)*/Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;// 这儿是Worker的关键所在，使用了线程工厂创建了一个线程。传入的参数为当前workerthis.thread = getThreadFactory().newThread(this);}/** Delegates main run loop to outer runWorker  */public void run() {runWorker(this);}// 省略代码...
}

addWorker和runWorker