明天说一说，线程池，从设计思维到源码解析。

前言

各位小伙伴儿，春节曾经完结了，在此献上一篇肝了一个春节假期的迟来的拜年之作，心愿读者敌人们都能有播种。多多点赞、评论、珍藏！

初识线程池

咱们晓得，线程的创立和销毁都须要映射到操作系统，因而其代价是比拟昂扬的。出于防止频繁创立、销毁线程以及不便线程治理的须要，线程池应运而生。

线程池劣势

  • 「升高资源耗费」：线程池通常会保护一些线程（数量为 corePoolSize），这些线程被重复使用来执行不同的工作，工作实现后不会销毁。在待处理任务量很大的时候，通过对线程资源的复用，防止了线程的频繁创立与销毁，从而升高了系统资源耗费。
  • 「进步响应速度」：因为线程池保护了一批 alive 状态的线程，当工作达到时，不须要再创立线程，而是间接由这些线程去执行工作，从而缩小了工作的等待时间。
  • 「进步线程的可管理性」：应用线程池能够对线程进行对立的调配，调优和监控。

线程池设计思路

有句话叫做艺术来源于生存，编程语言也是如此，很多设计思维能映射到日常生活中，比方面向对象思维、封装、继承，等等。明天咱们要说的线程池，它同样能够在事实世界找到对应的实体——工厂。

先假想一个工厂的生产流程：

线程池设计思路

工厂中有固定的一批工人，称为正式工人，工厂接管的订单由这些工人去实现。当订单减少，正式工人曾经忙不过来了，工厂会将生产原料临时沉积在仓库中，等有闲暇的工人时再解决（因为工人闲暇了也不会被动解决仓库中的生产工作，所以须要调度员实时调度）。仓库沉积满了后，订单还在减少怎么办？工厂只能长期扩招一批工人来应答生产顶峰，而这批工人顶峰完结后是要清退的，所以称为临时工。过后临时工也以招满后（受限于工位限度，临时工数量有下限），前面的订单只能忍痛回绝了。

咱们做如下一番映射：

  • 工厂——线程池
  • 订单——工作（Runnable）
  • 正式工人——外围线程
  • 临时工——一般线程
  • 仓库——工作队列
  • 调度员——getTask()

❝

getTask()是一个办法，将工作队列中的任务调度给闲暇线程，在解读线程池有具体介绍

❞

映射后，造成线程池流程图如下，两者是不是有殊途同归之妙？

线程池流程图

这样，线程池的工作原理或者说流程就很好了解了，提炼成一个简图：

线程池的工作原理

深刻线程池

那么接下来，问题来了，线程池是具体如何实现这套工作机制的呢？从Java线程池Executor框架体系能够看出：线程池的真正实现类是ThreadPoolExecutor，因而咱们接下来重点钻研这个类。

线程池工作机制

构造方法

钻研一个类，先从它的构造方法开始。ThreadPoolExecutor提供了4个有参构造方法：

public ThreadPoolExecutor(int corePoolSize,                          int maximumPoolSize,                          long keepAliveTime,                          TimeUnit unit,                          BlockingQueue<Runnable> workQueue) {    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             Executors.defaultThreadFactory(), defaultHandler);}public ThreadPoolExecutor(int corePoolSize,                          int maximumPoolSize,                          long keepAliveTime,                          TimeUnit unit,                          BlockingQueue<Runnable> workQueue,                          ThreadFactory threadFactory) {    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             threadFactory, defaultHandler);}public ThreadPoolExecutor(int corePoolSize,                          int maximumPoolSize,                          long keepAliveTime,                          TimeUnit unit,                          BlockingQueue<Runnable> workQueue,                          RejectedExecutionHandler handler) {    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             Executors.defaultThreadFactory(), handler);}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.corePoolSize = corePoolSize;    this.maximumPoolSize = maximumPoolSize;    this.workQueue = workQueue;    this.keepAliveTime = unit.toNanos(keepAliveTime);    this.threadFactory = threadFactory;    this.handler = handler;}

解释一下构造方法中波及到的参数：

  • 「corePoolSize」（必须）：外围线程数。即池中始终放弃存活的线程数，即便这些线程处于闲暇。然而将allowCoreThreadTimeOut参数设置为true后，外围线程处于闲暇一段时间以上，也会被回收。
  • 「maximumPoolSize」（必须）：池中容许的最大线程数。当外围线程全副忙碌且工作队列打满之后，线程池会长期追加线程，直到总线程数达到maximumPoolSize这个下限。
  • 「keepAliveTime」（必须）：线程闲暇超时工夫。当非核心线程处于闲暇状态的工夫超过这个工夫后，该线程将被回收。将allowCoreThreadTimeOut参数设置为true后，外围线程也会被回收。
  • 「unit」（必须）：keepAliveTime参数的工夫单位。有：TimeUnit.DAYS（天）、TimeUnit.HOURS（小时）、TimeUnit.MINUTES（分钟）、「TimeUnit.SECONDS（秒）」、「TimeUnit.MILLISECONDS（毫秒）」、TimeUnit.MICROSECONDS（微秒）、TimeUnit.NANOSECONDS（纳秒）
  • 「workQueue」（必须）：工作队列，采纳阻塞队列实现。当外围线程全副忙碌时，后续由execute办法提交的Runnable将寄存在工作队列中，期待被线程解决。
  • 「threadFactory」（可选）：线程工厂。指定线程池创立线程的形式。
  • 「handler」（可选）：回绝策略。当线程池中线程数达到maximumPoolSize且workQueue打满时，后续提交的工作将被回绝，handler能够指定用什么形式回绝工作。

放到一起再看一下：

工厂与线程池

工作队列

应用ThreadPoolExecutor须要指定一个实现了BlockingQueue接口的工作期待队列。在ThreadPoolExecutor线程池的API文档中，一共举荐了三种期待队列，它们是：SynchronousQueue、LinkedBlockingQueue和ArrayBlockingQueue；

  1. 「SynchronousQueue」：同步队列。这是一个外部没有任何容量的阻塞队列，任何一次插入操作的元素都要期待绝对的删除/读取操作，否则进行插入操作的线程就要始终期待，反之亦然。
  2. 「LinkedBlockingQueue」：无界队列（严格来说并非无界，下限是Integer.MAX_VALUE），基于链表构造。应用无界队列后，当外围线程都忙碌时，后续工作能够有限退出队列，因而线程池中线程数不会超过外围线程数。这种队列能够进步线程池吞吐量，但代价是就义内存空间，甚至会导致内存溢出。另外，应用它时能够指定容量，这样它也就是一种有界队列了。
  3. 「ArrayBlockingQueue」：有界队列，基于数组实现。在线程池初始化时，指定队列的容量，后续无奈再调整。这种有界队列有利于避免资源耗尽，但可能更难调整和管制。

另外，Java还提供了另外4种队列：

  1. 「PriorityBlockingQueue」：反对优先级排序的无界阻塞队列。寄存在PriorityBlockingQueue中的元素必须实现Comparable接口，这样能力通过实现compareTo()办法进行排序。优先级最高的元素将始终排在队列的头部；PriorityBlockingQueue不会保障优先级一样的元素的排序，也不保障以后队列中除了优先级最高的元素以外的元素，随时处于正确排序的地位。
  2. 「DelayQueue」：提早队列。基于二叉堆实现，同时具备：无界队列、阻塞队列、优先队列的特色。DelayQueue提早队列中寄存的对象，必须是实现Delayed接口的类对象。通过执行时延从队列中提取工作，工夫没到工作取不进去。更多内容请见DelayQueue：面试官：谈谈Java中的阻塞提早队列DelayQueue原理和用法。
  3. 「LinkedBlockingDeque」：双端队列。基于链表实现，既能够从尾部插入/取出元素，还能够从头部插入元素/取出元素。
  4. 「LinkedTransferQueue」：由链表构造组成的无界阻塞队列。这个队列比拟特地的时，采纳一种预占模式，意思就是消费者线程取元素时，如果队列不为空，则间接取走数据，若队列为空，那就生成一个节点（节点元素为null）入队，而后消费者线程被期待在这个节点上，前面生产者线程入队时发现有一个元素为null的节点，生产者线程就不入队了，间接就将元素填充到该节点，并唤醒该节点期待的线程，被唤醒的消费者线程取走元素。

回绝策略

线程池有一个重要的机制：回绝策略。当线程池workQueue已满且无奈再创立新线程池时，就要回绝后续工作了。回绝策略须要实现RejectedExecutionHandler接口，不过Executors框架曾经为咱们实现了4种回绝策略：

  1. 「AbortPolicy」（默认）：抛弃工作并抛出RejectedExecutionException异样。
  2. 「CallerRunsPolicy」：间接运行这个工作的run办法，但并非是由线程池的线程解决，而是交由工作的调用线程解决。
  3. 「DiscardPolicy」：间接抛弃工作，不抛出任何异样。
  4. 「DiscardOldestPolicy」：将以后处于期待队列列头的期待工作强行取出，而后再试图将以后被回绝的工作提交到线程池执行。

线程工厂指定创立线程的形式，这个参数不是必选项，Executors类曾经为咱们十分贴心地提供了一个默认的线程工厂：

/** * The default thread factory */static class DefaultThreadFactory 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;    DefaultThreadFactory() {        SecurityManager s = System.getSecurityManager();        group = (s != null) ? s.getThreadGroup() :                              Thread.currentThread().getThreadGroup();        namePrefix = "pool-" +                      poolNumber.getAndIncrement() +                     "-thread-";    }    public 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;    }}

线程池状态

线程池有5种状态：

volatile int runState;// runState is stored in the high-order bitsprivate 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;

runState示意以后线程池的状态，它是一个 volatile 变量用来保障线程之间的可见性。

上面的几个static final变量示意runState可能的几个取值，有以下几个状态：

  • 「RUNNING」：当创立线程池后，初始时，线程池处于RUNNING状态；
  • 「SHUTDOWN」：如果调用了shutdown()办法，则线程池处于SHUTDOWN状态，此时线程池不可能承受新的工作，它会期待所有工作执行结束；
  • 「STOP」：如果调用了shutdownNow()办法，则线程池处于STOP状态，此时线程池不能承受新的工作，并且会去尝试终止正在执行的工作；
  • 「TERMINATED」：当线程池处于SHUTDOWN或STOP状态，并且所有工作线程曾经销毁，工作缓存队列曾经清空或执行完结后，线程池被设置为TERMINATED状态。

初始化&容量调整&敞开

「1、线程初始化」

默认状况下，创立线程池之后，线程池中是没有线程的，须要提交工作之后才会创立线程。

在理论中如果须要线程池创立之后立刻创立线程，能够通过以下两个办法办到：

  • 「prestartCoreThread()」：boolean prestartCoreThread()，初始化一个外围线程
  • 「prestartAllCoreThreads()」：int prestartAllCoreThreads()，初始化所有外围线程，并返回初始化的线程数
public boolean prestartCoreThread() {    return addIfUnderCorePoolSize(null); //留神传进去的参数是null}public int prestartAllCoreThreads() {    int n = 0;    while (addIfUnderCorePoolSize(null))//留神传进去的参数是null        ++n;    return n;}

「2、线程池敞开」

ThreadPoolExecutor提供了两个办法，用于线程池的敞开：

  • 「shutdown()」：不会立刻终止线程池，而是要等所有工作缓存队列中的工作都执行完后才终止，但再也不会承受新的工作
  • 「shutdownNow()」：立刻终止线程池，并尝试打断正在执行的工作，并且清空工作缓存队列，返回尚未执行的工作

「3、线程池容量调整」

ThreadPoolExecutor提供了动静调整线程池容量大小的办法：

  • 「setCorePoolSize」：设置外围池大小
  • 「setMaximumPoolSize」：设置线程池最大能创立的线程数目大小

当上述参数从小变大时，ThreadPoolExecutor进行线程赋值，还可能立刻创立新的线程来执行工作。

应用线程池

ThreadPoolExecutor

通过构造方法应用ThreadPoolExecutor是线程池最间接的应用形式，上面看一个实例：

import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.ThreadPoolExecutor;import java.util.concurrent.TimeUnit;public class MyTest { public static void main(String[] args) {  // 创立线程池  ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3, 5, 5, TimeUnit.SECONDS,    new ArrayBlockingQueue<Runnable>(5));  // 向线程池提交工作  for (int i = 0; i < threadPool.getCorePoolSize(); i++) {   threadPool.execute(new Runnable() {    @Override    public void run() {     for (int x = 0; x < 2; x++) {      System.out.println(Thread.currentThread().getName() + ":" + x);      try {       Thread.sleep(2000);      } catch (InterruptedException e) {       e.printStackTrace();      }     }    }   });  }  // 敞开线程池  threadPool.shutdown(); // 设置线程池的状态为SHUTDOWN，而后中断所有没有正在执行工作的线程  // threadPool.shutdownNow(); // 设置线程池的状态为STOP，而后尝试进行所有的正在执行或暂停工作的线程，并返回期待执行工作的列表，该办法要慎用，容易造成不可控的结果 }}

运行后果：

pool-1-thread-2:0pool-1-thread-1:0pool-1-thread-3:0pool-1-thread-2:1pool-1-thread-3:1pool-1-thread-1:1

Executors封装线程池

另外，Executors封装好了4种常见的性能线程池（还是那么地贴心）：

「1、FixedThreadPool」

固定容量线程池。其特点是最大线程数就是外围线程数，意味着线程池只能创立外围线程，keepAliveTime为0，即线程执行完工作立刻回收。工作队列未指定容量，代表应用默认值Integer.MAX_VALUE。实用于须要管制并发线程的场景。

// 应用默认线程工厂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);}

应用示例：

// 1. 创立线程池对象，设置外围线程和最大线程数为5ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);// 2. 创立Runnable（工作）Runnable task =new Runnable(){  public void run() {     System.out.println(Thread.currentThread().getName() + "--->运行");  }};// 3. 向线程池提交工作fixedThreadPool.execute(task);

「2、 SingleThreadExecutor」

单线程线程池。特点是线程池中只有一个线程（外围线程），线程执行完工作立刻回收，应用有界阻塞队列（容量未指定，应用默认值Integer.MAX_VALUE）

public static ExecutorService newSingleThreadExecutor() {    return new FinalizableDelegatedExecutorService        (new ThreadPoolExecutor(1, 1,                                0L, TimeUnit.MILLISECONDS,                                new LinkedBlockingQueue<Runnable>()));}// 为节俭篇幅，省略了自定义线程工厂形式的源码

应用示例：

// 1. 创立单线程线程池ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();// 2. 创立Runnable（工作）Runnable task = new Runnable(){  public void run() {     System.out.println(Thread.currentThread().getName() + "--->运行");  }};// 3. 向线程池提交工作singleThreadExecutor.execute(task);

「3、 ScheduledThreadPool」

定时线程池。指定外围线程数量，一般线程数量有限，线程执行完工作立刻回收，工作队列为延时阻塞队列。这是一个比拟特地的线程池，实用于「执行定时或周期性的工作」。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {    return new ScheduledThreadPoolExecutor(corePoolSize);}// 继承了 ThreadPoolExecutorpublic class ScheduledThreadPoolExecutor extends ThreadPoolExecutor        implements ScheduledExecutorService {    // 构造函数，省略了自定义线程工厂的构造函数 public ScheduledThreadPoolExecutor(int corePoolSize) {     super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,           new DelayedWorkQueue()); }  // 延时执行工作 public ScheduledFuture<?> schedule(Runnable command,                                       long delay,                                       TimeUnit unit) {        ...    } // 定时执行工作 public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,                                                  long initialDelay,                                                  long period,                                                  TimeUnit unit) {...}}

应用示例：

// 1. 创立定时线程池ExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);// 2. 创立Runnable（工作）Runnable task = new Runnable(){  public void run() {     System.out.println(Thread.currentThread().getName() + "--->运行");  }};// 3. 向线程池提交工作scheduledThreadPool.schedule(task, 2, TimeUnit.SECONDS); // 提早2s后执行工作scheduledThreadPool.scheduleAtFixedRate(task,50,2000,TimeUnit.MILLISECONDS);// 提早50ms后、每隔2000ms执行工作

「4、CachedThreadPool」

缓存线程池。没有外围线程，一般线程数量为Integer.MAX_VALUE（能够了解为有限），线程闲置60s后回收，工作队列应用SynchronousQueue这种无容量的同步队列。实用于「任务量大但耗时低」的场景。

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

应用示例：

// 1. 创立缓存线程池ExecutorService cachedThreadPool = Executors.newCachedThreadPool();// 2. 创立Runnable（工作）Runnable task = new Runnable(){  public void run() {     System.out.println(Thread.currentThread().getName() + "--->运行");  }};// 3. 向线程池提交工作cachedThreadPool.execute(task);

解读线程池

OK，置信后面内容浏览起来还算轻松愉悦吧，那么从这里开始就进入深水区了，如果前面内容能吃透，那么线程池常识就真的被你把握了。

咱们晓得，向线程池提交工作是用ThreadPoolExecutor的execute()办法，但在其外部，线程工作的解决其实是相当简单的，波及到ThreadPoolExecutor、Worker、Thread三个类的6个办法：

向线程池提交工作

execute()

在ThreadPoolExecutor类中，工作提交办法的入口是execute(Runnable command)办法（submit()办法也是调用了execute()），该办法其实只在尝试做一件事：通过各种校验之后，调用 addWorker(Runnable command,boolean core)办法为线程池创立一个线程并执行工作，与之绝对应，execute() 的后果有两个：

「参数阐明：」

  1. 「Runnable command」：待执行的工作

「执行流程：」

1、通过 ctl.get() 失去线程池的以后线程数，如果线程数小于corePoolSize，则调用 addWorker(commond,true)办法创立新的线程执行工作，否则执行步骤2；

2、步骤1失败，阐明曾经无奈再创立新线程，那么思考将工作放入阻塞队列，期待执行完工作的线程来解决。基于此，判断线程池是否处于Running状态（只有Running状态的线程池能够承受新工作），如果工作增加到工作队列胜利则进入步骤3，失败则进入步骤4；

3、来到这一步须要阐明工作曾经退出工作队列，这时要二次校验线程池的状态，会有以下情景：

  • 线程池不再是Running状态了，须要将工作从工作队列中移除，如果移除胜利则回绝本次工作
  • 线程池是Running状态，则判断线程池工作线程是否为0，是则调用 addWorker(commond,true)增加一个没有初始工作的线程（这个线程将去获取曾经退出工作队列的本次工作并执行），否则进入步骤4；
  • 线程池不是Running状态，但从工作队列移除工作失败（可能已被某线程获取？），进入步骤4；

4、将线程池扩容至maximumPoolSize并调用 addWorker(commond,false)办法创立新的线程执行工作，失败则回绝本次工作。

「流程图：」

创立新的线程执行工作

「源码详读：」

/** * 在未来的某个时候执行给定的工作。工作能够在新线程中执行，也能够在现有的池线程中执行。 * 如果因为此执行器已敞开或已达到其容量而无奈提交工作以供执行，则由以后的{@code RejectedExecutionHandler}解决该工作。 *  * @param command the task to execute  待执行的工作命令 */public void execute(Runnable command) {    if (command == null)        throw new NullPointerException();    /*     * Proceed in 3 steps:     *      * 1. 如果运行的线程少于corePoolSize，将尝试以给定的命令作为第一个工作启动新线程。     *     * 2. 如果一个工作能够胜利排队，那么咱们依然须要仔细检查两点，其一，咱们是否应该增加一个线程     * （因为自从上次查看至今，一些存在的线程曾经死亡），其二，线程池状态此时已扭转成非运行态。因而，咱们从新查看状态，如果查看不通过，则移除曾经入列的工作，如果查看通过且线程池线程数为0，则启动新线程。     *      * 3. 如果无奈将工作退出工作队列，则将线程池扩容到极限容量并尝试创立一个新线程，如果失败则回绝工作。     */    int c = ctl.get();       // 步骤1：判断线程池以后线程数是否小于线程池大小    if (workerCountOf(c) < corePoolSize) {        // 减少一个工作线程并增加工作，胜利则返回，否则进行步骤2        // true代表应用coreSize作为边界束缚，否则应用maximumPoolSize        if (addWorker(command, true))            return;        c = ctl.get();    }    // 步骤2：不满足workerCountOf(c) < corePoolSize或addWorker失败，进入步骤2    // 校验线程池是否是Running状态且工作是否胜利放入workQueue（阻塞队列）    if (isRunning(c) && workQueue.offer(command)) {        int recheck = ctl.get();        // 再次校验，如果线程池非Running且从工作队列中移除工作胜利，则回绝该工作        if (! isRunning(recheck) && remove(command))            reject(command);        // 如果线程池工作线程数量为0，则新建一个空工作的线程        else if (workerCountOf(recheck) == 0)            // 如果线程池不是Running状态，是退出不进去的            addWorker(null, false);    }    // 步骤3：如果线程池不是Running状态或工作入列失败，尝试扩容maxPoolSize后再次addWorker，失败则回绝工作    else if (!addWorker(command, false))        reject(command);}

addWorker()

addWorker(Runnable firstTask, boolean core) 办法，顾名思义，向线程池增加一个带有工作的工作线程。

「参数阐明：」

  1. 「Runnable firstTask」：新创建的线程应该首先运行的工作（如果没有，则为空）。
  2. 「boolean core」：该参数决定了线程池容量的约束条件，即以后线程数量以何值为极限值。参数为 true 则应用corePollSize 作为束缚值，否则应用maximumPoolSize。

「执行流程：」

1、外层循环判断线程池的状态是否能够新增工作线程。这层校验基于上面两个准则：

  • 线程池为Running状态时，既能够承受新工作也能够解决工作
  • 线程池为敞开状态时只能新增空工作的工作线程（worker）解决工作队列（workQueue）中的工作不能承受新工作

2、内层循环向线程池增加工作线程并返回是否增加胜利的后果。

  • 首先校验线程数是否曾经超限度，是则返回false，否则进入下一步
  • 通过CAS使工作线程数+1，胜利则进入步骤3，失败则再次校验线程池是否是运行状态，是则持续内层循环，不是则返回外层循环

3、外围线程数量+1胜利的后续操作：增加到工作线程汇合，并启动工作线程

  • 首先获取锁之后，再次校验线程池状态（具体校验规定见代码注解），通过则进入下一步，未通过则增加线程失败
  • 线程池状态校验通过后，再查看线程是否曾经启动，是则抛出异样，否则尝试将线程退出线程池
  • 查看线程是否启动胜利，胜利则返回true，失败则进入 addWorkerFailed 办法

「流程图：」

向线程池增加一个带有工作的工作线程

「源码详读：」

private boolean addWorker(Runnable firstTask, boolean core) {    // 外层循环：判断线程池状态    retry:    for (;;) {        int c = ctl.get();        int rs = runStateOf(c);        /**          * 1.线程池为非Running状态（Running状态则既能够新增外围线程也能够接受任务）         * 2.线程为shutdown状态且firstTask为空且队列不为空         * 3.满足条件1且条件2不满足，则返回false         * 4.条件2解读：线程池为shutdown状态时且工作队列不为空时，能够新增空工作的线程来解决队列中的工作         */        if (rs >= SHUTDOWN &&            ! (rs == SHUTDOWN &&               firstTask == null &&               ! workQueue.isEmpty()))            return false;  // 内层循环：线程池增加外围线程并返回是否增加胜利的后果        for (;;) {            int wc = workerCountOf(c);            // 校验线程池已有线程数量是否超限：            // 1.线程池最大下限CAPACITY             // 2.corePoolSize或maximumPoolSize（取决于入参core）            if (wc >= CAPACITY ||                wc >= (core ? corePoolSize : maximumPoolSize))                 return false;            // 通过CAS操作使工作线程数+1，跳出外层循环            if (compareAndIncrementWorkerCount(c))                 break retry;            // 线程+1失败，重读ctl            c = ctl.get();   // Re-read ctl            // 如果此时线程池状态不再是running，则从新进行外层循环            if (runStateOf(c) != rs)                continue retry;            // 其余 CAS 失败是因为工作线程数量扭转了，持续内层循环尝试CAS对线程数+1            // else CAS failed due to workerCount change; retry inner loop        }    }    /**     * 外围线程数量+1胜利的后续操作：增加到工作线程汇合，并启动工作线程     */    boolean workerStarted = false;    boolean workerAdded = false;    Worker w = null;    try {        final ReentrantLock mainLock = this.mainLock;        w = new Worker(firstTask);        final Thread t = w.thread;        if (t != null) {            // 上面代码须要加锁：线程池主锁            mainLock.lock();             try {                // 持锁期间从新查看，线程工厂创立线程失败或获取锁之前敞开的状况产生时，退出                int c = ctl.get();                int rs = runStateOf(c);    // 再次测验线程池是否是running状态或线程池shutdown但线程工作为空                if (rs < SHUTDOWN ||                    (rs == SHUTDOWN && firstTask == null)) {                    // 线程曾经启动，则抛出非法线程状态异样                    // 为什么会存在这种状态呢？未解决                    if (t.isAlive()) // precheck that t is startable                        throw new IllegalThreadStateException();                    workers.add(w); //退出线程池                    int s = workers.size();                    // 如果当前工作线程数超过线程池已经呈现过的最大线程数，刷新后者值                    if (s > largestPoolSize)                        largestPoolSize = s;                     workerAdded = true;                }            } finally {                mainLock.unlock();  // 开释锁            }            if (workerAdded) { // 工作线程增加胜利，启动该线程                t.start();                workerStarted = true;            }        }    } finally {        //线程启动失败，则进入addWorkerFailed        if (! workerStarted)             addWorkerFailed(w);    }    return workerStarted;}

Worker类

Worker类是外部类，既实现了Runnable，又继承了AbstractQueuedSynchronizer（以下简称AQS），所以其既是一个可执行的工作，又能够达到锁的成果。

Worker类次要保护正在运行工作的线程的中断管制状态，以及其余主要的记录。这个类适时地继承了AbstractQueuedSynchronizer类，以简化获取和开释锁（该锁作用于每个工作执行代码）的过程。这样能够避免去中断正在运行中的工作，只会中断在期待从工作队列中获取工作的线程。

咱们实现了一个简略的不可重入互斥锁，而不是应用可重入锁，因为咱们不心愿工作工作在调用setCorePoolSize之类的池管制办法时可能从新获取锁。另外，为了在线程真正开始运行工作之前禁止中断，咱们将锁状态初始化为负值，并在启动时革除它（在runWorker中）。

private final class Worker    extends AbstractQueuedSynchronizer    implements 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) {        //设置AQS的同步状态        // state：锁状态，-1为初始值，0为unlock状态，1为lock状态        setState(-1); // inhibit interrupts until runWorker  在调用runWorker前，禁止中断               this.firstTask = firstTask;        // 线程工厂创立一个线程        this.thread = getThreadFactory().newThread(this);     }     /** Delegates main run loop to outer runWorker  */    public void run() {        runWorker(this); //runWorker()是ThreadPoolExecutor的办法    }     // Lock methods    // The value 0 represents the unlocked state. 0代表“没被锁定”状态    // The value 1 represents the locked state. 1代表“锁定”状态     protected boolean isHeldExclusively() {        return getState() != 0;    }     /**     * 尝试获取锁的办法     * 重写AQS的tryAcquire()，AQS原本就是让子类来实现的     */    protected boolean tryAcquire(int unused) {        // 判断原值为0，且重置为1，所以state为-1时，锁无奈获取。        // 每次都是0->1，保障了锁的不可重入性        if (compareAndSetState(0, 1)) {            // 设置exclusiveOwnerThread=以后线程            setExclusiveOwnerThread(Thread.currentThread());             return true;        }        return false;    }     /**     * 尝试开释锁     * 不是state-1，而是置为0     */    protected boolean tryRelease(int unused) {        setExclusiveOwnerThread(null);         setState(0);        return true;    }     public void lock()        { acquire(1); }    public boolean tryLock()  { return tryAcquire(1); }    public void unlock()      { release(1); }    public boolean isLocked() { return isHeldExclusively(); }     /**     * 中断（如果运行）     * shutdownNow时会循环对worker线程执行     * 且不须要获取worker锁，即便在worker运行时也能够中断     */    void interruptIfStarted() {        Thread t;        //如果state>=0、t!=null、且t没有被中断        //new Worker()时state==-1，阐明不能中断        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {            try {                t.interrupt();            } catch (SecurityException ignore) {            }        }    }}

runWorker()

能够说，runWorker(Worker w) 是线程池中真正解决工作的办法，后面的execute() 和 addWorker() 都是在为该办法做筹备和铺垫。

「参数阐明：」

  1. 「Worker w」：封装的Worker，携带了工作线程的诸多因素，包含Runnable（待处理工作）、lock（锁）、completedTasks（记录线程池已实现工作数）

「执行流程：」

1、判断当前任务或者从工作队列中获取的工作是否不为空，都为空则进入步骤2，否则进入步骤3

2、工作为空，则将completedAbruptly置为false（即线程不是忽然终止），并执行processWorkerExit(w,completedAbruptly)办法进入线程退出程序

3、工作不为空，则进入循环，并加锁

4、判断是否为线程增加中断标识，以下两个条件满足其一则增加中断标识：

  • 线程池状态>=STOP,即STOP或TERMINATED
  • 一开始判断线程池状态<STOP，接下来查看发现Thread.interrupted()为true，即线程曾经被中断，再次查看线程池状态是否>=STOP（以打消该霎时shutdown办法失效，使线程池处于STOP或TERMINATED）

5、执行前置办法 beforeExecute(wt, task)（该办法为空办法，由子类实现）后执行task.run() 办法执行工作（执行不胜利抛出相应异样）

6、执行后置办法 afterExecute(task, thrown)（该办法为空办法，由子类实现）后将线程池已实现的工作数+1，并开释锁。

7、再次进行循环条件判断。

「流程图：」

线程池流程图

「源码详读：」

final void runWorker(Worker w) {    Thread wt = Thread.currentThread();    Runnable task = w.firstTask;    w.firstTask = null;    // allow interrupts    // new Worker()是state==-1，此处是调用Worker类的tryRelease()办法，将state置为0，而interruptIfStarted()中只有state>=0才容许调用中断    w.unlock();                 // 线程退出的起因，true是工作导致，false是线程失常退出    boolean completedAbruptly = true;     try {        // 当前任务和从工作队列中获取的工作都为空，方进行循环        while (task != null || (task = getTask()) != null) {            //上锁能够避免在shutdown()时终止正在运行的worker，而不是应答并发            w.lock();                          // If pool is stopping, ensure thread is interrupted;            // if not, ensure thread is not interrupted.  This            // requires a recheck in second case to deal with            // shutdownNow race while clearing interrupt            /**             * 判断1：确保只有在线程处于stop状态且wt未中断时，wt才会被设置中断标识             * 条件1：线程池状态>=STOP,即STOP或TERMINATED             * 条件2：一开始判断线程池状态<STOP，接下来查看发现Thread.interrupted()为true，即线程曾经被中断，再次查看线程池状态是否>=STOP（以打消该霎时shutdown办法失效，使线程池处于STOP或TERMINATED），             * 条件1与条件2任意称心一个，且wt不是中断状态，则中断wt，否则进入下一步             */            if ((runStateAtLeast(ctl.get(), STOP) ||                 (Thread.interrupted() &&                  runStateAtLeast(ctl.get(), STOP))) &&                !wt.isInterrupted())                wt.interrupt(); //以后线程调用interrupt()中断                         try {                //执行前（空办法，由子类重写实现）                beforeExecute(wt, task);                                 Throwable thrown = null;                try {                    task.run();                }                 catch (RuntimeException x) {                    thrown = x; throw x;                }                 catch (Error x) {                    thrown = x; throw x;                }                 catch (Throwable x) {                    thrown = x; throw new Error(x);                }                 finally {                    //执行后（空办法，由子类重写实现）                    afterExecute(task, thrown);                 }            }             finally {                task = null;                 w.completedTasks++; //实现工作数+1                w.unlock(); //开释锁            }        }        //         completedAbruptly = false;    }     finally {        //解决worker的退出        processWorkerExit(w, completedAbruptly);    }}

「5、getTask()」

由函数调用关系图可知，在ThreadPoolExecutor类的实现中，Runnable getTask() 办法是为void runWorker(Worker w)办法服务的，它的作用就是在工作队列（workQueue）中获取 task（Runnable）。

「参数阐明」：无参数

「执行流程」：

  1. 将timedOut（上次获取工作是否超时）置为false（首次执行办法，无上次，天然为false），进入一个有限循环
  2. 如果线程池为Shutdown状态且工作队列为空（线程池shutdown状态能够解决工作队列中的工作，不再承受新工作，这个是重点）或者线程池为STOP或TERMINATED状态，则意味着线程池不用再获取工作了，当前工作线程数量-1并返回null，否则进入步骤3
  3. 如果线程池数量超限度或者工夫超限且（工作队列为空或以后线程数>1），则进入步骤4，否则进入步骤5。
  4. 移除工作线程，胜利则返回null，不胜利则进入下轮循环。
  5. 尝试用poll() 或者 take()（具体用哪个取决于timed的值）获取工作，如果工作不为空，则返回该工作。如果为空，则将timeOut 置为 true进入下一轮循环。如果获取工作过程产生异样，则将 timeOut置为 false 后进入下一轮循环。

「流程图」：

线程池流程图

「源码详读：」

private Runnable getTask() {    // 最新一次poll是否超时    boolean timedOut = false; // Did the last poll() time out?    for (;;) {        int c = ctl.get();        int rs = runStateOf(c);        // Check if queue empty only if necessary.        /**         * 条件1：线程池状态SHUTDOWN、STOP、TERMINATED状态         * 条件2：线程池STOP、TERMINATED状态或workQueue为空         * 条件1与条件2同时为true，则workerCount-1，并且返回null         * 注：条件2是思考到SHUTDOWN状态的线程池不会接受任务，但仍会解决工作         */        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {            decrementWorkerCount();            return null;        }        int wc = workerCountOf(c);        // Are workers subject to culling?        /**         * 下列两个条件满足任意一个，则给以后正在尝试获取工作的工作线程设置阻塞工夫限度（超时会被销毁？不太确定这点），否则线程能够始终放弃沉闷状态         * 1.allowCoreThreadTimeOut：以后线程是否以keepAliveTime为超时时限期待工作         * 2.以后线程数量曾经超过了外围线程数         */        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;                    // 两个条件全副为true，则通过CAS使工作线程数-1，即剔除工作线程        // 条件1：工作线程数大于maximumPoolSize，或（工作线程阻塞工夫受限且上次在工作队列拉取工作超时）        // 条件2：wc > 1或工作队列为空        if ((wc > maximumPoolSize || (timed && timedOut))            && (wc > 1 || workQueue.isEmpty())) {            // 移除工作线程，胜利则返回null，不胜利则进入下轮循环            if (compareAndDecrementWorkerCount(c))                return null;            continue;        }     // 执行到这里，阐明曾经通过后面重重校验，开始真正获取task了        try {            // 如果工作线程阻塞工夫受限，则应用poll(),否则应用take()            // poll()设定阻塞工夫，而take()无工夫限度，直到拿到后果为止            Runnable r = timed ?                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                workQueue.take();            // r不为空，则返回该Runnable            if (r != null)                return r;            // 没能获取到Runable，则将最近获取工作是否超时设置为true            timedOut = true;        } catch (InterruptedException retry) {            // 响应中断，进入下一次循环前将最近获取工作超时状态置为false            timedOut = false;        }    }}

processWorkerExit()

processWorkerExit(Worker w, boolean completedAbruptly)执行线程退出的办法

「参数阐明：」

  1. 「Worker w」：要完结的工作线程。
  2. 「boolean completedAbruptly」：是否忽然实现（异样导致），如果工作线程因为用户异样死亡，则completedAbruptly参数为 true。

「执行流程：」

1、如果 completedAbruptly 为 true，即工作线程因为异样忽然死亡，则执行工作线程-1操作。

2、主线程获取锁后，线程池曾经实现的工作数追加 w（当前工作线程） 实现的工作数，并从worker的set汇合中移除以后worker。

3、依据线程池状态进行判断是否执行tryTerminate()完结线程池。

4、是否须要减少工作线程，如果线程池还没有齐全终止，仍须要放弃肯定数量的线程。

  • 如果以后线程是忽然终止的，调用addWorker()创立工作线程
  • 以后线程不是忽然终止，但当前工作线程数量小于线程池须要保护的线程数量，则创立工作线程。须要保护的线程数量为corePoolSize（取决于成员变量 allowCoreThreadTimeOut是否为 false）或1。
  • 源码详读：
/** * Performs cleanup and bookkeeping for a dying worker. Called * only from worker threads. Unless completedAbruptly is set, * assumes that workerCount has already been adjusted to account * for exit.  This method removes thread from worker set, and * possibly terminates the pool or replaces the worker if either * it exited due to user task exception or if fewer than * corePoolSize workers are running or queue is non-empty but * there are no workers. * * @param w the worker * @param completedAbruptly if the worker died due to user exception */private void processWorkerExit(Worker w, boolean completedAbruptly) {    /**     * 1.工作线程-1操作     * 1）如果completedAbruptly 为true，阐明工作线程产生异样，那么将正在工作的线程数量-1     * 2）如果completedAbruptly 为false，阐明工作线程无工作能够执行，由getTask()执行worker-1操作     */    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted        decrementWorkerCount();    // 2.从线程set汇合中移除工作线程，该过程须要加锁    final ReentrantLock mainLock = this.mainLock;    mainLock.lock();    try {        // 将该worker已实现的工作数追加到线程池已实现的工作数        completedTaskCount += w.completedTasks;        // HashSet<Worker>中移除该worker        workers.remove(w);    } finally {        mainLock.unlock();    }     // 3.依据线程池状态进行判断是否完结线程池    tryTerminate();  /**     * 4.是否须要减少工作线程     * 线程池状态是running 或 shutdown     * 如果以后线程是忽然终止的，addWorker()     * 如果以后线程不是忽然终止的，但以后线程数量 < 要保护的线程数量，addWorker()     * 故如果调用线程池shutdown()，直到workQueue为空前，线程池都会维持corePoolSize个线程，而后再逐步销毁这corePoolSize个线程     */    int c = ctl.get();    if (runStateLessThan(c, STOP)) {       if (!completedAbruptly) {            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;            if (min == 0 && ! workQueue.isEmpty())                min = 1;            if (workerCountOf(c) >= min)                return; // replacement not needed        }        addWorker(null, false);    }}

好啦，以上就是Java线程池的全部内容啦，保持读完的伙伴儿们你们播种如何？感觉有帮忙的就棘手点个赞吧，祝大家新年新气象，降级加薪！