ThreadPoolExecutor原理解析（二）

心已赠人 2024-03-26 14:54 60阅读 0赞

上章通过分析线程池的execute方法，大致了解了线程池的任务调度逻辑，本章继续结合源码分析线程池的生命周期及Worker线程管理逻辑。

### 线程池的生命周期管理 ###

/**
         * The main pool control state, ctl, is an atomic integer packing
         * two conceptual fields
         *   workerCount, indicating the effective number of threads
         *   runState,    indicating whether running, shutting down etc
         *
         * In order to pack them into one int, we limit workerCount to
         * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
         * billion) otherwise representable. If this is ever an issue in
         * the future, the variable can be changed to be an AtomicLong,
         * and the shift/mask constants below adjusted. But until the need
         * arises, this code is a bit faster and simpler using an int.
         *
         * The workerCount is the number of workers that have been
         * permitted to start and not permitted to stop.  The value may be
         * transiently different from the actual number of live threads,
         * for example when a ThreadFactory fails to create a thread when
         * asked, and when exiting threads are still performing
         * bookkeeping before terminating. The user-visible pool size is
         * reported as the current size of the workers set.
         *
         * The runState provides the main lifecycle control, taking on values:
         *
         *   RUNNING:  Accept new tasks and process queued tasks
         *   SHUTDOWN: Don't accept new tasks, but process queued tasks
         *   STOP:     Don't accept new tasks, don't process queued tasks,
         *             and interrupt in-progress tasks
         *   TIDYING:  All tasks have terminated, workerCount is zero,
         *             the thread transitioning to state TIDYING
         *             will run the terminated() hook method
         *   TERMINATED: terminated() has completed
         *
         * The numerical order among these values matters, to allow
         * ordered comparisons. The runState monotonically increases over
         * time, but need not hit each state. The transitions are:
         *
         * RUNNING -> SHUTDOWN
         *    On invocation of shutdown(), perhaps implicitly in finalize()
         * (RUNNING or SHUTDOWN) -> STOP
         *    On invocation of shutdownNow()
         * SHUTDOWN -> TIDYING
         *    When both queue and pool are empty
         * STOP -> TIDYING
         *    When pool is empty
         * TIDYING -> TERMINATED
         *    When the terminated() hook method has completed
         *
         * Threads waiting in awaitTermination() will return when the
         * state reaches TERMINATED.
         *
         * Detecting the transition from SHUTDOWN to TIDYING is less
         * straightforward than you'd like because the queue may become
         * empty after non-empty and vice versa during SHUTDOWN state, but
         * we can only terminate if, after seeing that it is empty, we see
         * that workerCount is 0 (which sometimes entails a recheck -- see
         * below).
         */
        //对线程池的运行状态和线程池中有效线程的数量进行控制的一个字段， 
        //它同时包含两部分的信息：线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount)，
        //高3位保存runState，低29位保存workerCount，两个变量之间互不干扰
        private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
        // 表示线程数的bit数
        private static final int COUNT_BITS = Integer.SIZE - 3;
        // 线程池支持的最大的线程数量536870911
        private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
        // runState is stored in the high-order bits 运行状态保存在int的高三位
        //RUNNING：线程池正常运行中，可以正常的接受并处理任务
        private static final int RUNNING    = -1 << COUNT_BITS;
        //SHUTDOWN：线程池关闭了，不能接受新任务，但是线程池会把阻塞队列中的剩余任务执行完，
        //剩余任务都处理完之后，会中断所有工作线程
        private static final int SHUTDOWN   =  0 << COUNT_BITS;
        //STOP：线程池停止了，不能接受新任务，并且也不会处理阻塞队列中的任务，会中断所有工作线程
        private static final int STOP       =  1 << COUNT_BITS;
        //TIDYING：当前线程池中没有工作线程后，也就是线程池没有工作线程在运行了，就会进入TIDYING，
        //这个状态是自动发生在SHUTDOWN和STOP之后的
        private static final int TIDYING    =  2 << COUNT_BITS;
        //TERMINATED：线程池处于TIDYING状态后，会执行terminated()方法，执行完后就会进入TERMINATED状态，
        //在ThreadPoolExecutor中terminated()是一个空方法，可以自定义线程池重写这个方法
        private static final int TERMINATED =  3 << COUNT_BITS;
    
        // Packing and unpacking ctl
        //计算当前运行状态
        private static int runStateOf(int c)     { return c & ~CAPACITY; }
        //计算当前线程数量
        private static int workerCountOf(int c)  { return c & CAPACITY; }
        //通过状态和线程数生成ctl
        private static int ctlOf(int rs, int wc) { return rs | wc; }

#### **线程池工作状态** ####

*  RUNNING

能够继续接受新的任务，同时继续处理阻塞队列中的任务

*  SHUTDOWN

不能接受新的任务，但是可以继续处理阻塞队列中的任务

*  STOP

不能接收新的任务，也不能处理阻塞队列中的任务，直接终止正在进行任务

*  TIDYING

当前线程池中没有工作线程后，也就是线程池没有工作线程在运行了，就会进入TIDYING，这个状态是自动发生在SHUTDOWN和STOP之后的

*  TERMINATED

线程池处于TIDYING状态后，会执行terminated()方法，执行完后就会进入TERMINATED状态，在ThreadPoolExecutor中terminated()是一个空方法，可以自定义线程池重写这个方法。

#### **线程池状态流转** ####

![b0ce022606887bcf57b8f2a63e3c6284.png][]

### **工作线程管理addWorker()** ###

addWorker方法主要作用是检查是否可以根据当前池状态和给定的界限(核心或最大值)添加新的worker。如果是这样，将相应地调整worker计数，如果可能的话，将创建并启动一个新的worker，并将firstTask作为它的第一个任务运行。如果线程池已停止或符合关闭条件，则此方法返回false。如果线程工厂在被要求创建线程时失败，它也会返回false。如果线程创建失败，要么是由于线程工厂返回null，要么是由于异常(通常是thread .start()中的OutOfMemoryError)。

/*
        *  检查是否可以根据当前池状态和给定的界限(核心或最大值)添加新的worker。
        *  如果是这样，将相应地调整worker计数，如果可能的话，将创建并启动一个新的worker，
        *  并将firstTask作为它的第一个任务运行。如果池已停止或符合关闭条件，则此方法返回false。
        *  如果线程工厂在被要求创建线程时失败，它也会返回false。如果线程创建失败，要么是由于线程
        *  工厂返回null，要么是由于异常(通常是thread .start()中的OutOfMemoryError)，我们将干净地回滚。
        *
        *  @param firstTask 新线程的任务应该首先运行(如果没有则为空)。使用初始的第一个任务
        *  (在execute()方法中)创建工作者，以在线程数少于corePoolSize时(在这种情况下，
        *  我们总是启动一个线程)或队列已满时(在这种情况下，我们必须绕过队列)绕过队列。
        *  最初空闲线程通常是通过prestartCoreThread创建的，或者用来替换其他垂死的工作线程。
        *
        *  @param core 采用的线程数限制，如果为true，采用核心线程数限制，
        *  false采用 maximumPoolSize线程数限制 (传入的参数一般都是false)
        *
        *  @return boolean 表示任务是否已经启动
        *             true表示创建worker成功且线程启动成功
        *             false： ①线程池状态大于SHUTDOWN时，一定会失败。
        *                     ②当前状态 = SHUTDOWN 但是队列中已经没有任务了或者当前状态是
        *                       SHUTDOWN且队列未空，但是firstTask不为NULL。
        *                     ③当前线程池已经达到了指定指标(core或max)
        *                     ④ThreadFactory创建的线程是NULL。
        * 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 (;;) {
                
                //获取当前的ctl值保存到c中
                int c = ctl.get();
                
                //rs：当前线程池的运行状态
                int rs = runStateOf(c);
                
                // 线程池的状态不是RUNNING，并且不是特例情况(线程池状态是SHUTDOWN并且队列不为空），
                // 那就直接return false，表示不新建工作线程
                // 如果是RUNNING，或者是特例情况，就准备新建工作线程
                // 总的来说就是判断当前线程池的状态是否允许继续添加线程
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null  &&
                       ! workQueue.isEmpty()))
                    return false;
                
                
                // 自旋
                for (;;) {
                    // 获取线程数量，
                    int wc = workerCountOf(c);
                    // 根据core参数判断当前线程数是否大于等于核心线程数或者是最大线程数。
                    // 大于等于的话直接return false
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    
                    // 使用CAS更新ctl，将线程数量+1
                    // cas成功则退出外层retry循环，去创建新的工作线程
                    // cas失败，则表示有其他线程也在提交任务，也在增加工作线程数，此时重新获取ctl
                    // 外部线程可能调用了shutdown或者shutdownNow(),线程池状态发生变化，CAS也会失败
                    if (compareAndIncrementWorkerC ount(c))
                       
                        break retry;
                 
                    // CAS失败，获取最新的ctl值
                    c = ctl.get();  
                    // 判断线程池的状态是不是RUNNING，如果变化了，继续自旋，
                    if (runStateOf(c) != rs)
                        // 状态发生变化，直接返回到外层循环，继续判断当前线程池状态和是否允许创建线程
                        continue retry;
                }
            }
            
            // 表示创建的worker是否已经启动
            boolean workerStarted = false;
           
            // 表示创建的worker是否添加到了workers(HashSet)中
            boolean workerAdded = false;
           
            Worker w = null;
            try {
                
                // 构造一个Worker对象，同时利用ThreadFactory新建一个线程
                w = new Worker(firstTask);
                
                // 将新创建的worker节点的线程 赋值给t
                final Thread t = w.thread;
                
                if (t != null) {
                    
                    // 将全局锁的引用保存到mainLock中
                    final ReentrantLock mainLock = this.mainLock;
                    // 尝试加锁
                    mainLock.lock();
                    
                    try {
                        
                        int rs = runStateOf(ctl.get());
                        
                        // 判断当前线程池的状态是否RUNNING
                        // 或者线程池的状态变成了SHUTDOWN，但是当前线程没有自己的第一个任务，
                        // 那就表示当前调用addWorker方法是为了从队列中获取任务来执行
                        // 正常情况下线程池的状态如果是SHUTDOWN，是不能创建新的工作线程的，
                        // 但是队列中如果有任务，那就是上面说的特例情况
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                            
                            // 如果Worker对象对应的线程已经在运行了，直接抛异常
                            if (t.isAlive()) 
                                // 抛出异常
                                throw new IllegalThreadStateException();
                            
                            // 将创建的worker添加到线程池中。
                            workers.add(w);
                            
                            // 获取最新线程池线程的数量
                            int s = workers.size();
                            
                            // 条件成立，更新largestPoolSize 
                            // largestPoolSize用来跟踪线程池在运行过程中工作线程数的峰值
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            
                            // 更新标志位，表示线程已经追加到线程池中了。
                            workerAdded = true;
                        }
                    } finally {
                        // 释放全局锁
                        mainLock.unlock();
                    }
                    
                    if (workerAdded) {
                        // 线程启动。
                        t.start();
                        // 启动标记设置为true。
                        workerStarted = true;
                    }
                }
            } finally {
                // 新建工作线程失败
                if (!workerStarted)
                    // 添加失败，回滚工作线程的创建
                    addWorkerFailed(w);
            }
            // 添加工作线程成功
            return workerStarted;
        }

#### addWorker()执行流程 ####

![1415ccd6310cea5a119806485fe491f8.png][]

#### addWorkerFailed(Worker w) ####

/**
         * 回滚工作线程的创建。从workers中移除失败的worker，以防这个worker的存在阻碍了线程池终止
         * 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中移除失败的worker
                    workers.remove(w);
                // 修改ctl，工作线程数-1
                decrementWorkerCount();
                tryTerminate();
            } finally {
                mainLock.unlock();
            }
        }

[b0ce022606887bcf57b8f2a63e3c6284.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/9099e81670624ac1a575a94d7dc0cad3.png
[1415ccd6310cea5a119806485fe491f8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/a344be11c0944195ab5e8a900112b47a.png