深入理解Java线程池

Java中开辟出了一种管理线程的概念，这个概念叫做线程池，从概念以及应用场景中，我们可以看出，线程池的好处，就是可以方便的管理线程，也可以减少内存的消耗。

线程池的概念是Executor这个接口，具体实现为ThreadPoolExecutor类，JDK也提供了线程池的工具类Executors可以方便我们创建线程池（不推荐使用）。

Executors

JDK 的 Executors 类中提供了几种具有代表性的线程池，这些线程池 都是基于 ThreadPoolExecutor 的定制化实现。

newSingleThreadExecutor

创建一个单线程的线程池，只会创建唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。 如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它 。

单工作线程最大的特点是：可保证顺序地执行各个任务。

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

注意：LinkedBlockingQueue没有设置大小，默认为Integer.MAX_VALUE，可能导致内存溢出。

newFixedThreadPool

创建一个固定大小的线程池。每次提交一个任务就会新创建一个工作线程，如果工作线程数量达到线程池最大线程数，则将提交的任务存入到阻塞队列中。

FixedThreadPool 是一个典型且优秀的线程池，它具有线程池提高程序效率和节省创建线程时所耗的开销的点。但是，在线程池空闲时，即线程池中没有可运行任务时，它不会释放工作线程，还会占用一定的系统资源。

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

注意：LinkedBlockingQueue没有设置大小，默认为Integer.MAX_VALUE，可能导致内存溢出。

newCachedThreadPool

创建一个可缓存的线程池。

• 如果线程池长度超过处理任务所需要的线程数，就会回收部分空闲的线程；
• 如果长时间没有往线程池中提交任务，即如果工作线程空闲了指定的时间（默认为 1 分钟），则该工作线程将自动终止。终止后，如果你又提交了新的任务，则线程池重新创建一个工作线程。
• 此线程池不会对线程池大小做限制，线程池大小完全依赖于操作系统（或者说 JVM）能够创建的最大线程大小。 因此，使用 CachedThreadPool时，一定要注意控制任务的数量，否则，由于大量线程同时运行，很有会造成系统瘫痪。

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

注意：线程池最大为Integer.MAX_VALUE，容易导致内存溢出。

newScheduleThreadPool

创建一个定长的线程池。此线程池支持定时以及周期性执行任务的需求。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

注意：线程池最大为Integer.MAX_VALUE，但是因为DelayedWorkQueue是无界队列，所以实际上最大线程数是corePoolSize，但是也因为DelayedWorkQueue是无界队列，所以容易导致内存溢出。

newSingleThreadScheduledExecutor

创建一个单线程的线程池。此线程池支持定时以及周期性执行任务的需求。实际上调用了newScheduleThreadPool的构造方法。

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

注意：因为newScheduleThreadPool使用了无界队列，所以也会导致内存溢出。

执行流程

注意：如果使用的是无界队列，那最大线程数maximumPoolSize则无效。

核心参数

ThreadPoolExecutor继承了AbstractExecutorService类，并提供了四个构造器，事实上，发现前面三个构造器最终都是调用的第四个构造器进行的初始化工作。

/**
 * Creates a new {@code ThreadPoolExecutor} with the given initial
 * parameters.
 *
 * @param corePoolSize the number of threads to keep in the pool, even
 *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
 * @param maximumPoolSize the maximum number of threads to allow in the
 *        pool
 * @param keepAliveTime when the number of threads is greater than
 *        the core, this is the maximum time that excess idle threads
 *        will wait for new tasks before terminating.
 * @param unit the time unit for the {@code keepAliveTime} argument
 * @param workQueue the queue to use for holding tasks before they are
 *        executed.  This queue will hold only the {@code Runnable}
 *        tasks submitted by the {@code execute} method.
 * @param threadFactory the factory to use when the executor
 *        creates a new thread
 * @param handler the handler to use when execution is blocked
 *        because the thread bounds and queue capacities are reached
 * @throws IllegalArgumentException if one of the following holds:<br>
 *         {@code corePoolSize < 0}<br>
 *         {@code keepAliveTime < 0}<br>
 *         {@code maximumPoolSize <= 0}<br>
 *         {@code maximumPoolSize < corePoolSize}
 * @throws NullPointerException if {@code workQueue}
 *         or {@code threadFactory} or {@code handler} is null
 */
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;
}

corePoolSize

核心线程池的大小，这个参数跟线程池的实现原理有非常大的关系。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这2个方法的名字就可以看出，是预创建线程的意思，即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到阻塞队列当中。

合理设置线程池大小

经验估计：

1. CPU 密集型应用，线程池大小设置为 N + 1
2. IO 密集型应用，线程池大小设置为 2N

精准计算：

最佳线程数目 = （（线程等待时间+线程CPU时间）/线程CPU时间 ）* CPU数目

因为很显然，线程等待时间所占比例越高，需要越多线程。线程CPU时间所占比例越高，需要越少线程。

下面举个例子：比如平均每个线程CPU运行时间为0.5s，而线程等待时间（非CPU运行时间，比如IO）为1.5s，CPU核心数为8，那么根据上面这个公式估算得到：((0.5+1.5)/0.5)*8=32。

maximumPoolSize

线程池最大线程数，它表示在线程池中最多能创建多少个线程

keepAliveTime

表示线程没有任务执行时最多保持多久时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时keepAliveTime参数也会起作用，直到线程池中的线程数为0。

unit

参数keepAliveTime的时间单位，有7种取值，在TimeUnit类中有7种静态属性：

TimeUnit.DAYS;              // 天
TimeUnit.HOURS;             // 小时
TimeUnit.MINUTES;           // 分钟
TimeUnit.SECONDS;           // 秒
TimeUnit.MILLISECONDS;      // 毫秒
TimeUnit.MICROSECONDS;      // 微妙
TimeUnit.NANOSECONDS;       // 纳秒

workQueue

一个阻塞队列，用来存储等待执行的任务。这个参数的选择也很重要，会对线程池的运行过程产生重大影响。

ArrayBlockingQueue // 一个由数组结构组成的有界阻塞队列。
LinkedBlockingQueue // 一个由链表结构组成的有界（但大小默认值为Interger.MAX_VALUE）阻塞队列。
PriorityBlockingQueue // 一个支持优先级排序的无界阻塞队列。
DelayQueue // 一个使用优先级队列实现的无界阻塞队列。
SynchronousQueue // 一个不存储元素的阻塞队列，也即单个元素的阻塞队列。
LinkedTransferQueue // 一个由链表结构组成的无界阻塞队列
LinkedBlockingDeque // 一个由链表结构组成的双向有界阻塞队列

注意：使用阻塞队列最好定义队列大小，防止出现OOM。

threadFactory

线程工厂，主要用来创建线程，可以自定义线程名字，使用示例：Executors.defaultThreadFactory()，cn.hutool.core.thread.NamedThreadFactory

handler

拒绝策略，表示当拒绝处理任务时的策略，有以下四种取值：

ThreadPoolExecutor.AbortPolicy       // 默认策略，丢弃任务并抛出RejectedExecutionException异常
ThreadPoolExecutor.DiscardPolicy       // 也是丢弃任务，但是不抛出异常
ThreadPoolExecutor.DiscardOldestPolicy // 丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy    // 由调用线程处理该任务

关键成员变量

// 用来标记线程池状态和线程个数（高3位表示线程池状态，低29位表示线程个数）
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 线程个数掩码，一般int是32位，32-3=29表示ctl低29用来表示线程个数
private static final int COUNT_BITS = Integer.SIZE - 3;
// 线程最大个数
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS; // 101，后面补29个0
private static final int SHUTDOWN   =  0 << COUNT_BITS; // 000，后面补29个0
private static final int STOP       =  1 << COUNT_BITS; // 001，后面补29个0
private static final int TIDYING    =  2 << COUNT_BITS; // 010，后面补29个0
private static final int TERMINATED =  3 << COUNT_BITS; // 011，后面补29个0

// 阻塞队列，用来存放等待执行的任务
private final BlockingQueue<Runnable> workQueue;
// 线程池的主要状态锁，对线程池状态（比如线程池大小、runState等）的改变都要使用这个锁
private final ReentrantLock mainLock = new ReentrantLock();
// 用来存放工作集
private final HashSet<Worker> workers = new HashSet<Worker>();
// 线程存活时间
private volatile long keepAliveTime;   
// 是否允许为核心线程设置存活时间
private volatile boolean allowCoreThreadTimeOut;
// 核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）
private volatile int corePoolSize;
// 线程池最大能容忍的线程数
private volatile int maximumPoolSize;
// 线程池中当前的线程数
private volatile int poolSize;
// 任务拒绝策略
private volatile RejectedExecutionHandler handler;
// 线程工厂，用来创建线程
private volatile ThreadFactory threadFactory;
// 用来记录线程池中曾经出现过的最大线程数
private int largestPoolSize;
// 用来记录已经执行完毕的任务个数
private long completedTaskCount;

线程池状态含义

1. RUNNING：接受新任务并且处理阻塞队列里的任务；

2. SHUTDOWN：拒绝新任务但是处理阻塞队列里的任务；

3. STOP：拒绝新任务并且抛弃阻塞队列里的任务，同时会中断正在处理的任务；

4. TIDYING：所有任务都执行完（包含阻塞队列里面任务）当前线程池活动线程为 0，将要调用 terminated方法；

5. TERMINATED：终止状态，terminated方法调用完成以后的状态。

线程池状态转换

• RUNNING -> SHUTDOWN：显式调用 shutdown() 方法，或者隐式调用了 finalize()，它里面调用了 shutdown() 方法。

• RUNNING or SHUTDOWN -> STOP：显式调用 shutdownNow() 方法时候。

• SHUTDOWN -> TIDYING：当线程池和任务队列都为空的时候。

• STOP -> TIDYING：当线程池为空的时候。

• TIDYING -> TERMINATED：执行tryTerminate()方法，当 terminated()方法（实际上什么都没干）执行完成后。

execute方法

/**
 * Executes the given task sometime in the future.  The task
 * may execute in a new thread or in an existing pooled thread.
 *
 * If the task cannot be submitted for execution, either because this
 * executor has been shutdown or because its capacity has been reached,
 * the task is handled by the current {@code RejectedExecutionHandler}.
 *
 * @param command the task to execute
 * @throws RejectedExecutionException at discretion of
 *         {@code RejectedExecutionHandler}, if the task
 *         cannot be accepted for execution
 * @throws NullPointerException if {@code command} is null
 */
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();
    // 如果当前线程池线程个数是否小于corePoolSize,小于则开启新线程运行
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        // 
        c = ctl.get();
    }
    // 如果线程池处于RUNNING状态，则添加任务到阻塞队列
    if (isRunning(c) && workQueue.offer(command)) {
        // 重新检查
        int recheck = ctl.get();
        // 如果当前线程池状态不是RUNNING则从队列删除任务，并执行拒绝策略
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 否者如果当前线程池线程空，则添加一个线程
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 如果队列满了，则新增线程，新增失败则执行拒绝策略
    else if (!addWorker(command, false))
        reject(command);
}

addWorker方法

/**
 * Checks if a new worker can be added with respect to current
 * pool state and the given bound (either core or maximum). If so,
 * the worker count is adjusted accordingly, and, if possible, a
 * new worker is created and started, running firstTask as its
 * first task. This method returns false if the pool is stopped or
 * eligible to shut down. It also returns false if the thread
 * factory fails to create a thread when asked.  If the thread
 * creation fails, either due to the thread factory returning
 * null, or due to an exception (typically OutOfMemoryError in
 * Thread.start()), we roll back cleanly.
 *
 * @param firstTask the task the new thread should run first (or
 * null if none). Workers are created with an initial first task
 * (in method execute()) to bypass queuing when there are fewer
 * than corePoolSize threads (in which case we always start one),
 * or when the queue is full (in which case we must bypass queue).
 * Initially idle threads are usually created via
 * prestartCoreThread or to replace other dying workers.
 *
 * @param core if true use corePoolSize as bound, else
 * maximumPoolSize. (A boolean indicator is used here rather than a
 * value to ensure reads of fresh values after checking other pool
 * state).
 * @return true if successful
 */
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 &&
            ! (rs == SHUTDOWN &&
                firstTask == null &&
                ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            // 如果线程个数超过限制则返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // cas 增加线程个数
            if (compareAndIncrementWorkerCount(c))
                break retry;
            // cas 失败则检查线程池状态，如果状态发生了改变，则跳到最外层循环，否则内层循环重新进行cas
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    // 到这里说明CAS增加线程个数成功
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 创建Worker
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            // 加独占锁，为了workers同步，因为可能多个线程调用了线程池的execute方法
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                // 重新检查线程池状态，为了避免在获取锁前调用了shutdown接口
                int rs = runStateOf(ctl.get());

                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 {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}
// 添加失败，则移除Worker
private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        if (w != null)
            workers.remove(w);
        decrementWorkerCount();
        tryTerminate();
    } finally {
        mainLock.unlock();
    }
}
// cas 减小线程个数
private void decrementWorkerCount() {
    do {} while (! compareAndDecrementWorkerCount(ctl.get()));
}

上面的

if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
                firstTask == null &&
                ! workQueue.isEmpty()))
            return false;

在下面几种情况下会返回 false：

1. 当前线程池状态为 STOP，TIDYING，TERMINATED
2. 当前线程池状态为 SHUTDOWN 并且已经有了第一个任务
3. 当前线程池状态为 SHUTDOWN 并且任务队列为空

Worker

使用了线程工厂创建线程，Worker类实现了Runnable接口

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    
    Worker(Runnable firstTask) {
        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);
    }
}
public ThreadFactory getThreadFactory() {
    return threadFactory;
}

构造函数内首先设置 Worker 的状态为 -1，是为了避免当前 worker 在调用 runWorker 方法前被中断（当其它线程调用了线程池的 shutdownNow 时候，如果 worker 状态 >= 0 则会中断该线程）。这里设置了线程的状态为 -1，所以该线程就不会被中断了。运行 runWorker 的代码时候会调用 unlock 方法，该方法把 status 变为了 0，所以这时候调用 shutdownNow 就会中断 worker 线程了。

runWorker方法

如果task为空，则往阻塞队列中获取

/**
 * Main worker run loop.  Repeatedly gets tasks from queue and
 * executes them, while coping with a number of issues:
 *
 * 1. We may start out with an initial task, in which case we
 * don't need to get the first one. Otherwise, as long as pool is
 * running, we get tasks from getTask. If it returns null then the
 * worker exits due to changed pool state or configuration
 * parameters.  Other exits result from exception throws in
 * external code, in which case completedAbruptly holds, which
 * usually leads processWorkerExit to replace this thread.
 *
 * 2. Before running any task, the lock is acquired to prevent
 * other pool interrupts while the task is executing, and then we
 * ensure that unless pool is stopping, this thread does not have
 * its interrupt set.
 *
 * 3. Each task run is preceded by a call to beforeExecute, which
 * might throw an exception, in which case we cause thread to die
 * (breaking loop with completedAbruptly true) without processing
 * the task.
 *
 * 4. Assuming beforeExecute completes normally, we run the task,
 * gathering any of its thrown exceptions to send to afterExecute.
 * We separately handle RuntimeException, Error (both of which the
 * specs guarantee that we trap) and arbitrary Throwables.
 * Because we cannot rethrow Throwables within Runnable.run, we
 * wrap them within Errors on the way out (to the thread's
 * UncaughtExceptionHandler).  Any thrown exception also
 * conservatively causes thread to die.
 *
 * 5. After task.run completes, we call afterExecute, which may
 * also throw an exception, which will also cause thread to
 * die. According to JLS Sec 14.20, this exception is the one that
 * will be in effect even if task.run throws.
 *
 * The net effect of the exception mechanics is that afterExecute
 * and the thread's UncaughtExceptionHandler have as accurate
 * information as we can provide about any problems encountered by
 * user code.
 *
 * @param w the worker
 */
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts（将Worker状态设置为0，运行中断）
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            // 加锁，将state修改为1
            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(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;
                // 统计当前Worker完成多少任务
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 执行清理任务
        processWorkerExit(w, completedAbruptly);
    }
}

下面这段代码体现了线程池的复用机制

while (task != null || (task = getTask()) != null) {
    ...
}

processWorkerExit方法

/**
 * 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) {
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 统计整个线程池完成的任务个数,并从工作集里面删除当前woker
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    // 尝试设置线程池状态为TERMINATED，如果当前是shutdonw状态并且工作队列为空
    // 或者当前是stop状态当前线程池里面没有活动线程则设置线程池状态为 TERMINATED
    // 如果设置为了 TERMINATED 状态还需要调用条件变量 termination 的 signalAll() 方法
    // 激活所有因为调用线程池的 awaitTermination 方法而被阻塞的线程
    tryTerminate();

    // 如果当前线程个数小于核心个数，则新增一个线程
    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);
    }
}

getTask方法

如果getTask方法返回null，那么这个Worker线程则会被删除（processWorkerExit方法）

以下几种情况，getTask方法会返回NULL：

1. 当前线程数超过了最大线程数（可能重新调用了setMaximumPoolSize方法）
2. 线程池状态为大于STOP（状态为STOP、TIDYING、TERMINATED）
3. 线程池状态为SHUTDOWN，阻塞队列为空
4. 等待超时

注意：这里不会判断该线程是不是核心线程。线程池中的线程都是放在HashSet<Worker> workers中，只有创建前会标识它是否为核心线程，但是创建后没有存储该标识，所以删除时根据线程的执行情况删除（先获取到空的任务的线程被删除）。

/**
 * Performs blocking or timed wait for a task, depending on
 * current configuration settings, or returns null if this worker
 * must exit because of any of:
 * 1. There are more than maximumPoolSize workers (due to
 *    a call to setMaximumPoolSize).
 * 2. The pool is stopped.
 * 3. The pool is shutdown and the queue is empty.
 * 4. This worker timed out waiting for a task, and timed-out
 *    workers are subject to termination (that is,
 *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
 *    both before and after the timed wait, and if the queue is
 *    non-empty, this worker is not the last thread in the pool.
 *
 * @return task, or null if the worker must exit, in which case
 *         workerCount is decremented
 */
private Runnable getTask() {
    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.
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

shutdown方法

调用 shutdown() 后，线程池就不会在接受新的任务了，但是工作队列里面的任务还是要执行的，该方法立刻返回的，并不等待队列任务完成在返回。

/**
 * Initiates an orderly shutdown in which previously submitted
 * tasks are executed, but no new tasks will be accepted.
 * Invocation has no additional effect if already shut down.
 *
 * <p>This method does not wait for previously submitted tasks to
 * complete execution.  Use {@link #awaitTermination awaitTermination}
 * to do that.
 *
 * @throws SecurityException {@inheritDoc}
 */
public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 检查权限
        checkShutdownAccess();
        // 设置当前线程池状态为SHUTDOWN，如果已经是SHUTDOWN则直接返回
        advanceRunState(SHUTDOWN);
        // 设置中断标志
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    // 尝试状态变为TERMINATED
    tryTerminate();
}

private void advanceRunState(int targetState) {
    for (;;) {
        int c = ctl.get();
        // 如果当前状态 >= targetState 则直接返回，否则CAS设置当前状态为 targetState
        if (runStateAtLeast(c, targetState) ||
            ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
            break;
    }
}

如果设置了安全管理器，则看当前调用 shutdown 命令的线程是否有关闭线程的权限，如果有权限则还要看调用线程是否有中断工作线程的权限，如果没有权限则抛出 SecurityException 或者 NullPointerException 异常。

interruptIdleWorkers方法

设置所有空闲线程的中断标志，这里首先加了全局锁，同时只有一个线程可以调用 shutdown 设置中断标志，然后尝试获取 worker 自己的锁，获取成功则设置中断标识，由于正在执行的任务已经获取了锁，所以正在执行的任务没有被中断。这里中断的是阻塞到 getTask() 方法，企图从队列里面获取任务的线程，也就是空闲线程。

private void interruptIdleWorkers() {
    interruptIdleWorkers(false);
}
private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            // 如果工作线程没有被中断，并且没有正在运行则设置设置中断
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

shutdownnow方法

调用 shutdownNow 后，线程池就不会在接受新的任务了，并且丢弃工作队列里面里面的任务，正在执行的任务会被中断，该方法是立刻返回的，并不等待激活的任务执行完成再返回。返回值为这时候队列里面被丢弃的任务列表。

/**
 * Attempts to stop all actively executing tasks, halts the
 * processing of waiting tasks, and returns a list of the tasks
 * that were awaiting execution. These tasks are drained (removed)
 * from the task queue upon return from this method.
 *
 * <p>This method does not wait for actively executing tasks to
 * terminate.  Use {@link #awaitTermination awaitTermination} to
 * do that.
 *
 * <p>There are no guarantees beyond best-effort attempts to stop
 * processing actively executing tasks.  This implementation
 * cancels tasks via {@link Thread#interrupt}, so any task that
 * fails to respond to interrupts may never terminate.
 *
 * @throws SecurityException {@inheritDoc}
 */
public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 权限检查
        checkShutdownAccess();
        // 设置线程池状态为stop
        advanceRunState(STOP);
        // 中断所有线程（包含空闲线程和正在执行任务的线程）
        interruptWorkers();
        // 移动队列任务到tasks
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

interruptWorkers方法

中断所有的线程，包含空闲线程和正在执行任务的线程

/**
 * Interrupts all threads, even if active. Ignores SecurityExceptions
 * (in which case some threads may remain uninterrupted).
 */
private void interruptWorkers() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers)
            w.interruptIfStarted();
    } finally {
        mainLock.unlock();
    }
}
void interruptIfStarted() {
    Thread t;
    // state >=0 才需要中断，而state状态启动之前是-1，启动后是0，运行前加锁修改为1
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        try {
            t.interrupt();
        } catch (SecurityException ignore) {
        }
    }
}

awaitTermination方法

当线程调用 awaitTermination 方法后，当前线程会被阻塞，知道线程池状态变为了 TERMINATED 才返回，或者等待时间超时才返回。

public boolean awaitTermination(long timeout, TimeUnit unit)
    throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (;;) {
            if (runStateAtLeast(ctl.get(), TERMINATED))
                return true;
            if (nanos <= 0)
                return false;
            nanos = termination.awaitNanos(nanos);
        }
    } finally {
        mainLock.unlock();
    }
}

上述判断当前线程池状态是否至少是 TERMINATED 状态，如果是则直接返回。否则说明当前线程池里面还有线程在执行，则看设置的超时时间 nanos 是否小于 0，小于 0 则说明不需要等待，则直接返回；如果大于0则调用条件变量 termination 的 awaitNanos 方法等待 nanos 时间，期望在这段时间内线程池状态内变为 TERMINATED 状态。

processWorkerExit方法时候提到当线程池状态变为 TERMINATED 后，会调用 termination.signalAll() 用来激活调用条件变量 termination 的 await 系列方法被阻塞的所有线程，所以如果在调用了 awaitTermination 之后调用了 shutdown 方法，并且 shutdown 内部设置线程池状态为 TERMINATED 了，则 termination.awaitNanos 方法会返回。

tryTerminate方法

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    // 激活调用条件变量termination的await系列方法被阻塞的所有线程
                    termination.signalAll(); 
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

注意事项

创建线程池时候要指定与业务相关的名字，以便于追溯问题

自定义ThreadFactory，可参考Executors.defaultThreadFactory()，cn.hutool.core.thread.NamedThreadFactory

线程池中使用 ThreadLocal 导致内存泄露

线程池里面设置了 ThreadLocal 变量一定要记得及时清理，因为线程池里面的核心线程是一直存在的，如果不清理，那么线程池的核心线程的 threadLocals 变量一直会持有 ThreadLocal 变量。