Java 线程池源码深度解析

// ctl，原子变量，存储状态和线程数量，初始化运行状态+0private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));// 静态常量，表示线程数量存放的位数29=32-3private static final int COUNT_BITS = Integer.SIZE - 3;// 线程数量最大的容量，低 COUNT_BITS 位所能表达的最大数值，000 11111111111111111111 => 5亿多private static final int CAPACITY  = (1 << COUNT_BITS) - 1;

// 111 000000000000000000，转换成整数后其实就是一个【负数】private static final int RUNNING    = -1 << COUNT_BITS;// 000 000000000000000000private static final int SHUTDOWN   =  0 << COUNT_BITS;// 001 000000000000000000private static final int STOP       =  1 << COUNT_BITS;// 010 000000000000000000private static final int TIDYING    =  2 << COUNT_BITS;// 011 000000000000000000private static final int TERMINATED =  3 << COUNT_BITS;

// ~CAPACITY = ~000 11111111111111111111 = 111 000000000000000000000（取反）// &运算符，和1&是它本身，和0&就是0,就可以获得高位值。private static int runStateOf(int c)     { return c & ~CAPACITY; }

// CAPACITY = 000 11111111111111111111// &运算符，和1&是它本身，和0&就是0,就可以获得低29位private static int workerCountOf(int c)  { return c & CAPACITY; }

// rs 表示线程池状态，wc 表示当前线程池中 worker（线程）数量，相与以后就是合并后的状态private static int ctlOf(int rs, int wc) { return rs | wc; }

// 比较当前线程池 ctl 所表示的状态，是否小于某个状态 sprivate static boolean runStateLessThan(int c, int s) { return c < s; }// 比较当前线程池 ctl 所表示的状态，是否大于等于某个状态sprivate static boolean runStateAtLeast(int c, int s) { return c >= s; }// 小于 SHUTDOWN 的一定是 RUNNING，SHUTDOWN == 0private static boolean isRunning(int c) { return c < SHUTDOWN; }

// 使用 CAS 方式 让 ctl 值 +1 ，成功返回 true, 失败返回 falseprivate boolean compareAndIncrementWorkerCount(int expect) {    return ctl.compareAndSet(expect, expect + 1);}// 使用 CAS 方式 让 ctl 值 -1 ，成功返回 true, 失败返回 falseprivate boolean compareAndDecrementWorkerCount(int expect) {    return ctl.compareAndSet(expect, expect - 1);}// 将 ctl 值减一，do while 循环会一直重试，直到成功为止private void decrementWorkerCount() {    do {} while (!compareAndDecrementWorkerCount(ctl.get()));}

// 线程池用于保存任务并将任务传递给工作线程的队列private final BlockingQueue<Runnable> workQueue;

// 增加减少 worker 或者时修改线程池运行状态需要持有 mainLockprivate final ReentrantLock mainLock = new ReentrantLock();

private final HashSet<Worker> workers = new HashSet<Worker>();

// 核心线程数量private volatile int corePoolSize;// 线程池最大线程数量private volatile int maximumPoolSize;  // 空闲线程存活时间private volatile long keepAliveTime;  // 创建线程时使用的线程工厂，默认是 DefaultThreadFactoryprivate volatile ThreadFactory threadFactory;  // 【超过核心线程提交任务就放入 阻塞队列】private final BlockingQueue<Runnable> workQueue;// 拒绝策略private volatile RejectedExecutionHandler handler;  

// 记录线程池生命周期内线程数最大值private int largestPoolSize;  // 记录线程池所完成任务总数，当某个 worker 退出时将完成的任务累加到该属性private long completedTaskCount;  

//  ThreadPoolExecutor#executepublic void execute(Runnable command) {        // 如果任务为空，直接抛空指针        if (command == null)            throw new NullPointerException();        // 获取ctl的值，其中高3位是状态信息，低3位是线程数量        int c = ctl.get();        // workerCountOf获取当前线程的数量        // 当前线程数量小于核心线程数，调用addWorker创建一个工作线程        if (workerCountOf(c) < corePoolSize) {            // 调用addWorker方法创建工作线程，直接执行任务。如果成功的话，直接结束方法。            if (addWorker(command, true))                return;            // 由于并发等原因，addWorker添加失败，会走到这里，再次获取ctl的值            c = ctl.get();        }      // 如果线程池是运行状态的话，就把任务加入到队列中        if (isRunning(c) && workQueue.offer(command)) {            // 双重检查，因为从上次检查到进入此方法，线程池可能已成为SHUTDOWN状态            int recheck = ctl.get();            // 如果发现线程池不是运行状态的话，那就移除这个任务            if (!isRunning(recheck) && remove(command))                // 任务出队成功，走拒绝策略                reject(command);             // 执行到这说明线程池是 running 状态，获取线程池中的线程数量，判断是否是 0             // 【担保机制】，保证线程池在 running 状态下，最起码得有一个线程在工作            else if (workerCountOf(recheck) == 0)                addWorker(null, false);        }        // 走到这里说明线程不是运行状态，或者就是队列满了，offer返回false        // 再次调用addWoker创建新的线程，如果不成功(一般是超过了线程池最大线程数量),执行拒绝策略        else if (!addWorker(command, false))            // 执行拒绝策略            reject(command);    }

//  ThreadPoolExecutor#addWorker// core == true 表示采用核心线程数量限制，false 表示采用 maximumPoolSizeprivate boolean addWorker(Runnable firstTask, boolean core) {     // 自旋【判断当前线程池状态是否允许创建线程】，允许就设置线程数量 + 1    retry:    for (;;) {         // 获取 ctl 的值        int c = ctl.get();        // 获取当前线程池运行状态        int rs = runStateOf(c);
         // 判断当前线程池状态【是否允许添加线程】               // 如果线程池状态大于SHUTDOWN 或者是SHUTDOWN状态，队列是空了的话，都不允许创建新的线程        if (rs >= SHUTDOWN &&            ! (rs == SHUTDOWN &&               firstTask == null &&               ! workQueue.isEmpty()))            // false，没有创建线程            return false;        // 再次自旋        for (;;) {            // 获取线程池中线程数量            int wc = workerCountOf(c);            // 如果线程数量超过阈值的话，返回false            if (wc >= CAPACITY ||                wc >= (core ? corePoolSize : maximumPoolSize))                return false;          // 记录线程数量已经加 1，类比于申请到了一块令牌，条件失败说明其他线程修改了数量            if (compareAndIncrementWorkerCount(c))                // 申请成功，跳出了 retry 这个 for 自旋                break retry;             // CAS 失败，没有成功的申请到令牌            c = ctl.get();             // 判断当前线程池状态是否发生过变化，被其他线程修改了，可能其他线程调用了 shutdown() 方法            if (runStateOf(c) != rs)                // 重新回到retry的执行点                continue retry;            // else CAS failed due to workerCount change; retry inner loop        }    }
    // 下面开始真正创建线程了    // 运行标记，表示创建的 worker 是否已经启动，false未启动  true启动    boolean workerStarted = false;    // 添加标记，表示创建的 worker 是否添加到池子中了，默认false未添加，true是添加。    boolean workerAdded = false;    Worker w = null;    try {        //【创建 Worker，底层通过线程工厂 newThread 方法创建执行线程，指定了首先执行的任务】        w = new Worker(firstTask);        // 将新创建的 worker 节点中的线程赋值给 t        final Thread t = w.thread;        // 这里的判断为了防止 程序员自定义的 ThreadFactory 实现类有 bug，创造不出线程        if (t != null) {            final ReentrantLock mainLock = this.mainLock;            // 加互斥锁，要添加 worker 了            mainLock.lock();            try {                 // 获取最新线程池运行状态                int rs = runStateOf(ctl.get());              // 判断线程池是否为RUNNING状态，不是再【判断当前是否为SHUTDOWN状态且firstTask为空，特殊情况】                if (rs < SHUTDOWN ||                    (rs == SHUTDOWN && firstTask == null)) {                     // 当线程start后，线程isAlive会返回true，这里还没开始启动线程，如果被启动了就需要报错                    if (t.isAlive())                         throw new IllegalThreadStateException();                    //将新建的 Worker 添加到线程池中                    workers.add(w);                    int s = workers.size();                    // 当前池中的线程数量是一个新高，更新 largestPoolSize                    if (s > largestPoolSize)                        largestPoolSize = s;                     // 添加标记置为 true                    workerAdded = true;                }            } finally {                mainLock.unlock();            }             // 添加成功就【启动线程执行任务】            if (workerAdded) {                // 启动线程                t.start();                // 运行标记置为 true                workerStarted = true;            }        }    } finally {        // 线程启动失败        if (! workerStarted)            // 清理工作，比如从线程池中移除。            addWorkerFailed(w);    }    return workerStarted;}
private void addWorkerFailed(Worker w) {    final ReentrantLock mainLock = this.mainLock;    // 持有线程池全局锁，因为操作的是线程池相关的东西    mainLock.lock();    try {        //条件成立需要将 worker 在 workers 中清理出去。        if (w != null)            workers.remove(w);        // 将线程池计数 -1，相当于归还令牌。        decrementWorkerCount();        // 尝试停止线程池        tryTerminate();    } finally {        //释放线程池全局锁。        mainLock.unlock();    }}

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {  // worker 内部封装的工作线程    final Thread thread;        // worker 第一个执行的任务，普通的 Runnable 实现类或者是 FutureTask    Runnable firstTask;      // 记录当前 worker 所完成任务数量    volatile long completedTasks;          // 构造方法    Worker(Runnable firstTask) {        // 设置AQS独占模式为初始化中状态，这个状态不能被抢占锁         setState(-1);        // firstTask不为空时，当worker启动后，内部线程会优先执行firstTask，执行完后会到queue中去获取下个任务        this.firstTask = firstTask;        // 使用线程工厂创建一个线程，并且【将当前worker指定为Runnable】，所以thread启动时会调用 worker.run()        this.thread = getThreadFactory().newThread(this);    }    // 不可重入锁，重写了AQS中的方法    protected boolean tryAcquire(int unused) {        if (compareAndSetState(0, 1)) {            setExclusiveOwnerThread(Thread.currentThread());            return true;        }        return false;    }
 protected boolean tryRelease(int unused) {        setExclusiveOwnerThread(null);        // 设置state为0，开始抢锁        setState(0);        return true;    }}

// Worker#runpublic void run() {    // 调用自身的runWoker方法    runWorker(this);}

// Worker#runWorkerfinal void runWorker(Worker w) {    Thread wt = Thread.currentThread();    // 获取 worker 的 firstTask    Runnable task = w.firstTask;    // 引用置空，【防止复用该线程时重复执行该任务】    w.firstTask = null;  // 初始化 worker 时设置 state = -1，表示不允许抢占锁    // 这里需要设置 state = 0 和 exclusiveOwnerThread = null，开始独占模式抢锁    w.unlock();    // true 表示发生异常退出，false 表示正常退出。    boolean completedAbruptly = true;    try {        // firstTask 不是 null 就直接运行，否则去 queue 中获取任务        while (task != null || (task = getTask()) != null) {            // worker 加锁，shutdown 时会判断当前 worker 状态，【根据独占锁状态判断是否空闲】            w.lock();            // 说明线程池状态大于 STOP，目前处于 STOP/TIDYING/TERMINATION，此时给线程一个中断信号            if ((runStateAtLeast(ctl.get(), STOP) ||                 (Thread.interrupted() &&                  runStateAtLeast(ctl.get(), STOP))) &&                // 线程不是处于中断的情况                !wt.isInterrupted())                 // 中断线程，设置线程的中断标志位为 true                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，代表任务执行完成                task = null;                // 更新worker完成任务数量                w.completedTasks++;                // 解锁                w.unlock();            }        }         // getTask()方法返回null时会走到这里，表示queue为空并且线程空闲超过保活时间，【当前线程执行退出逻辑】        completedAbruptly = false;    } finally {        // 正常退出 completedAbruptly = false         // 异常退出 completedAbruptly = true，【从 task.run() 内部抛出异常】时，跳到这一行        processWorkerExit(w, completedAbruptly);    }}

private Runnable getTask() {     // 超时标记，表示当前线程获取任务是否超时，true 表示已超时    boolean timedOut = false; 
    for (;;) {        int c = ctl.get();         // 获取线程池当前运行状态        int rs = runStateOf(c);
        // 如果发现线程池被关闭了，直接返回null        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {            // 使用 CAS 自旋的方式让 ctl 值 -1            decrementWorkerCount();            return null;        }
        // 获取线程池中的线程数量        int wc = workerCountOf(c);
        //timed用来判断当前线程是否超过一定时间没有获取任务就进行销毁回收，true是需要，false不需要， 有两种情况        //1. allowCoreThreadTimeOut为true代表允许回收核心线程，那就无所谓了，全部线程都执行超时回收        //2. 线程数量大于核心线程数，当前线程认为是非核心线程        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        // 同时满足下面1和2条件下，说明线程要回收，直接返回null        // 1. 如果线程数量超过最大线程数 或者 上面的timed和超时时间timedOut都为true        if ((wc > maximumPoolSize || (timed && timedOut))            // 2.如果线程数量大于1并且队列时空的情况            && (wc > 1 || workQueue.isEmpty())) {            // 使用 CAS 机制将 ctl 值 -1 ,减 1 成功的线程，返回 null，代表可以退出            if (compareAndDecrementWorkerCount(c))                return null;            continue;        }
        try {            // 从队列中获取任务，有下面两种方法            // timed为true, 调用超时方法poll获取任务            // timed为false,调用阻塞方法take获取            Runnable r = timed ?                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                workQueue.take();            if (r != null)                return r;            获取任务为 null 说明超时了，将超时标记设置为 true，进入下一次循环，就可以销毁这个线程了            timedOut = true;        } catch (InterruptedException retry) {             // 阻塞线程被打断后超时标记置为 false，【说明被打断不算超时】，要继续获取，直到超时或者获取到任务            // 如果线程池 SHUTDOWN 状态下的打断，会在循环获取任务前判断，返回 null            timedOut = false;        }    }}

// 正常退出 completedAbruptly = false，异常退出为 trueprivate void processWorkerExit(Worker w, boolean completedAbruptly) {    // 条件成立代表当前 worker 是发生异常退出的，task 任务执行过程中向上抛出异常了    if (completedAbruptly)         // 从异常时到这里 ctl 一直没有 -1，需要在这里 -1        decrementWorkerCount();
    final ReentrantLock mainLock = this.mainLock;    // 加锁    mainLock.lock();    try {        // 将当前 worker 完成的 task 数量，汇总到线程池的 completedTaskCount        completedTaskCount += w.completedTasks;    // 将 worker 从线程池中移除        workers.remove(w);    } finally {        mainLock.unlock();  // 解锁    }  // 尝试停止线程池，唤醒下一个线程    tryTerminate();
    int c = ctl.get();    // 线程池不是停止状态就应该有线程运行【担保机制】    if (runStateLessThan(c, STOP)) {        // 正常退出的逻辑，是对空闲线程回收，不是执行出错        if (!completedAbruptly) {            // 根据是否回收核心线程确定【线程池中的线程数量最小值】            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;            // 最小值为 0，但是线程队列不为空，需要一个线程来完成任务担保机制            if (min == 0 && !workQueue.isEmpty())                min = 1;            // 线程池中的线程数量大于最小值可以直接返回            if (workerCountOf(c) >= min)                return;        }        // 执行 task 时发生异常，有个线程因为异常终止了，需要添加        // 或者线程池中的数量小于最小值，这里要创建一个新 worker 加进线程池        addWorker(null, false);    }}