带你走近线程池（二）源码视角分析ThreadPoolExecutor

刺骨的言语ヽ痛彻心扉 2022-05-14 02:40 181阅读 0赞

# 一、主要方法罗列 #

上篇博客中[带你走近线程池（一 ）：线程池工作机制和原理][Link 1]大概介绍了线程池的工作原理，本节主要看一下ThreadPoolExecutor类里面的具体参数和方法。

主要讲下面几个方法

# 二、解析片段 #

### addWorker  创建线程 ###

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);
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    if (compareAndIncrementWorkerCount(c))
                        break 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 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();
                            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;
        }

首先是两个死循环，外循环主要检查线程池运行状态，内循环检查workerCount之后再检查运行状态。下面简单分析一下哪些情况下才可以进入到内循环，否则就直接返回false了。下面是可以进入到内循环的情况：  
（1）rs>=SHUTDOWN为false，即线程池处于RUNNING状态  
（2）rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()这个条件为true，也就意味着三个条件同时满足，即线程池状态为SHUTDOWN且firstTask为null且队列不为空，这种情况为处理队列中剩余任务。上面提到过当处于SHUTDOWN状态时，不接受新任务，但是会处理完队列里面的任务。如果firstTask不为null，那么就属于添加新任务；如果firstTask为null，并且队列为空，那么就不需要再处理了。  
当进入到内循环后，会首先获取当前运行的线程数量。首先判断当前运行线程数量是否大于等于CAPACITYA（2^29-1）,其次根据是否是核心线程与corePoolSize或者maximumPoolSize比较。所以线程的数量不会超过CAPACITY和maximumPoolSize的较小值。如果数量符合条件，那么就让ctl加1，然后跳出外部循环。如果线程数量达到了最大，那么回再判断当前状态，如果状态和之前的不一致了，那么继续外循环。下面是可以跳出外循环的情况：  
（1）如果是核心线程，当前线程数量小于CAPACITY和corePoolSize中的较小值  
（2）如果是非核心线程，当前线程数量小于CAPACITY和maximumPoolSize中的较小值。  
一旦跳出外循环，表示可以创建创建线程，这里具体是Worker对象，Worker实现了Runnable接口并且继承AbstractQueueSynchronizer，内部维持一个Runnbale的队列。try块中主要就是创建Worker对象，然后将其保存到workers中，workers是一个HashSet，表示工作线程的集合。然后如果添加成功，则开启Worker所在的线程。如果开启线程失败，则调用addWorkerFailed方法，addWokerFailed用于回滚worker线程的创建。

### addWorkerFailed 回滚创建的线程 ###

/**
         * Rolls back the worker thread creation.
         * - removes worker from workers, if present
         * - decrements worker count
         * - rechecks for termination, in case the existence of this
         *   worker was holding up termination
         */
        private void addWorkerFailed(Worker w) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                if (w != null)
                    workers.remove(w);
                decrementWorkerCount();
                tryTerminate();
            } finally {
                mainLock.unlock();
            }
        }

将整个线程加锁，移除worker，将wokerCount减1，最后检查终结状态。

### tryTerminate 检查终结状态  ###

理解这个检查终结状态的方法之前需要知道线程池的几个状态变量：

// runState is stored in the high-order bits
        private static final int RUNNING    = -1 << COUNT_BITS;
        private static final int SHUTDOWN   =  0 << COUNT_BITS;
        private static final int STOP       =  1 << COUNT_BITS;
        private static final int TIDYING    =  2 << COUNT_BITS;
        private static final int TERMINATED =  3 << COUNT_BITS;

**RUNNING**  
接受新任务并且处理已经进入队列的任务  
**SHUTDOWN**  
不接受新任务，但是处理已经进入队列的任务  
**STOP**  
不接受新任务，不处理已经进入队列的任务，并且中断正在执行的任务  
**TIDYING**  
所有任务执行完成，workerCount为0。线程转到了状态TIDYING会执行terminated()钩子方法  
**TERMINATED**  
terminated()已经执行完成

/**
         * Transitions to TERMINATED state if either (SHUTDOWN and pool
         * and queue empty) or (STOP and pool empty).  If otherwise
         * eligible to terminate but workerCount is nonzero, interrupts an
         * idle worker to ensure that shutdown signals propagate. This
         * method must be called following any action that might make
         * termination possible -- reducing worker count or removing tasks
         * from the queue during shutdown. The method is non-private to
         * allow access from ScheduledThreadPoolExecutor.
         */
        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.signalAll();
                        }
                        return;
                    }
                } finally {
                    mainLock.unlock();
                }
                // else retry on failed CAS
            }
        }

tryTerminate内部是一个死循环，首先判断状态，下面是跳出循环的情况：  
（1）线程池处于RUNNING状态  
（2）线程池状态处于TIDYING状态  
（3）线程池状态处于SHUTDOWN状态并且队列不为空  
如果不满足上述的情况，那么目前状态属于SHUTDOWN切队列为空，或者状态属于STOP，那么调用interruptIdleWorkers方法停止一个Worker线程，然后退出。  
接下来如果没有退出循环的话，那么就首先将状态设置成TIDYING，然后调用terminated方法，最后设置状态为TERMINATED。

[Link 1]: https://blog.csdn.net/zhou2s_101216/article/details/82262977