通俗易懂的JUC源码剖析-ThreadPoolExecutor

Posted 2021-10-06 小强大人

前言

ThreadPoolExecutor相信大家都很熟悉：线程池的实现类。今天我们就来看看它内部是怎么实现的。

实现原理

先来看看它的类结构：

public class ThreadPoolExecutor extends AbstractExecutorService {
}

public abstract class AbstractExecutorService implements ExecutorService {
}

public interface ExecutorService extends Executor {
    void shutdown();
    <T> Future<T> submit(Callable<T> task);
    // ...
}

public interface Executor {
     void execute(Runnable command);
}

再来看看它的关键属性：

// ctl高3位表示线程池的运行状态，低29位表示线程个数
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 线程个数掩码，Integer位数-3，与具体平台Integer位数有关，大部分是32-3=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;
// 关闭状态：拒绝新任务但会继续处理阻塞队列里的任务
// 调用shutdown()方法会变成这个状态
private static final int SHUTDOWN =  0 << COUNT_BITS;
// 停止状态：拒绝新任务并且会丢弃阻塞队列里的任务，同时还会中止正在运行的任务
// 调用shutdownNow()方法会变成这个状态
private static final int STOP =  1 << COUNT_BITS;
// 清洁状态：所有任务都执行完（包括阻塞队列里的任务），活动线程个数为0，将要调用terminated方法
private static final int TIDYING =  2 << COUNT_BITS;
// 终止状态：调用完terminated方法后的状态
private static final int TERMINATED =  3 << COUNT_BITS;
// 互斥锁，用来控制新增Worker操作的原子性
private final ReentrantLock mainLock = new ReentrantLock();
// 工作线程集合
private final HashSet<Worker> workers = new HashSet<Worker>();
// 线程池终止条件
private final Condition termination = mainLock.newCondition();
// 线程池核心参数
// 阻塞队列
private final BlockingQueue<Runnable> workQueue;
// 线程工厂
private volatile ThreadFactory threadFactory;
// 拒绝策略
private volatile RejectedExecutionHandler handler;
// 线程闲置时长
private volatile long keepAliveTime;
// 核心线程数
private volatile int corePoolSize;
// 最大线程数
private volatile int maximumPoolSize;
// 默认拒绝策略：AbortPolicy抛出异常
private static final RejectedExecutionHandler defaultHandler =
    new AbortPolicy();

获取线程池状态和线程个数方法：

// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

Worker是它的内部类，代表工作线程。

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable {
    Worker(Runnable firstTask) {
       // 先设置state为-1，目前是为了调用runWorker()前禁止中断
       setState(-1); // inhibit interrupts until runWorker
       this.firstTask = firstTask;
       // 创建一个线程，当前Worker实例作为Runnable接口实现
       // addWorker()里调用t.start()时，会调用Worker的runWorker()方法
       this.thread = getThreadFactory().newThread(this);
    }
}

它继承了AQS，实现了Runnable接口

再来看关键方法：
execute()

public void execute(Runnable command) {
    // 任务不能为空
    if (command == null)
        throw new NullPointerException();
    // 获取代表线程池状态和线程个数的控制变量ctl    
    int c = ctl.get();
    // 1.当前线程个数小于核心线程数时，尝试创建一个新的核心线程来执行任务
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;   
        c = ctl.get();
    }
    // 2.当前线程个数大于核心线程数且线程池正在运行时，尝试将任务放入到阻塞队列中
    if (isRunning(c) && workQueue.offer(command)) {
        // 二次检查 
        int recheck = ctl.get();
        // 线程池不处于运行状态时，则从队列中移除任务，并执行拒绝策略
        if (!isRunning(recheck) && remove(command))
            reject(command);
        // 如果当前线程池为空（没有工作线程），则创建一个非核心线程   
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 3.阻塞队列满了，且未达到最大线程数时，则创建新的非核心线程来执行任务
    // 4.阻塞队列满了，且超过最大线程数时，采用拒绝策略来拒绝任务
    else if (!addWorker(command, false))
        reject(command);
}

其中addWorker()方法如下：

// core变量代表待添加的是否为核心线程
private boolean addWorker(Runnable firstTask, boolean core) {
    // goto语句块
    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操作尝试增加线程个数的值workerCount
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get(); // Re-read ctl
            // 重新读取ctl变量的值，如果线程池运行状态发生变化，
            // 跳到retry重新读取新的状态并重试CAS操作
            if (runStateOf(c) != rs)
                continue retry;
            // 如果只是线程个数发生变化，说明只是CAS操作失败，继续内层循环重试CAS    
            // 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) {
           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();
                // 更新largestPoolSize的值
                if (s > largestPoolSize)
                    largestPoolSize = s;
                workerAdded = true;
            }
          } finally {
                mainLock.unlock();
          }
          if (workerAdded) {
             // 工作线程添加成功，则启动线程
             t.start();
             workerStarted = true;
          }
       }
    } finally {
        // 线程未能启动，处理添加工作线程的逻辑
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

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

来单独分析下这块代码，将!(a & b & c)展开可以得到:
rs != SHUTDOWN || (1)
firstTask != null || (2)
workQueue.isEmpty() (3)
也就是，以下三种情况addWorker()会直接返回false，不添加新的工作线程
(1)线程池状态rs > SHUTDOWN，即STOP、TIYDING、TERMINATED的一种
(2)线程池状态为SHUTDOWN，并且已经有了第一个任务
(3)线程池状态为SHUTDOWN，并且阻塞队列为空

任务提交到线程池后，就交给Worker来执行了。Worker实现了Runable接口，来看看它的run()方法：

public void run() {
    runWorker(this);
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    // 设置state为0，允许中断
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 循环从任务队列拉取任务getTask()
        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(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 {
        // 执行worker清理工作
        processWorkerExit(w, completedAbruptly);
    }
}

processWorkerExit()方法如下：

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 {
         // 将当前Worker完成的任务数增加到全局统计变量completedTaskCount
         completedTaskCount += w.completedTasks;
         // 从工作线程集合中移除当前线程
         workers.remove(w);
    } finally {
         mainLock.unlock();
    }
    // 以下2种情况会尝试将线程池状态置为terminated
    // 1.线程池状态为SHUTDOWN，且阻塞队列任务为空
    // 2.线程池状态为STOP，且当前活动线程个数为0
    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()方法如下：

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()方法：

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

shutdownNow()方法：

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(STOP);
        interruptWorkers();
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

还有一些细节明天再分析吧~ 祝大家新年快乐，牛气冲天！