• jdk线程池ThreadPoolExecutor优雅停止原理解析（自己动手实现线程池）（二）

    ThreadPoolExecutor优雅停止源码分析(自己动手实现线程池v2版本)

    ThreadPoolExecutor为了实现优雅停止功能，为线程池设置了一个状态属性，其共有5种情况。
    在第一篇博客中曾介绍过，AtomicInteger类型的变量ctl同时维护了两个业务属性当前活跃工作线程个数与线程池状态，其中ctl的高3位用于存放线程池状态。

    线程池工作状态介绍

    线程池工作状态是单调推进的，即从运行时->停止中->完全停止。共有以下五种情况

    1. RUNNING

    RUNNING状态，代表着线程池处于正常运行(运行时)。RUNNING状态的线程池能正常的接收并处理提交的任务
    ThreadPoolExecutor初始化时对ctl赋予的默认属性便是RUNNING（private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));）
    RUNNING状态下线程池正常工作的原理已经在第一篇博客中详细的介绍过了，这里不再赘述。

    2. SHUTDOWN

    SHUTDOWN状态，代表线程池处于停止对外服务的状态(停止中)。不再接收新提交的任务，但依然会将workQueue工作队列中积压的任务逐步处理完。
    用户可以通过调用shutdown方法令线程池由RUNNING状态进入SHUTDOWN状态，shutdown方法会在下文详细展开分析。

    3. STOP

    STOP状态，代表线程池处于停止状态。不再接受新提交的任务(停止中)，同时也不再处理workQueue工作队列中积压的任务，当前还在处理任务的工作线程将收到interrupt中断通知
    用户可以通过调用shutdownNow方法令线程池由RUNNING或者SHUTDOWN状态进入STOP状态，shutdownNow方法会在下文详细展开分析。

    4. TIDYING

    TIDYING状态，代表着线程池即将完全终止，正在做最后的收尾工作(停止中)。
    在线程池中所有的工作线程都已经完全退出，且工作队列中的任务已经被清空时会由SHUTDOWN或STOP状态进入TIDYING状态。

    5. TERMINATED

    TERMINATED状态，代表着线程池完全的关闭(完全停止)。

    1. public class MyThreadPoolExecutorV2 implements MyThreadPoolExecutor {
    2.     /**
    3.      * 当前线程池中存在的worker线程数量 + 状态的一个聚合（通过一个原子int进行cas，来避免对两个业务属性字段加锁来保证一致性）
    4.      */
    5.     private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    6.     private static final int COUNT_BITS = Integer.SIZE - 3;
    7.  
    8.     /**
    9.      * 32位的有符号整数，有3位是用来存放线程池状态的，所以用来维护当前工作线程个数的部分就只能用29位了
    10.      * 被占去的3位中，有1位原来的符号位，2位是原来的数值位
    11.      * */
    12.     private static final int CAPACITY = (1 << COUNT_BITS) - 1;
    13.  
    14.     /**
    15.      * 线程池状态poolStatus常量（状态值只会由小到大，单调递增）
    16.      * 线程池状态迁移图：
    17.      *         ↗ SHUTDOWN ↘
    18.      * RUNNING       ↓       TIDYING → TERMINATED
    19.      *         ↘   STOP   ↗
    20.      * 1 RUNNING状态，代表着线程池处于正常运行的状态。能正常的接收并处理提交的任务
    21.      * 线程池对象初始化时，状态为RUNNING
    22.      * 对应逻辑：private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    23.      *
    24.      * 2 SHUTDOWN状态，代表线程池处于停止对外服务的状态。不再接收新提交的任务，但依然会将workQueue工作队列中积压的任务处理完
    25.      * 调用了shutdown方法时，状态由RUNNING -> SHUTDOWN
    26.      * 对应逻辑：shutdown方法中的advanceRunState(SHUTDOWN);
    27.      *
    28.      * 3 STOP状态，代表线程池处于停止状态。不再接受新提交的任务，同时也不再处理workQueue工作队列中积压的任务，当前还在处理任务的工作线程将收到interrupt中断通知
    29.      * 之前未调用shutdown方法，直接调用了shutdownNow方法，状态由RUNNING -> STOP
    30.      * 之前先调用了shutdown方法，后调用了shutdownNow方法，状态由SHUTDOWN -> STOP
    31.      * 对应逻辑：shutdownNow方法中的advanceRunState(STOP);
    32.      *
    33.      * 4 TIDYING状态，代表着线程池即将完全终止，正在做最后的收尾工作
    34.      * 当前线程池状态为SHUTDOWN,任务被消费完工作队列workQueue为空，且工作线程全部退出完成工作线程集合workers为空时，tryTerminate方法中将状态由SHUTDOWN->TIDYING
    35.      * 当前线程池状态为STOP,工作线程全部退出完成工作线程集合workers为空时，tryTerminate方法中将状态由STOP->TIDYING
    36.      * 对应逻辑：tryTerminate方法中的ctl.compareAndSet(c, ctlOf(TIDYING, 0)
    37.      *
    38.      * 5 TERMINATED状态，代表着线程池完全的关闭。之前线程池已经处于TIDYING状态，且调用的钩子函数terminated已返回
    39.      * 当前线程池状态为TIDYING，调用的钩子函数terminated已返回
    40.      * 对应逻辑：tryTerminate方法中的ctl.set(ctlOf(TERMINATED, 0));
    41.      * */
    43.     //  11100000 00000000 00000000 00000000
    44.     private static final int RUNNING = -1 << COUNT_BITS;
    45.     //  00000000 00000000 00000000 00000000
    46.     private static final int SHUTDOWN = 0 << COUNT_BITS;
    47.     //  00100000 00000000 00000000 00000000
    48.     private static final int STOP = 1 << COUNT_BITS;
    49.     //  01000000 00000000 00000000 00000000
    50.     private static final int TIDYING = 2 << COUNT_BITS;
    51.     //  01100000 00000000 00000000 00000000
    52.     private static final int TERMINATED = 3 << COUNT_BITS;
    53.  
    54.     private static int runStateOf(int c) {
    55.         return c & ~CAPACITY;
    56.     }
    57.     
    58.     private static int ctlOf(int rs, int wc) {
    59.         return rs | wc;
    60.     }
    61.     
    62.     private static boolean runStateLessThan(int c, int s) {
    63.         return c < s;
    64.     }
    65.  
    66.     private static boolean runStateAtLeast(int c, int s) {
    67.         return c >= s;
    68.     }
    69.  
    70.     private static boolean isRunning(int c) {
    71.         return c < SHUTDOWN;
    72.     }
    73.  
    74.     /**
    75.      * 推进线程池工作状态
    76.      * */
    77.     private void advanceRunState(int targetState) {
    78.         for(;;){
    79.             // 获得当前的线程池状态
    80.             int currentCtl = this.ctl.get();
    81.  
    82.             // 1 （runState >= targetState）如果当前线程池状态不比传入的targetState小
    83.             // 代表当前状态已经比参数要制定的更加快(或者至少已经处于对应阶段了)，则无需更新poolStatus的状态(或语句中第一个条件为false，直接break了)
    84.             // 2  (this.ctl.compareAndSet)，cas的将runState更新为targetState
    85.             // 如果返回true则说明cas更新成功直接break结束（或语句中第一个条件为false，第二个条件为true）
    86.             // 如果返回false说明cas争抢失败，再次进入while循环重试（或语句中第一个和第二个条件都是false，不break而是继续执行循环重试）
    87.             if (runStateAtLeast(currentCtl, targetState) ||
    88.                     this.ctl.compareAndSet(
    89.                             currentCtl,
    90.                             ctlOf(targetState, workerCountOf(currentCtl)
    91.                             ))) {
    92.                 break;
    93.             }
    94.         }
    95.     }
    96. }
    • 因为线程池状态不是单独存放，而是放在ctl这一32位数据的高3位的，读写都比较麻烦，因此提供了runStateOf和ctlOf等辅助方法（位运算）来简化操作。
    • 线程池的状态是单调递进的，由于巧妙的将状态靠前的值设置的更小，因此通过直接比较状态的值来判断当前线程池状态是否推进到了指定的状态（runStateLessThan、runStateAtLeast、isRunning、advanceRunState）。

    jdk线程池ThreadPoolExecutor优雅停止具体实现原理

    线程池的优雅停止一般要能做到以下几点：

    1. 线程池在中止后不能再受理新的任务
    2. 线程池中止的过程中，已经提交的现存任务不能丢失（等待剩余任务执行完再关闭或者能够把剩余的任务吐出来还给用户）
    3. 线程池最终关闭前，确保创建的所有工作线程都已退出，不会出现资源的泄露

    下面我们从源码层面解析ThreadPoolExecutor，看看其是如何实现上述这三点的.

    如何中止线程池

    ThreadPoolExecutor线程池提供了shutdown和shutdownNow这两个public方法给使用者用于发出线程池的停止指令。

    shutdown方法

    shutdown方法用于关闭线程池，并令线程池从RUNNING状态转变位SHUTDOWN状态。位于SHUTDOWN状态的线程池，不再接收新任务，但已提交的任务会全部被执行完。

    1.     /**
    2.      * 关闭线程池（不再接收新任务，但已提交的任务会全部被执行）
    3.      * 但不会等待任务彻底的执行完成（awaitTermination）
    4.      */
    5.     public void shutdown() {
    6.         final ReentrantLock mainLock = this.mainLock;
    7.  
    8.         // shutdown操作中涉及大量的资源访问和更新，直接通过互斥锁防并发
    9.         mainLock.lock();
    10.         try {
    11.             // 用于shutdown/shutdownNow时的安全访问权限
    12.             checkShutdownAccess();
    13.             // 将线程池状态从RUNNING推进到SHUTDOWN
    14.             advanceRunState(SHUTDOWN);
    15.             // shutdown不会立即停止所有线程，而仅仅先中断idle状态的多余线程进行回收，还在执行任务的线程就慢慢等其执行完
    16.             interruptIdleWorkers();
    17.             // 单独为ScheduledThreadPoolExecutor开的一个钩子函数（hook for ScheduledThreadPoolExecutor）
    18.             onShutdown();
    19.         } finally {
    20.             mainLock.unlock();
    21.         }
    22.  
    23.         // 尝试终止线程池
    24.         tryTerminate();
    25.     }
    26.  
    27.     /**
    28.      * 用于shutdown/shutdownNow时的安全访问权限
    29.      * 检查当前调用者是否有权限去通过interrupt方法去中断对应工作线程
    30.      * */
    31.     private void checkShutdownAccess() {
    32.         // 判断jvm启动时是否设置了安全管理器SecurityManager
    33.         SecurityManager security = System.getSecurityManager();
    34.         // 如果没有设置，直接返回无事发生
    35.  
    36.         if (security != null) {
    37.             // 设置了权限管理器，验证当前调用者是否有modifyThread的权限
    38.             // 如果没有，checkPermission会抛出SecurityException异常
    39.             security.checkPermission(shutdownPerm);
    40.     
    41.             // 通过上述校验，检查工作线程是否能够被调用者访问
    42.             final ReentrantLock mainLock = this.mainLock;
    43.             mainLock.lock();
    44.             try {
    45.                 for (MyWorker w : workers) {
    46.                     // 检查每一个工作线程中的thread对象是否有权限被调用者访问
    47.                     security.checkAccess(w.thread);
    48.                 }
    49.             } finally {
    50.                 mainLock.unlock();
    51.             }
    52.         }
    53.     }
    54.  
    55.     /**
    56.      * 中断所有处于idle状态的线程
    57.      * */
    58.     private void interruptIdleWorkers() {
    59.         // 默认打断所有idle状态的工作线程
    60.         interruptIdleWorkers(false);
    61.     }
    62.  
    63.     private static final boolean ONLY_ONE = true;
    64.  
    65.     /**
    66.      * 中断处于idle状态的线程
    67.      * @param onlyOne 如果为ture，至多只中断一个工作线程(可能一个都不中断)
    68.      *                如果为false，中断workers内注册的所有工作线程
    69.      * */
    70.     private void interruptIdleWorkers(boolean onlyOne) {
    71.         final ReentrantLock mainLock = this.mainLock;
    72.         mainLock.lock();
    73.         try {
    74.             for (MyWorker w : workers) {
    75.                 Thread t = w.thread;
    76.                 // 1. t.isInterrupted()，说明当前线程存在中断信号，之前已经被中断了，无需再次中断
    77.                 // 2. w.tryLock(), runWorker方法中如果工作线程获取到任务开始工作，会先进行Lock加锁
    78.                 // 则这里的tryLock会加锁失败，返回false。 而返回true的话，就说明当前工作线程是一个idle线程，需要被中断
    79.                 if (!t.isInterrupted() && w.tryLock()) {
    80.                     try {
    81.                         t.interrupt();
    82.                     } catch (SecurityException ignore) {
    83.                     } finally {
    84.                         // tryLock成功时，会将内部state的值设置为1，通过unlock恢复到未加锁的状态
    85.                         w.unlock();
    86.                     }
    87.                 }
    88.                 if (onlyOne) {
    89.                     // 参数onlyOne为true，至多只中断一个工作线程
    90.                     // 即使上面的t.interrupt()没有执行，也在这里跳出循环
    91.                     break;
    92.                 }
    93.             }
    94.         } finally {
    95.             mainLock.unlock();
    96.         }
    97.     }
    98.     
    99.     /**
    100.      * 单独为jdk的ScheduledThreadPoolExecutor开的一个钩子函数
    101.      * 由ScheduledThreadPoolExecutor继承ThreadExecutor时重写（包级别访问权限）
    102.      * */
    103.     void onShutdown() {}
    1. shutdown方法在入口处使用mainLock加锁后，通过checkShutdownAccess检查当前是否有权限访问工作线程（前提是设置了SecurityManager），如果无权限则会抛出SecurityException异常。
    2. 通过advanceRunState方法将线程池状态推进到SHUTDOWN。
    3. 通过interruptIdleWorkers使用中断指令（Thread.interrupt）唤醒所有处于idle状态的工作线程（存在idle状态的工作线程代表着当前工作队列是空的）。
      idle的工作线程在被唤醒后从getTask方法中退出（getTask中对应的退出逻辑在下文中展开），进而退出runWorker方法，最终系统回收掉工作线程占用的各种资源（第一篇博客中runWorker的解析中提到过）。
    4. 调用包级别修饰的钩子函数onShutdown。这一方法是作者专门为同为java.util.concurrent包下的ScheduledThreadPoolExecutor提供的拓展，不在本篇博客中展开。
    5. 前面提到SHUTDOWN状态的线程池在工作线程都全部退出且工作队列为空时会转变为TIDYING状态，因此通过调用tryTerminate方法尝试终止线程池（当前不一定会满足条件，比如调用了shutdown但工作队列还有很多任务等待执行）。
      tryTerminate方法中细节比较多，下文中再展开分析。

    shutdownNow方法

    shutdownNow方法同样用于关闭线程池，但比shutdown方法更加激进。shutdownNow方法令线程池从RUNNING状态转变为STOP状态，不再接收新任务，而工作队列中未完成的任务会以列表的形式返回给shutdownNow的调用者。

    • shutdown方法在调用后，虽然不再接受新任务，但会等待工作队列中的队列被慢慢消费掉；而shutdownNow并不会等待，而是将当前工作队列中的所有未被捞取执行的剩余任务全部返回给shutdownNow的调用者，并对所有的工作线程(包括非idle的线程)发出中断通知。
    • 这样做的好处是线程池可以更快的进入终止态，而不必等剩余的任务都完成，都返回给用户后也不会丢任务。
    1.     /**
    2.      * 立即关闭线程池（不再接收新任务，工作队列中未完成的任务会以列表的形式返回）
    3.      * @return 当前工作队列中未完成的任务
    4.      * */
    5.     public List<Runnable> shutdownNow() {
    6.         List<Runnable> tasks;
    7.  
    8.         final ReentrantLock mainLock = this.mainLock;
    9.  
    10.         // shutdown操作中涉及大量的资源访问和更新，直接通过互斥锁防并发
    11.         mainLock.lock();
    12.         try {
    13.             // 用于shutdown/shutdownNow时的安全访问权限
    14.             checkShutdownAccess();
    15.             // 将线程池状态从RUNNING推进到STOP
    16.             advanceRunState(STOP);
    17.             interruptWorkers();
    18.  
    19.             // 将工作队列中未完成的任务提取出来(会清空线程池的workQueue)
    20.             tasks = drainQueue();
    21.         } finally {
    22.             mainLock.unlock();
    23.         }
    24.  
    25.         // 尝试终止线程池
    26.         tryTerminate();
    27.         return tasks;
    28.     }
    29.  
    30.    /**
    31.     * shutdownNow方法内，立即终止线程池时该方法被调用
    32.     * 中断通知所有已经启动的工作线程（比如等待在工作队列上的idle工作线程，或者run方法内部await、sleep等，令其抛出中断异常快速结束）
    33.     * */
    34.     private void interruptWorkers() {
    35.         final ReentrantLock mainLock = this.mainLock;
    36.         mainLock.lock();
    37.         try {
    38.             for (MyWorker w : workers) {
    39.               // 遍历所有的worker线程，已启动的工作线程全部调用Thread.interrupt方法，发出中断信号
    40.               w.interruptIfStarted();
    41.             }
    42.         } finally {
    43.             mainLock.unlock();
    44.         }
    45.     }
    46.  
    47.    /**
    48.     * 将工作队列中的任务全部转移出来
    49.     * 用于shutdownNow紧急关闭线程池时将未完成的任务返回给调用者，避免任务丢失
    50.     * */
    51.    private List<Runnable> drainQueue() {
    52.         BlockingQueue<Runnable> queue = this.workQueue;
    53.         ArrayList<Runnable> taskList = new ArrayList<>();
    54.         queue.drainTo(taskList);
    55.         // 通常情况下，普通的阻塞队列的drainTo方法可以一次性的把所有元素都转移到taskList中
    56.         // 但jdk的DelayedQueue或者一些自定义的阻塞队列，drainTo方法无法转移所有的元素
    57.         // （比如DelayedQueue的drainTo方法只能转移已经不需要延迟的元素，即getDelay()<=0）
    58.         if (!queue.isEmpty()) {
    59.            // 所以在这里打一个补丁逻辑：如果drainTo方法执行后工作队列依然不为空，则通过更基础的remove方法把队列中剩余元素一个一个的循环放到taskList中
    60.            for (Runnable r : queue.toArray(new Runnable[0])) {
    61.               if (queue.remove(r)) {
    62.                 taskList.add(r);
    63.               }
    64.            }
    65.         }
    66.         
    67.         return taskList;
    68.    }
    1. shutdownNow方法在入口处使用mainLock加锁后，与shutdown方法一样也通过checkShutdownAccess检查当前是否有权限访问工作线程（前提是设置了SecurityManager），如果无权限则会抛出SecurityException异常。
    2. 通过advanceRunState方法将线程池状态推进到STOP。
    3. 通过interruptWorkers使用中断指令（Thread.interrupt）唤醒所有工作线程（区别于shutdown中的interruptIdleWorkers）。区别在于除了idle的工作线程，所有正在执行任务的工作线程也会收到中断通知，期望其能尽快退出任务的执行。
    4. 通过drainQueue方法将当前工作线程中剩余的所有任务以List的形式统一返回给调用者。
    5. 通过调用tryTerminate方法尝试终止线程池。

    如何保证线程池在中止后不能再受理新的任务？

    在execute方法作为入口，提交任务的逻辑中，v2版本相比v1版本新增了一些基于线程池状态的校验（和jdk的实现保持一致了）。

    execute方法中的校验

    • 首先在execute方法中，向工作队列加入新任务前（workQueue.offer）对当前线程池的状态做了一个校验（isRunning(currentCtl)）。希望非RUNNING状态的线程池不向工作队列中添加新任务
      但在做该检查时可能与shutdown/shutdownNow内推进线程池状态的逻辑并发执行，所以在工作队列成功加入任务后还需要再检查一次线程池状态，如果此时已经不是RUNNING状态则需要通过remove方法将刚入队的任务从队列中移除，并调用reject方法(拒绝策略)

    addWorker方法中的校验

    • 在addWorker方法的入口处(retry:第一层循环通过(runState >= SHUTDOWN && !(runState == SHUTDOWN && firstTask == null && !workQueue.isEmpty())))逻辑，
      保证了不是RUNNING状态的线程池（runState >= SHUTDOWN），无法创建新的工作线程（addWorker返回false）。
      但有一种特殊情况：即SHUTDOWN状态下(runState == SHUTDOWN)，工作队列不为空(!workQueue.isEmpty())，且不是第一次提交任务时创建新工作线程（firstTask == null），
      依然允许创建新的工作线程，因为即使在SHUTDOWN状态下，某一存活的工作线程发生中断异常时，会调用processWorkerExit方法，在销毁原有工作线程后依然需要调用addWorker重新创建一个新的（firstTask == null）

    execute与shutdown/shutdownNow并发时的处理

    execute提交任务时addWorker方法和shutdown/shutdownNow方法是可能并发执行的，但addWorker中有多处地方都对线程池的状态进行了检查，尽最大的可能避免线程池停止时继续创建新的工作线程。

    1. retry循环中，compareAndIncrementWorkerCount方法会cas的更新状态（此前获取到的ctl状态必然是RUNNING，否则走不到这里），cas成功则会跳出retry:循环（ break retry;）。
      而cas失败可能有两种情况：
      如果是workerCount发生了并发的变化，则在内层的for (;;)循环中进行重试即可
      如果线程池由于收到终止指令而推进了状态，则随后的if (runStateOf(currentCtl) != runState)将会为true，跳出到外层的循环重试（continue retry）
    2. 在new Worker(firstTask)后，使用mainLock获取锁后再一次检查线程池状态（if (runState < SHUTDOWN ||(runState == SHUTDOWN && firstTask == null))）。
      由于shutdown、shutdownNow也是通过mainLock加锁后才推进的线程池状态，因此这里获取到的状态是准确的。
      如果校验失败（if结果为false），则workers中不会加入新创建的工作线程，临时变量workerAdded=false，则工作线程不会启动（t.start()）。临时变量workerStarted也为false，最后会调用addWorkerFailed将新创建的工作线程回收掉（回滚）

    基于execute方法和addWorker方法中关于各项关于线程池停止状态校验，最大程度的避免了线程池在停止过程中新任务的提交和可能的新工作线程的创建。使得execute方法在线程池接收到停止指令后（>=SHUTDOWN），最终都会去执行reject拒绝策略逻辑。

    1. /**
    2.      * 提交任务，并执行
    3.      * */
    4.     @Override
    5.     public void execute(Runnable command) {
    6.         if (command == null){
    7.             throw new NullPointerException("command参数不能为空");
    8.         }
    9.  
    10.         int currentCtl = this.ctl.get();
    11.         if (workerCountOf(currentCtl) < this.corePoolSize) {
    12.             // 如果当前存在的worker线程数量低于指定的核心线程数量，则创建新的核心线程
    13.             boolean addCoreWorkerSuccess = addWorker(command,true);
    14.             if(addCoreWorkerSuccess){
    15.                 // addWorker添加成功，直接返回即可
    16.                 return;
    17.             }
    18.  
    19.             // addWorker失败了
    20.             // 失败的原因主要有以下几个：
    21.             // 1 线程池的状态出现了变化，比如调用了shutdown/shutdownNow方法，不再是RUNNING状态，停止接受新的任务
    22.             // 2 多个线程并发的execute提交任务，导致cas失败，重试后发现当前线程的个数已经超过了限制
    23.             // 3 小概率是ThreadFactory线程工厂没有正确的返回一个Thread
    24.  
    25.             // 获取最新的ctl状态
    26.             currentCtl = this.ctl.get();
    27.         }
    28.  
    29.         // 走到这里有两种情况
    30.         // 1 因为核心线程超过限制（workerCountOf(currentCtl) < corePoolSize == false），需要尝试尝试将任务放入阻塞队列
    31.         // 2 addWorker返回false，创建核心工作线程失败
    32.  
    33.         // 判断当前线程池状态是否为running
    34.         // 如果是running状态，则进一步执行任务入队操作
    35.         if(isRunning(currentCtl) && this.workQueue.offer(command)){
    36.             // 线程池是running状态，且workQueue.offer入队成功
    37.  
    38.             int recheck = this.ctl.get();
    39.             // 重新检查状态，避免在上面入队的过程中线程池并发的关闭了
    40.             // 如果是isRunning=false，则进一步需要通过remove操作将刚才入队的任务删除，进行回滚
    41.             if (!isRunning(recheck) && remove(command)) {
    42.                 // 线程池关闭了，执行reject操作
    43.                 reject(command);
    44.             } else if(workerCountOf(currentCtl) == 0){
    45.                 // 在corePoolSize为0的情况下，当前不存在存活的核心线程
    46.                 // 一个任务在入队之后，如果当前线程池中一个线程都没有，则需要兜底的创建一个非核心线程来处理入队的任务
    47.                 // 因此firstTask为null，目的是先让任务先入队后创建线程去拉取任务并执行
    48.                 addWorker(null,false);
    49.             }else{
    50.                 // 加入队列成功，且当前存在worker线程，成功返回
    51.                 return;
    52.             }
    53.         }else{
    54.             // 阻塞队列已满，尝试创建一个新的非核心线程处理
    55.             boolean addNonCoreWorkerSuccess = addWorker(command,false);
    56.             if(!addNonCoreWorkerSuccess){
    57.                 // 创建非核心线程失败，执行拒绝策略（失败的原因和前面创建核心线程addWorker的原因类似）
    58.                 reject(command);
    59.             }else{
    60.                 // 创建非核心线程成功，成功返回
    61.                 return;
    62.             }
    63.         }
    64.     }
    1. /**
    2.      * 向线程池中加入worker
    3.      * */
    4.     private boolean addWorker(Runnable firstTask, boolean core) {
    5.         // retry标识外层循环
    6.         retry:
    7.         for (;;) {
    8.             int currentCtl = ctl.get();
    9.             int runState = runStateOf(currentCtl);
    10.  
    11.             // Check if queue empty only if necessary.
    12.             // 线程池终止时需要返回false,避免新的worker被创建
    13.             // 1 先判断runState >= SHUTDOWN
    14.             // 2 runState >= SHUTDOWN时，意味着不再允许创建新的工作线程，但有一种情况例外
    15.             // 即SHUTDOWN状态下(runState == SHUTDOWN)，工作队列不为空(!workQueue.isEmpty())，还需要继续执行
    16.             // 比如在当前存活的线程发生中断异常时，会调用processWorkerExit方法，在销毁原有工作线程后调用addWorker重新创建一个新的（firstTask == null）
    17.             if (runState >= SHUTDOWN && !(runState == SHUTDOWN && firstTask == null && !workQueue.isEmpty())) {
    18.                 // 线程池已经是关闭状态了，不再允许创建新的工作线程，返回false
    19.                 return false;
    20.             }
    21.  
    22.             // 用于cas更新workerCount的内层循环（注意这里面与jdk的写法不同，改写成了逻辑一致但更可读的形式）
    23.             for (;;) {
    24.                 // 判断当前worker数量是否超过了限制
    25.                 int workerCount = workerCountOf(currentCtl);
    26.                 if (workerCount >= CAPACITY) {
    27.                     // 当前worker数量超过了设计上允许的最大限制
    28.                     return false;
    29.                 }
    30.                 if (core) {
    31.                     // 创建的是核心线程，判断当前线程数是否已经超过了指定的核心线程数
    32.                     if (workerCount >= this.corePoolSize) {
    33.                         // 超过了核心线程数，创建核心worker线程失败
    34.                         return false;
    35.                     }
    36.                 } else {
    37.                     // 创建的是非核心线程，判断当前线程数是否已经超过了指定的最大线程数
    38.                     if (workerCount >= this.maximumPoolSize) {
    39.                         // 超过了最大线程数，创建非核心worker线程失败
    40.                         return false;
    41.                     }
    42.                 }
    43.  
    44.                 // cas更新workerCount的值
    45.                 boolean casSuccess = compareAndIncrementWorkerCount(currentCtl);
    46.                 if (casSuccess) {
    47.                     // cas成功，跳出外层循环
    48.                     break retry;
    49.                 }
    50.  
    51.                 // 重新检查一下当前线程池的状态与之前是否一致
    52.                 currentCtl = ctl.get();  // Re-read ctl
    53.                 if (runStateOf(currentCtl) != runState) {
    54.                     // 从外层循环开始continue（因为说明在这期间 线程池的工作状态出现了变化，需要重新判断）
    55.                     continue retry;
    56.                 }
    57.  
    58.                 // compareAndIncrementWorkerCount方法cas争抢失败，重新执行内层循环
    59.             }
    60.         }
    61.  
    62.         boolean workerStarted = false;
    63.         boolean workerAdded = false;
    64.  
    65.         MyWorker newWorker = null;
    66.         try {
    67.             // 创建一个新的worker
    68.             newWorker = new MyWorker(firstTask);
    69.             final Thread myWorkerThread = newWorker.thread;
    70.             if (myWorkerThread != null) {
    71.                 // MyWorker初始化时内部线程创建成功
    72.  
    73.                 // 加锁，防止并发更新
    74.                 final ReentrantLock mainLock = this.mainLock;
    75.                 mainLock.lock();
    76.  
    77.                 try {
    78.                     // Recheck while holding lock.
    79.                     // Back out on ThreadFactory failure or if
    80.                     // shut down before lock acquired.
    81.                     int runState = runStateOf(ctl.get());
    82.  
    83.                     // 重新检查线程池运行状态，满足以下两个条件的任意一个才创建新Worker
    84.                     // 1 runState < SHUTDOWN
    85.                     // 说明线程池处于RUNNING状态正常运行，可以创建新的工作线程
    86.                     // 2 runState == SHUTDOWN && firstTask == null
    87.                     // 说明线程池调用了shutdown，但工作队列不为空，依然需要新的Worker。
    88.                     // firstTask == null标识着其不是因为外部提交新任务而创建新Worker，而是在消费SHUTDOWN前已提交的任务
    89.                     if (runState < SHUTDOWN ||
    90.                             (runState == SHUTDOWN && firstTask == null)) {
    91.                         if (myWorkerThread.isAlive()) {
    92.                             // 预检查线程的状态，刚初始化的worker线程必须是未唤醒的状态
    93.                             throw new IllegalThreadStateException();
    94.                         }
    95.  
    96.                         // 加入worker集合
    97.                         this.workers.add(newWorker);
    98.  
    99.                         int workerSize = workers.size();
    100.                         if (workerSize > largestPoolSize) {
    101.                             // 如果当前worker个数超过了之前记录的最大存活线程数，将其更新
    102.                             largestPoolSize = workerSize;
    103.                         }
    104.  
    105.                         // 创建成功
    106.                         workerAdded = true;
    107.                     }
    108.                 } finally {
    109.                     // 无论是否发生异常，都先将主控锁解锁
    110.                     mainLock.unlock();
    111.                 }
    112.  
    113.                 if (workerAdded) {
    114.                     // 加入成功，启动worker线程
    115.                     myWorkerThread.start();
    116.                     // 标识为worker线程启动成功，并作为返回值返回
    117.                     workerStarted = true;
    118.                 }
    119.             }
    120.         }finally {
    121.             if (!workerStarted) {
    122.                 addWorkerFailed(newWorker);
    123.             }
    124.         }
    125.  
    126.         return workerStarted;
    127.     }
    1.     public boolean remove(Runnable task) {
    2.         boolean removed = workQueue.remove(task);
    3.         // 当一个任务从工作队列中被成功移除，可能此时工作队列为空。尝试判断是否满足线程池中止条件
    4.         tryTerminate();
    5.         return removed;
    6.     }

    如何保证中止过程中不丢失任务？

    1. 通过shutdown关闭线程池时，SHUTDOWN状态的线程池会等待所有剩余的任务执行完毕后再进入TIDYING状态。
    2. 通过shutdownNow关闭线程池时，以返回值的形式将剩余的任务吐出来还给用户

    中止前已提交的任务不会丢失；而中止后线程池也不会再接收新的任务（走拒绝策略）。这两点共同保证了提交的任务不会丢失。

    如何保证线程池最终关闭前，所有工作线程都已退出？

    线程池在收到中止命令进入SHUTDOWN或者STOP状态时，会一直等到工作队列为空且所有工作线程都中止退出后才会推进到TIDYING阶段。
    上面描述的条件是一个复合的条件，其只有在“收到停止指令（进入SHUTDOWN或者STOP状态）”、"工作队列中任务被移除或消费（工作队列为空）"或是“工作线程退出（所有工作线程都中止退出）”这三类事件发生时才有可能满足。
    而判断是否满足条件并推进到TIDYING状态的关键就在tryTerminate方法中。tryTerminate顾名思义便是用于尝试终止线程池的，当上述任意事件触发时便判断是否满足终止条件，如果满足则将线程池推进到TIDYING阶段。
    因此在ThreadPoolExecutor中tryTerminate一共在6个地方被调用，分别是shutdown、shutdownNow、remove、purge、addWorkerFailed和processWorkerExit方法。

    • shutdown、shutdownNow方法触发收到停止指令的事件
    • remove、purge方法触发工作队列中任务被移除的事件
    • addWorkerFailed、processWorkerExit方法触发工作线程退出的事件

    tryTerminate源码分析

    1.    /**
    2.      * 尝试判断是否满足线程池中止条件，如果满足条件，将其推进到最后的TERMINATED状态
    3.      * 注意：必须在任何可能触发线程池中止的场景下调用（例如工作线程退出，或者SHUTDOWN状态下队列工作队列为空等）
    4.      * */
    5.     final void tryTerminate() {
    6.         for (;;) {
    7.             int currentCtl = this.ctl.get();
    8.             if (isRunning(currentCtl)
    9.                     || runStateAtLeast(currentCtl, TIDYING)
    10.                     || (runStateOf(currentCtl) == SHUTDOWN && !workQueue.isEmpty())) {
    11.                 // 1 isRunning(currentCtl)为true，说明线程池还在运行中，不满足中止条件
    12.                 // 2 当前线程池状态已经大于等于TIDYING了，说明之前别的线程可能已经执行过tryTerminate，且通过了这个if校验，不用重复执行了
    13.                 // 3 当前线程池是SHUTDOWN状态，但工作队列中还有任务没处理完，也不满足中止条件
    14.                 // 以上三个条件任意一个满足即直接提前return返回
    15.                 return;
    16.             }
    17.  
    18.             // 有两种场景会走到这里
    19.             // 1 执行了shutdown方法(runState状态为SHUTDOWN)，且当前工作线程已经空了
    20.             // 2 执行了shutdownNow方法(runState状态为STOP)
    21.             // 这个时候需要令所有的工作线程都主动的退出来回收资源
    22.             if (workerCountOf(currentCtl) != 0) {
    23.                 // 如果当前工作线程个数不为0，说明还有别的工作线程在工作中。
    24.                 // 通过interruptIdleWorkers(true)，打断其中的一个idle线程，尝试令其也执行runWorker中的processWorkerExit逻辑，并执行tryTerminate
    25.                 // 被中断的那个工作线程也会执行同样的逻辑（getTask方法返回->processWorkerExit->tryTerminate）
    26.                 // 这样可以一个接着一个的不断打断每一个工作线程，令其逐步的退出（比起一次性的通知所有的idle工作线程，这样相对平滑很多）
    27.                 interruptIdleWorkers(ONLY_ONE);
    28.                 return;
    29.             }
    30.  
    31.             // 线程池状态runState为SHUTDOWN或者STOP，且存活的工作线程个数已经为0了
    32.             // 虽然前面的interruptIdleWorkers是一个一个中断idle线程的，但实际上有的工作线程是因为别的原因退出的（恰好workerCountOf为0了）
    33.             // 所以这里是可能存在并发的，因此通过mainLock加锁防止并发，避免重复的terminated方法调用和termination.signalAll方法调用
    34.             final ReentrantLock mainLock = this.mainLock;
    35.             mainLock.lock();
    36.             try {
    37.                 // cas的设置ctl的值为TIDYING+工作线程个数0（防止与别的地方ctl并发更新）
    38.                 if (ctl.compareAndSet(currentCtl, ctlOf(TIDYING, 0))) {
    39.                     try {
    40.                         // cas成功，调用terminated钩子函数
    41.                         terminated();
    42.                     } finally {
    43.                         // 无论terminated钩子函数是否出现异常
    44.                         // cas的设置ctl的值为TERMINATED最终态+工作线程个数0（防止与别的地方ctl并发更新）
    45.                         ctl.set(ctlOf(TERMINATED, 0));
    46.                         // 通知使用awaitTermination方法等待线程池关闭的其它线程（通过termination.await等待）
    47.                         termination.signalAll();
    48.                     }
    49.                     return;
    50.                 }
    51.             } finally {
    52.                 mainLock.unlock();
    53.             }
    54.  
    55.             // 如果上述对ctl变量的cas操作失败了，则进行重试，再来一次循环
    56.             // else retry on failed CAS
    57.         }
    58.     }

    如何保证工作线程一定能成功退出？

    从上面tryTerminate方法的实现中可以看到，线程池必须等到所有工作线程都全部退出（workerCount为0），工作线程占用的全部资源都回收后才会推进到终止态。
    那么之前启动的工作线程一定能通过processWorkerExit退出并销毁吗？答案是不一定，这主要取决于用户是否正确的编写了令工作线程安全退出的任务逻辑。
    因为只有能退出任务执行逻辑(runWorker方法中的task.run())的工作线程才有机会执行processWorkerExit，无法从任务中跳出(正常退出or抛异常)的工作线程将永远无法退出，导致线程池也永远无法推进到终态。

    下面分情况讨论：

    • 任务中的逻辑是一定会执行完正常结束的（没有无限循环也没有令线程陷入阻塞态的操作）。那么这是没问题的
    1.    ()->{
    2.         // 会正常结束的
    3.         System.out.println("hello world!");
    4.    };
    • 任务中存在需要无限循环的逻辑。那么最好在循环条件内监听一个volatile的变量，当需要线程池停止时，修改这个变量，从而令任务从无限循环中正常退出。
    1.     ()->{
    2.         // 无限循环
    3.         while(true){
    4.             System.out.println("hello world!");
    5.         }
    6.     };
    1.     ()->{
    2.         // 无限循环时监听一个变量
    3.         while(!isStop) {
    4.             System.out.println("hello world!");
    5.         }
    6.     };
    • 任务中存在Condition.await等会阻塞当前线程，令其无法自然退出的逻辑。
      tryTerminate中停止工作线程时会调用Worker类的interruptIfStarted方法发出中断指令（Thread.interrupt方法），如果被阻塞的方法是响应中断的，那么业务代码中不能无脑吞掉InterruptedException，而要能感知到中断异常，在确实要关闭线程池时令任务退出（向上抛异常或正常退出）。
      而如果是不响应中断的阻塞方法（如ReentrantLock.lock）,则需要用户自己保证这些方法最终能够被唤醒，否则工作线程将无法正常退出而阻止线程池进入终止状态。
    1.     ()->{
    2.             try {
    3.                 new ReentrantLock().newCondition().await();
    4.             } catch (InterruptedException e) {
    5.                 // doSomething处理一些逻辑后。。。
    6.                 // 向上抛出异常
    7.                 throw new XXXException(e);
    8.             }
    9.         }
    1.     ()->{
    2.         try {
    3.             new ReentrantLock().newCondition().await();
    4.         } catch (InterruptedException e) {
    5.  
    6.         }
    7.         // doSomething处理一些逻辑后。。。正常退出
    8.     }

     

  • 相关阅读:
    springboot操作es
    CPU架构
    TikTok的伦理挑战：虚拟世界与现实世界的交汇
    【.NET6+WPF】WPF使用prism框架+Unity IOC容器实现MVVM双向绑定和依赖注入
    训练集测试集验证集的区别
    ansible 中的变量及加密
    zookeeper集群搭建Windows 7
    【生信分析】基于TCGA肿瘤数据进行基因共表达网络分析
    改变金融贷款市场营销方式 ---- 运营商大数据精准获客
    安装搭建私有仓库Harbor
  • 原文地址：https://blog.csdn.net/jh035512/article/details/127956903