一文搞懂线程池的底层原理

线程池

较稀缺的资源，在应用程序中大量的创建和销毁线程是非常消耗资源的，在并发的情况下对性能有很大的影响。而使用线程池则可以减少创建和销毁线程的次数，对每个线程可以进行复用，提供使用效率，减少资源的消耗。

1. 线程池的创建

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

corePoolSize：最小核心线程数;

maximumPoolSize：最大核心线程数;

keepAliveTime：空闲线程最大存活时间;

unit：存活时间的单位：秒/分/时…;

workQueue：任务队列，用来存放提交的任务;

threadFactory：线程工厂用来创建线程池中工作的线程;

handler：拒绝策略，当线程数达到最大线程数，在提交任务线程池该如何操作。那么就由拒绝策略来决定，一共有4种，默认采用的是抛出异常。其他3种下面会进行介绍。

2. 线程池的工作原理

这里大致介绍线程池的工作原理，具体的细节需要查看源码，在以下的源码介绍中会进行详细介绍。假设我们提交的任务为while(1)无限循环，会一直执行下去。

线程池的工作原理：

a. 在初始化线程池的时候，线程池的核心线程数和最大核心线程数都为0，任务队列为空；

b. 当当前线程第一次通过execute()方法提交任务的时候，线程池会创建一个核心线程，来执行任务。

c. 当提交的任务超过核心线程数的时候，线程池会把多余的任务加入到任务队列中；

d. 当任务队列被填充满了的时候，在提交其他任务，线程池会创建最大核心线程来处理任务。

e. 当创建的线程超过最大核心线程数的时候，继续提交任务会走相应的拒绝策略。默认拒绝策略就是报错。

3. 线程池的状态转换

线程池共有五种状态：RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED。

private static final int RUNNING = -1 << COUNT_BITS;  // 111-00000 00000000 00000000 00000000
private static final int SHUTDOWN = 0 << COUNT_BITS;  // 000-00000 00000000 00000000 00000000
private static final int STOP = 1 << COUNT_BITS;    // 001-00000 00000000 00000000 00000000
private static final int TIDYING = 2 << COUNT_BITS;  // 010-00000 00000000 00000000 00000000
private static final int TERMINATED = 3 << COUNT_BITS; // 011-00000 00000000 00000000 00000000

其中在线程池创建的时候初始状态为RUNNING。

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 111-00000 00000000 00000000 00000000

线程池的状态转换图：

1、RUNNING状态：能够接收新任务，以及对添加的任务进行处理。线程池的初始状态为RUNNING，并且线程池的任务数为0；

2、SHUTDOWN状态：不接收新任务，但能够处理已添加的任务。调用线程池的shutdown()方法，线程池由RUNNING -> SHUTDOWN；

3、STOP状态：不接受新任务，也不在处理已添加的任务，并且会中断正在处理的任务。调用线程池的shutdownNow()方法，线程池由(RUNNING or SHUTDOWN) -> STOP；

4、IDYING状态：当所有的任务已终止，ctl记录的 “任务数量” 为0，线程池会变为TIDYING状态。当线程池变为TIDYING状态时，会执行钩子函数terminated()。

terminated()方法在线程池中的实现是空的，若用户想在线程池变为TIDYING时，进行相应的处理，可以通过重载terminated()函数进行实现。

当在SHUTDOWN状态时，阻塞队列为空并且线程池中工作的线程数也为 0 时，就会由SHUTDOWN -> TIDYING。

当在STOP状态下，线程池中执行的任务为空时，就会由STOP -> TIDYING；

5、TERMINATED状态：线程池彻底终止，就会变为TERMINATED状态。线程池处于TIDYING状态时，执行完terminated()后, 就会由TIDYING -> TERMINATED。

4. 线程池源码解析

主要看线程池是如何工作的，那么就看它的提交方法的工作原理。线程池提供了两个函数进行任务提交，分别为execute方法和submit方法。通过底层可以看到submit方法底层仍然是调用execute方法。所以接下来主要以execute方法解析为主。

在介绍线程池的 execute 方法之前，先介绍一下线程池中的一个核心状态属性 ctl，类型是AtomicInteger，可以看作是一个整数类型。它表示线程池的控制状态，这一个状态包含了两个原子的属性，分别为工作的线程数和线程的状态。

可能会奇怪，一个整数类型，怎么表达两个意思呢？这就是它设计的巧妙之处，采用的是位操作，前 3 位用来表示线程池的状态。后 29 位表示创建的线程数。

// ctlOf 进行与操作，wc 都为 0, 所以结果由 rc 来决定
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 111-00000 00000000 00000000 00000000

private static int ctlOf(int rs, int wc) { return rs | wc; }

execute方法的源码解析，execute这个方法就可以体现出线程池的大致工作原理。其中工作线程数也就是线程池中已有的线程数。在 execute 代码中创建线程的工作主要在 addWorker 方法中。

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    // 获取线程池的状态，这个状态包括 线程池的状态（高3位）| 工作的线程数 (低29位)
    int c = ctl.get();
    // 计算工作线程数是否小于最小核心线程数
    if (workerCountOf(c) < corePoolSize) { 
    // true表示创建最小核心线程数的逻辑，一个工作线程 可以看成对应一个 Worker对象
        if (addWorker(command, true)) 
            // 表示创建成功
            return;
        c = ctl.get();
    }
    // 判断线程池的状态是否为Running状态，如果为true才加入到队列中
    if (isRunning(c) && workQueue.offer(command)) { 
        int recheck = ctl.get(); // 再次进行判断，确保线程安全
        if (! isRunning(recheck) && remove(command)) // 如果当前线程池不是运行状态，则移除队列中的任务
            reject(command); // 使用拒绝策略
        else if (workerCountOf(recheck) == 0) // 如果工作线程数为0, 则需要在创建新的线程进行处理任务
            addWorker(null, false);
    }
    else if (!addWorker(command, false)) // 绑定一个非核心线程数 (创建最大核心线程数逻辑)
        // 如果创建最大核心线程数失败，使用拒绝策略
        reject(command);
}

addWorker 方法创建线程，处理提交任务。在addWorker 方法中我们可以把一些判断逻辑去掉，主要看它主要的方法，其中最主要的就是 w = new Worker(firstTask); 创建一个工作线程对象。然后，获取工作线程的线程对象，调用它的start方法，启动工作线程。

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // Check if queue empty only if necessary.
        /**
         * 如果 rs >= SHUTDOWN 表示当前线程池不可能在接收任务了
         *
         */
        if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                    firstTask == null &&
                    !workQueue.isEmpty()))
            return false;
    
        for (;;) {
            int wc = workerCountOf(c);
            // core 为 true，表示是最小核心线程数的步骤, 如果 wc >= corePoolSize, 表示无法在创建最小核心线程数了
            // maximumPoolSize 也是一样，表示无法创建最大核心线程数了
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 先把工作线程数CAS + 1 操作
            if (compareAndIncrementWorkerCount(c))
                break retry; // 跳出 retry 标记层的循环
            c = ctl.get(); // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }
    
    boolean workerStarted = false;
    boolean workerAdded = false;
    // 一个Worker 包含一个线程和一个任务
    Worker w = null;
    try {
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            // 获取不到锁，在此阻塞
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if shut down before lock acquired.
                int rs = runStateOf(ctl.get());
    
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    // 把工作的线程添加到Set集合中
                    workers.add(w);
                    // 记录线程池中工作线程的值
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                // 任务开始执行，实现在 Worker中的run方法
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

Worker 工作线程类的具体实现。类中有两个非常重要的属性一个是创建的线程和提交的任务。也就是线程和任务绑定到一块。其中构造方法getThreadFactory()获取线程工程默认DefaultThreadFactory类的实现，然后调用newThread方法创建一个线程，参数是Worker 类自身。

也就是说，当线程启动的时候 (在addWorker 方法的 t.start() 完成) ，会调用 Worker的run方法实现。

而 Worker 的run方法调用 runWorker 方法。其中 runWoker 方法是线程池工作的主要方法。

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /** Thread this worker is running in. Null if factory fails. */
    final Thread thread;
    /** Initial task to run. Possibly null. */
    Runnable firstTask;
    ...
    Worker(Runnable firstTask) {
        ...
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
    
    /** Delegates main run loop to outer runWorker */
    public void run() {
        runWorker(this);
    }
}

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;
}

线程池工作的主要实现就是在 runWorker 方法中。去掉一些不重要的代码，主要看它的 while 的循环条件和 task.run(); 方法。线程池能够不停的进行工作原理，就是依赖阻塞队列完成的。当前线程如果完成了提交任务，会从阻塞队列中进行拉取任务，如果阻塞队列任务为空，那么当前线程就会阻塞。直到往阻塞队列里面添加任务，即可唤醒线程继续工作。

注意，当创建的核心线程数已满的时候，执行的任务已完成并且任务队列为空，那么创建的核心线程数拉取任务的时候就会阻塞，不会被释放掉。当再次提交任务的时候，不会在创建核心线程数了，而是唤醒阻塞的核心线程数继续从阻塞队列拉取任务，进行执行，如果执行完后，任务队列又为空，那么核心线程数再次阻塞。依次循环往复，到达线程复用的效果。

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // task != null 表示是创建新的 Worker线程
        // 如果 task == null, 那么会走 task = getTask() 表示该线程之前的任务已经执行完成了，从任务队列里获取任务
       // 通过循环不断的从当前提交的任务和从任务队列获取任务进行工作
        while (task != null || (task = getTask()) != null) { 
            w.lock();
            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(); // 直接调用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++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

线程复用的实现就是 getTask 方法。最大核心线程数过期的具体实现也在 getTask 方法中。如果 timed 为 true，说明当前线程数超过最小核心线程数（启用了最大核心线程数），会从阻塞队列中拉取并设置过期时间。

如果在指定时间内，拉取不到就会解开阻塞，表示任务队列中没有任务。那么最大核心线程数就可以关闭了。把timedOut 值为 true，在下一次的 for(;;) 的循环中调用compareAndDecrementWorkerCount 把 ctl 的线程数减1，并返回null。当返回为 null 的时候 runWorker的 while 条件就不满足了，会直接跳出去，这个函数也就执行完成了，那么对应的线程也就释放了。

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?
        // 是否为非核心线程数， (allowCoreThreadTimeOut 如果为ture 是允许关闭核心线程数的)
        // 保证线程数 符合核心线程数即可
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            // wc > maximumPoolSize 大于最大核心线程数
            // timed && timedOut (timed 为 true 表示当前线程数大于最小核心线程数) (timedOut 表示任务队列为空，并超过当前线程的最大存活时间)
            // wc > 1 表示有可以剔除的线程数量
            // 表示队列为空，目前没有任务
            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;
        }
    }
}