是时候揭开线程池这块面纱了（二）

线程池状态

定义

在简单介绍完线程池的基本概念后，这篇文章我们来分析一下，线程池它具体是如何工作的。首先我们看一下线程池的状态。线程池的设计是这样子的，它将线程池的状态和目前工作线程的线程数用一个变量来进行了存储，就是ctl变量

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

首先这是一个原子类，int类型，32位，前3位是用来标识线程池的状态的，后29位是用来表示线程池当中工作的线程数的。

通过使用原子类，这样就能保证线程池的状态和工作线程数的原子性了。那么，我们当前线程池的状态常量是如何标识的呢，以及当我们想要查询线程池的工作线程数，是怎样运算的呢。

private static final int COUNT_BITS = Integer.SIZE - 3;		// 29位用来存工作线程数
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;	// 所能存放工作线程的最大数 000 1111...

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;		// 运行状态 补码形式为：1110 00……
private static final int SHUTDOWN   =  0 << COUNT_BITS;		// 关闭状态 0000 ....
private static final int STOP       =  1 << COUNT_BITS; 	// 停止状态 0010 00……
private static final int TIDYING    =  2 << COUNT_BITS;		// 清理状态 0100 00……
private static final int TERMINATED =  3 << COUNT_BITS;		// 终止状态 0110 00……

// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~CAPACITY; }  // 后28位取1 按位与操作
private static int workerCountOf(int c)  { return c & CAPACITY; }   // 后29为取0 与运算
private static int ctlOf(int rs, int wc) { return rs | wc; }        // 根据状态和工作数生成ctl

private static boolean runStateLessThan(int c, int s) {	  return c < s;}
private static boolean runStateAtLeast(int c, int s) {  return c >= s;}
private static boolean isRunning(int c) {  return c < SHUTDOWN;}

1. 线程池有5种状态，按小到大的顺序如下：RUNNING<SHUTDOWN<STOP<TIDYING<TERMINATED
2. 使用原子类来存储两个变量，保证了共享变量的原子性，以及它按照位操作提升了很大的性能

状态解析

RUNNING：in operation 运行，当线程池初始化的时候，线程池就处于RUNNING状态，处于该状态时，线程池可以接受任务，并且处理缓冲的任务。

SHUTDOWN：turn off 关闭，当调用shutdown()方法时，线程池从RUNING转化为SHUTDOWN状态，该状态的线程池不接受任务，只处理缓冲的任务

STOP：stop 停止，当调用shutdownNow()方法时，会将（RUNING/SHUTDOWN状态）转换成该状态，处于该状态的线程池不接受任务，不处理缓冲任务，并且中断全部正在处理的任务interrupt running tasks

TIDYING：清理，当任务中断了，并且线程工作数为0，就会进入清理状态，清理现场。

When the thread pool is in the SHUTDOWN state, the blocking queue is empty and the tasks executed in the thread pool are also empty, it will be SHUTDOWN - > tidying. When the task executed in the thread pool is empty in the STOP state, it will be STOP - > tidying.

TERMINATED：终止，结束，当terminated()方法执行完后，就会进入该状态

When the thread pool terminates completely, it becomes terminated. When the thread pool is in tidying state, after executing terminated(), it will be tidying - > terminated.

状态转化

* RUNNING -> SHUTDOWN
*    On invocation of shutdown(), perhaps implicitly in finalize()
* (RUNNING or SHUTDOWN) -> STOP
*    On invocation of shutdownNow()
* SHUTDOWN -> TIDYING
*    When both queue and pool are empty
* STOP -> TIDYING
*    When pool is empty
* TIDYING -> TERMINATED
*    When the terminated() hook method has completed
*

通过介绍我们知道，线程池初始化时处于RUNNING状态，当线程池调用shutdown()方法时，线程池就关闭了，只处理当前已经进来的任务。当任务处理完之后，线程池会调用terminated()方法，执行完后线程池进入TERMINATED状态。当然还有另一种方式，不管线程池有没有调用shutdown()方法，如果调用了shutdownNow()，会中断正在执行的线程，当队列和线程池线程为空时，线程池就会进入stop状态。当池中线程为空时，terminated()方法执行完后进入TERMINATED状态。

构造方法

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();	// 主线程不能小于0
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();   // 队列和线程工厂、策略不能为null
    this.acc = System.getSecurityManager() == null ?null :AccessController.getContext();
   
  
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);//将时间传入的unit和keepAliveTime转换为ns
    this.threadFactory = threadFactory;
    this.handler = handler;
}

提交执行task的过程

​

解释

当提交一个新任务到线程池时，线程池的处理流程如下：

1. 线程池判断核心线程数是否都在执行任务。如果不是，则创建一个新的工作线程来执行任务。如果是，则进入步骤2。
2. 线程池判断工作队列是否已满。如果没有满，将新提交的任务存储在工作队列中。如果满了，则进入步骤3。
3. 线程池判断线程池中线程是否都处于工作状态（已满）。如果没有，则创建一个新的工作线程。如果满了，则交给饱和策略来处理这个任务。

execute方法

/**
 *  如果任务不能提交，要么是因为当前线程池关闭了，要么是因为队列满了需要触发饱和策略
 */
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
    int c = ctl.get();
  	// worker数量比核心线程数小，直接创建worker执行任务
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // worker数量超过核心线程数，任务直接进入队列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 线程池状态不是RUNNING状态，说明执行过shutdown,需要对新加入的任务做reject操作
      	// 这儿为什么需要recheck,是因为任务入队前后，线程池的状态可能发生变化。但是一般检查一下，并没啥用。
        if (! isRunning(recheck) && remove(command))
            reject(command);
      	// 这儿为什么需要判断0值，是因为线程池构造方法中，允许线程池的核心数为0，如果遇上SynchronousQueue类型的队列，我也能创建线程去消费，不会一直阻塞
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 如果线程池不是运行状态，或者任务进入队列失败，则会尝试创建worker执行任务
    // 注意3点
    // 1. 线程池不是运行状态时，addWorker方法内部会去判断线程池状态
    // 2. addWorker的第二个参数表示是否创建核心线程
    // 3. addWorker方法返回false，说明任务执行失败，需要执行reject操作
    else if (!addWorker(command, false))
        reject(command);
}

execute方法本身来说并不难，我们来这样子分析。

第一步，传入来的command不能为空

第二步，取到当前ctl，拿到当前工作线程数与核心数比较，如果小于，直接创建worker执行任务，否则进入下一步。

注意，addWorkerf方法的第二个参数表示是否核心线程。方法本身会去判断当前线程池状态和数量的，如果addWork失败，方法会返回false，表示添加任务失败，在重新获取一下ctl，用于下一步

第三步，走到这一步说明，不是核心数满了，就是线程池shutdown了。看一下当前线程池状态，如果是运行状态，就尝试往队列中放任务，如果放成功了返回true，放任务前后都得去拿一下当前的ctl，再去判断一下线程池是否shutdown了，如果shutdown了，就把任务从队列中remove掉。如果没有shutdown，会去判断当前工作线程数。

如果为0，我就启动一个线程，并且这里传入的firstTask还是null，这是为什么呢?

第四步，这一步相当重要，走到这一步，说明线程池不是运行状态了，或者任务进入队列失败了，为了消费任务，会尝试创建（非核心）worker执行任务，如果创建失败，则要采取饱和策略，执行reject操作。

addWorker源码解析

干了两件事情

1. 使用CAS算法将workerCount加一

2. 将任务装入到worker中，并且将work添加到工作线程集合workers中,开始执行任务

   worker是一个任务单元，继承了AQS类，本身是把锁，而且构造方法是RUNNALE任务，因此可以使用start方法执行线程中的run方法

   如果线程启动失败，将这个worker从当前线程池works集合中删除，并且将workerCount减一

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
  	// 外层自旋
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&						// 如果状态>SHUTDOWN,返回false
            ! (rs == SHUTDOWN &&				// 如果状态=SHUTDOWN，且task!=null,返回false
               firstTask == null &&			// 如果状态=SHUTDOWN，且队列为空，返回false
               ! workQueue.isEmpty()))
            return false;

      	// 内层自旋
        for (;;) {
            int wc = workerCountOf(c);
          	// 判断工作线程数，是否超过最大值和标准值
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
          	// 使用一次CAS算法，使得c++,就是wc++
            if (compareAndIncrementWorkerCount(c))
              	// 如果成功，跳出外部循环进入第二部分
                break retry;
          	// 如果失败，应该是有线程把ctl改了，比较一下
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
              	// 不等于的话进入外部循环自旋，重新获取ctl
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop（等于的话，内部自旋）
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
          	// worker的添加必须是串行的，必须得加锁
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
              	// 重新检查线程池状态
                int rs = runStateOf(ctl.get());
								
                //	如果处于RUNNING状态或者SHUTDOWN时传递的任务为空
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                  	// 如果worker已经调用过start方法了，则不再创建了
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                  	// 创建并且加入到works成功
                    workers.add(w);
                  	// 更新largestPoolSize变量
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            // 启动worker线程
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
      	// worker启动失败，说明线程状态发生了变化（关闭操作被执行），需要执行shutdown相关操作
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

worker任务单元

worker是一个任务单元，他里面包含了一个线程，和一个任务，然后还记录了我这个worker干过多少个任务等等。

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) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
      	// 这里使用了我们之前定义的线程工厂，传入的参数我当前的worker
        this.thread = getThreadFactory().newThread(this);
    }

    /** Delegates main run loop to outer runWorker  */
    public void run() {
      	// run方法的实现在runWorker里面
        runWorker(this);
    }
    
   //省略代码……
}

runworker核心线程执行逻辑

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;		// for GC
  	// 允许外部中断
    w.unlock(); // allow interrupts
    // 用于判读是否进入过自旋
    boolean completedAbruptly = true;
    try {
        // 自旋
      	// 如果task不为空，则开始执行task
        // 如果不为空，则从队列中获取task开始执行
        // 阻塞队列的特性：就是当队列为空的时候，当前线程会被阻塞
        while (task != null || (task = getTask()) != null) {
            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
          	// 如果线程池正在停止，则对当前线程中断
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
              	// beforeExecute和afterExecute 是为扩展功能用的，在当前类实现默认为空
                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;   // for GC
                w.completedTasks++;  // 已完成任务数加一
                w.unlock();
            }
        }
        completedAbruptly = false;  //表示已经自旋完成
    } finally {
      	// 自旋操作被退出，说明线程池正在结束，要把wc--，并且从workers删除当前worker
        processWorkerExit(w, completedAbruptly);
    }
}

这里相当于用到了一种装饰者模式，将一个task包装成一个带有锁特性的执行者。这个执行者不仅可以消费当前的任务，getTask 还能从阻塞队列中获取任务消费，当队列为空的时候，线程会被阻塞住。当有任务过来的时候，又可以执行了。当任务执行完之后，processWorkerExit 方法是线程worker的出口。他会处理wc和workers集合，并且尝试将当前线程中断。

参考：

Deep understanding of thread pool ThreadPool Executor

Do you know the underlying implementation of thread pools?Thread Pool Module Summary

能说说什么是线程吗？线程模块总结