【JUC并发编程】1 ThreadPoolExecutor线程池源码解析

「爱情、让人受尽委屈。」 2023-10-04 14:57 90阅读 0赞

文章目录

    • 一、为什么要使用线程池？
    • 二、ThreadPoolExecutor源码分析
        1. 常用变量
        1. 构造方法
        1. 任务执行过程execute()
        1. addWorker()方法
        1. runWorker()方法
        1. processWorkerExit()
        • 1）如果线程池中的线程异常终止，线程池的工作线程数量会减少吗？
        1. shutdown() 和 shutdownNow()的区别
    • 三、 总结

一、为什么要使用线程池？

  • 降低资源消耗：通过重复利用已创建的线程 以降低线程创建和销毁带来的额外消耗。
  • 提高响应速度：当任务到达时，可以不需要等待线程创建就能立即执行。
  • 提高线程的可管理性：线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，监控和调优。
  • ThreadPoolExecutor可以保存一些基本数据，例如当前线程池完成的任务数。

二、ThreadPoolExecutor源码分析

ThreadPoolExecutor所在JUC包的作者是做C出身的道格-李，而这位大佬的代码风格是：先改状态后做事，做不通再回滚。

下面我们主要从常用变量、构造方法、任务执行过程execute()、addWorker()、runWorker()、processWorkerExit()去分析线程池的实现原理。

1. 常用变量

ThreadPooleExecutor中有一个AtomicInteger类型的成员变量ctl，其高3位用来表示线程池的状态，低29位用来表示线程池的工作线程数量；成员变量COUNT_BITS的值为29，成员变量CAPACITY用来表示worker的最大个数，即2 ^ 29 -1;

  1. // ctl 可以看做一个int类型的数字，高3位表示线程池状态，低29位表示worker数量
  2. private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
  3. // COUNT_BITS    Integer.SIZE为32，所以COUNT_BITS为29
  4. private static final int COUNT_BITS = Integer.SIZE - 3;
  5. // 线程允许的最大线程数，1左移29位，然后减1，即2^29 -1
  6. private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

线程池状态的获取方式runStateOf()为capacity取反 然后和ctl 做与操作。
线程池工作线程的获取方式workerCountOf()为capacity 和 ctl 做与操作。

  1. // runStateOf() 获取线程池的状态，通过按位与操作，低29位将全部变为0
  2. private static int runStateOf(int c)     {
  3.      return c & ~CAPACITY; }
  4. // workerCountOf() 获取线程池worker数量，通过按位与操作，高3位将全部变成0
  5. private static int workerCountOf(int c)  {
  6.      return c & CAPACITY; }

线程池的状态：

  • RUNNING：接受新任务并且处理阻塞队列里的任务；
  • SHUTDOWN：拒绝新任务但是处理阻塞队列里的任务；
  • STOP：拒绝新任务并且抛弃阻塞队列里的任务，同时会中断正在处理的任务；
  • TIDYING：所有任务都执行完（包含阻塞队列里面任务）当前线程池活动线程为 0，将要调用 terminated 方法；
  • TERMINATED：终止状态，terminated方法调用完成以后的状态。

线程池状态转换：

  1. RUNNING -> SHUTDOWN：显式调用 shutdown() 方法；或隐式调用了 finalize()，finalize()里调用了 shutdown() 方法。
  2. RUNNING or SHUTDOWN -> STOP：显式调用 shutdownNow() 方法时。
  3. SHUTDOWN -> TIDYING：当线程池和任务队列都为空时。
  4. STOP -> TIDYING：当线程池为空时。

  5. TIDYING -> TERMINATED：当 terminated() hook 方法执行完成时。

    // 线程池有5中状态，按大小排序如下：Running< SHUTDOWN < STOP < TIDING < TERMINATED
    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;

    // ctlOf(),根据线程池状态和线程池worker数量，生成ctl值
    private static int ctlOf(int rs, int wc) {

    1.  return rs | wc; }

    // runStateLessThan()，线程池状态小于xx
    private static boolean runStateLessThan(int c, int s) {

    1. return c < s;

    }
    // runStateAtLeast()，线程池状态大于等于xx
    private static boolean runStateAtLeast(int c, int s) {

    1. return c >= s;

    }

2. 构造方法

我们从最全的构造方法来看其中的参数。

  1. public ThreadPoolExecutor(int corePoolSize,
  2.                           int maximumPoolSize,
  3.                           long keepAliveTime,
  4.                           TimeUnit unit,
  5.                           BlockingQueue<Runnable> workQueue,
  6.                           ThreadFactory threadFactory,
  7.                           RejectedExecutionHandler handler) {
  9.     // 基本类型参数校验
  10.     if (corePoolSize < 0 ||
  11.         maximumPoolSize <= 0 ||
  12.         maximumPoolSize < corePoolSize ||
  13.         keepAliveTime < 0)
  14.         throw new IllegalArgumentException();
  15.     // 空指针校验
  16.     if (workQueue == null || threadFactory == null || handler == null)
  17.         throw new NullPointerException();
  18.     this.corePoolSize = corePoolSize;
  19.     this.maximumPoolSize = maximumPoolSize;
  20.     this.workQueue = workQueue;
  21.     // 根据传入参数`unit`和`keepAliveTime`，将存活时间转换为纳秒存到变量`keepAliveTime `中
  22.     this.keepAliveTime = unit.toNanos(keepAliveTime);
  23.     this.threadFactory = threadFactory;
  24.     this.handler = handler;
  25. }

从构造中可以看出：核心线程数不能小于0，最大线程数不能小于1，最大线程数不能小于核心线程数，超过核心线程数的线程存活时间不能小于0。

参数介绍：

  1. corePoolSize：线程池核心线程数。
  2. maximunPoolSize：线程池最大线程数。
  3. keepAliveTime：存活时间。如果当前线程池中的线程数量比核心线程数量要多，并且有闲置状态的线程，keepAliveTime表示这些闲置的线程能存活的最大时间。
  4. TimeUnit，存活时间的时间单位。
  5. workQueue：当任务数超过核心线程数时，用于保存等待执行的任务的阻塞队列；比如基于数组的有界 ArrayBlockingQueue，基于链表的无界 LinkedBlockingQueue，最多只有一个元素的同步队列 SynchronousQueue，优先级队列 PriorityBlockingQueue 等。
  6. threadFactory：创建线程的工厂。
  7. handler：拒绝策略，当队列满了并且线程个数达到 maximunPoolSize 后采取的策略。比如 AbortPolicy （直接抛出异常），CallerRunsPolicy（使用调用者所在线程来运行任务），DiscardOldestPolicy（丢弃一个最老的任务，然后执行当前任务），DiscardPolicy（丢失当前任务，没有异常信息）。

3. 任务执行过程execute()

老生常谈，任务的执行过程为：

  1. 首先获取线程池的信息，判断工作线程数是否小于核心线程数，小于则创建核心线程数；
  2. 如果工作线程数超过核心线程数，则将任务添加到阻塞队列中；
  3. 如果阻塞队列满了且工作线程数小于最大线程数，则创建非核心线程；
  4. 如果工作线程数达到最大线程数、线程池状态不是RUNNING状态、或 任务入队失败，则对新添加的任务执行拒绝策略。

这其中有几个细节：

  • 由于其他线程改变线程池的状态导致添加核心线程失败，当前线程重新获取线程池的状态。
  • 多线程环境下，任务进入阻塞队列之后，线程池状态可能会发生改变，所以需要重新获取线程池状态进行recheck。进而判断线程池状态是不是RUNNING状态：
    1.如果不是，说明执行shutdown命令，需要对新加入的任务执行reject()操作；
    2.如果是，还需要判断当前线程池的工作线程数是否为0，如果核心线程数为0，需要创建一个非核心线程来执行任务；因为创建线程池的时候允许核心线程数为0。

  • 如果线程池状态不是RUNNING状态，或 任务进入阻塞队列失败，会尝试创建worker来执行任务，如果创建worker失败，会执行拒绝策略。

    public void execute(Runnable command) {

    1.     if (command == null)
    2.         throw new NullPointerException();
    3.     // 获取线程池的状态和线程数信息
    4.     int c = ctl.get();
    5.     // 线程数小于核心线程数
    6.     if (workerCountOf(c) < corePoolSize) {
    8.         if (addWorker(command, true))
    9.             return;
    10.         // 其他线程可能会修改线程池的状态，导致核心线程添加失败，所以需要重新获取线程池的状态。
    11.         c = ctl.get();
    12.     }
    13.     // worker数量超过核心线程数，任务添加到队列中
    14.     if (isRunning(c) && workQueue.offer(command)) {
    16.     // 多线程环境下，任务入队之后，线程池状态也可能发生改变，所以需要重新获取线程池状态进行recheck。
    17.         int recheck = ctl.get();
    18.         // 如果线程池状态不是RUNNING状态，说明执行过shutdown命令，需要对新加入的任务执行reject()操作。
    19.         if (! isRunning(recheck) && remove(command))
    20.             reject(command);
    21.         // 这儿为什么需要判断0值呢？主要是因为线程池构造方法中，核心线程数允许为0。所以当线程数为0的时候，需要使用非核心线程来执行任务
    22.         else if (workerCountOf(recheck) == 0)
    23.             // 添加非核心线程
    24.             addWorker(null, false);
    25.     }
    26.     // 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务。
    27.     // 这儿有3点需要注意：
    28.     // 1. 线程池不是运行状态时，addWorker内部会判断线程池状态
    29.     // 2. addWorker第2个参数表示是否创建核心线程
    30.     // 3. addWorker返回false，则说明任务执行失败，需要执行reject操作
    31.     else if (!addWorker(command, false))
    32.         reject(command);
    33. }

4. addWorker()方法

1）worker介绍

Worker和和task是有区别的，task是用户提交的线程任务，而worker则是ThreadPoolExecutor自己内部的一个类，主要是对进行中的任务进行中断控制，顺带着对其他行为进行记录，其保存在HashSet中。另外，Worker类继承了AQS，其本身就是一把锁，在runWorker()方法中会用到。

  1. private final HashSet<Worker> workers = new HashSet<Worker>();
  3. private final class Worker
  4.     extends AbstractQueuedSynchronizer
  5.     implements Runnable
  6. {
  8.     /**
  9.          * This class will never be serialized, but we provide a
  10.          * serialVersionUID to suppress a javac warning.
  11.          */
  12.     private static final long serialVersionUID = 6138294804551838833L;
  14.     // 正在跑的线程，如果是null标识factory失败
  15.     final Thread thread;
  16.     // 初始化一个任务以运行，可能为空
  17.     Runnable firstTask;
  18.     // 每个线程的完成任务数
  19.     volatile long completedTasks;
  21.     /**
  22.          * Creates with given first task and thread from ThreadFactory.
  23.          * @param firstTask the first task (null if none)
  24.          */
  25.     Worker(Runnable firstTask) {
  27.         setState(-1); // inhibit interrupts until runWorker
  28.         this.firstTask = firstTask;
  29.         // 这儿是Worker的关键所在，使用了线程工厂创建一个线程。传入的参数为当前Worker
  30.         this.thread = getThreadFactory().newThread(this);
  31.     }
  33.     /** Delegates main run loop to outer runWorker  */
  34.     public void run() {
  36.         runWorker(this);
  37.     }
  39.     // Lock methods
  40.     //
  41.     // The value 0 represents the unlocked state.
  42.     // The value 1 represents the locked state.
  44.     protected boolean isHeldExclusively() {
  46.         return getState() != 0;
  47.     }
  49.     // 获取锁
  50.     protected boolean tryAcquire(int unused) {
  52.         if (compareAndSetState(0, 1)) {
  54.             setExclusiveOwnerThread(Thread.currentThread());
  55.             return true;
  56.         }
  57.         return false;
  58.     }
  60.     // 释放锁
  61.     protected boolean tryRelease(int unused) {
  63.         setExclusiveOwnerThread(null);
  64.         setState(0);
  65.         return true;
  66.     }
  68.     public void lock()        {
  69.      acquire(1); }
  70.     public boolean tryLock()  {
  71.      return tryAcquire(1); }
  72.     public void unlock()      {
  73.      release(1); }
  74.     public boolean isLocked() {
  75.      return isHeldExclusively();}
  76. }

2）addWorker()分析
添加工作线程主要分为两步：判断线程池状态并使用CAS增加工作线程数量、封装Worker并启动工作线程。

第一步：

  • 首先会在retry外层循环中，保存一个CTL的快照，用于后面进行CAS判断线程池的状态是否发生改变。
  • 然后，判断线程池的状态决定是否直接返回false；有三种情况会直接返回FALSE，如下：
    1.当线程池的状态大于SHUTDOWN时；
    2.当线程池的状态等于SHUTDOWN，且firstTask不为null时；
    3.当线程池的状态等于SHUTDOWN，且任务队列为空时；
  • 接着，内循环中获取线程池的工作线程数量，如果core参数为TRUE，即插入核心线程，会判断当前线程数是否大于等于corePoolSize；或者判断当前工作线程数量是否大于等于 工作线程数的最大值Capacity；大于则直接返回FALSE；
  • 然后，尝试使用CAS增加工作线程数，增加成功则退出外层retry循环。
  • 接着，获取线程池的信息ctl，和最初的CTL快照值进行比较，如果线程池状态发生改变，continue 外层循环retry，对外层循环进行自旋。
  • 最后，如果CAS改变工作线程数失败，会进行内循环尝试改变线程池的工作线程数。

第二步：

  • 先定义了一个workerStarted变量，用于判断工作线程是否正常启动；如果启动异常，会调用addWorkerFailed()方法：对workers加独占锁，回滚减少工作线程数量、中断当前工作线程；最后返回FALSE。
  • 接着，将当前任务封装为一个新的Worker；因为Worker的添加必须保证是线程安全的，所以需要加ReentrantLock独占锁。
  • 然后会recheck线程池状态，因为ThreadFactory出现故障 或 在获取锁之前 关闭的话，直接运行finally块中的addWorkerFailed()方法退出即可。
  • 只有当线程池的状态是RUNNING状态 或 线程池的状态是SHUTDOWN且firstTask为null时 才会继续执行线程启动相关后续逻辑：包括check工作线程是否已经启动、添加工作线程到workers，更新largestPoolSize。

  • 最后，释放锁，启动工作线程，将workerStarted变量设置为true并返回。

    private boolean addWorker(Runnable firstTask, boolean core) {

    1. // 1. 判断状态并增加工作线程数量
    2. retry:
    3. // 外层自旋
    4. for (;;) {
    6.     //存一个快照值，后面进行CAS，如果最新的结果不等于快照值，就再跑一遍
    7.     int c = ctl.get();
    8.     int rs = runStateOf(c);
    10.     // 1. 线程池状态大于SHUTDOWN时，直接返回false
    11.     // 2. 线程池状态等于SHUTDOWN，且firstTask不为null，直接返回false
    12.     // 3. 线程池状态等于SHUTDOWN，且队列为空，直接返回false
    13.     if (rs >= SHUTDOWN &&
    14.         ! (rs == SHUTDOWN &&
    15.            firstTask == null &&
    16.            ! workQueue.isEmpty()))
    17.         return false;
    19.     // 内层自旋
    20.     for (;;) {
    22.         int wc = workerCountOf(c);
    23.         // worker数量超过容量，直接返回false
    24.         if (wc >= CAPACITY ||
    25.             wc >= (core ? corePoolSize : maximumPoolSize))
    26.             return false;
    27.         // 使用CAS的方式增加worker数量。
    28.         // 若增加成功，则直接跳出外层循环进入到第二部分
    29.         if (compareAndIncrementWorkerCount(c))
    30.             break retry;
    31.         c = ctl.get();  // Re-read ctl
    32.         // 线程池状态发生变化，对外层循环进行自旋
    33.         if (runStateOf(c) != rs)
    34.             continue retry;
    35.         // 其他情况，直接内层循环进行自旋即可
    36.         // else CAS failed due to workerCount change; retry inner loop
    37.     } 
    38. }
    40. // 2. 封装worker并启动工作线程
    41. boolean workerStarted = false;
    42. boolean workerAdded = false;
    43. Worker w = null;
    44. try {
    46.     w = new Worker(firstTask);
    47.     final Thread t = w.thread;
    48.     if (t != null) {
    50.         final ReentrantLock mainLock = this.mainLock;
    51.         // worker的添加必须是串行的，因此需要加锁
    52.         mainLock.lock();
    53.         try {
    55.             // Recheck while holding lock.
    56.             // Back out on ThreadFactory failure or if
    57.             // shut down before lock acquired.
    58.             // 这儿需要重新检查线程池状态
    59.             int rs = runStateOf(ctl.get());
    61.             if (rs < SHUTDOWN ||
    62.                 (rs == SHUTDOWN && firstTask == null)) {
    64.                 // worker已经调用过了start()方法，则不再创建worker
    65.                 if (t.isAlive()) // precheck that t is startable
    66.                     throw new IllegalThreadStateException();
    67.                 // worker创建并添加到workers成功
    68.                 workers.add(w);
    69.                 // 更新`largestPoolSize`变量
    70.                 int s = workers.size();
    71.                 if (s > largestPoolSize)
    72.                     largestPoolSize = s;
    73.                 workerAdded = true;
    74.             }
    75.         } finally {
    77.             mainLock.unlock();
    78.         }
    79.         // 启动worker线程
    80.         if (workerAdded) {
    82.             t.start();
    83.             workerStarted = true;
    84.         }
    85.     }
    86. } finally {
    88.     // worker线程启动失败，说明线程池状态发生了变化（关闭操作被执行），需要进行shutdown相关操作
    89.     if (! workerStarted)
    90.         addWorkerFailed(w);
    91. }
    92. return workerStarted;

    }

5. runWorker()方法

runWorker()方法是Worker的运行逻辑：

  • runWorker会先获取Worker中的firstTask，然后将worker的firstTask引用置为null，接着为了让外部可以中断Worker的运行，会先给Worker解锁。
  • 然后while 自旋调用getTask()方法从任务阻塞队列中获取任务，接着，对Worker进行加锁，保证每个Worker中任务的执行是串行的。
  • 如果线程池正在停止，对当前线程进行中断操作；因为在if条件判断中和代码段中会分别执行一次interrupt()方法，所以会确保执行shutdownNow时，清除中断标志位，中断线程。
  • 然后执行任务，并预留beforeExecute()和afterExecute()两个钩子函数。最后将completedTasks + 1。

此外：runWorker中最外层的try，太容易的是任务的异常，它没有catch，所以监控不到异常。

另外：runWorker()方法中有一个completedAbruptly变量，用来表示任务的执行有没有被异常中断，进而在processWorkerExit()方法保证线程池中的可执行工作线程数量不变。

completedAbruptly为true的三种情况：

  • 调用getTask()从队列获取任务时线程被中断，出现异常。但是这里给包掉了，不会抛出异常。
  • beforeExecute()钩子函数出异常。

  • afterExecute()钩子函数出异常。

    final void runWorker(Worker w) {

    1. Thread wt = Thread.currentThread();
    2. Runnable task = w.firstTask;
    3. w.firstTask = null;
    4. // 调用unlock()是为了让外部可以中断，在interruptedIdleWorkers方法中会中断空闲线程
    5. w.unlock();
    6. // 这个变量用于判断是否进入过自旋(while循环)
    7. boolean completedAbruptly = true;
    8. try {
    10.     // 这儿是自旋
    11.     // 1. 如果firstTask不为null，则执行firstTask；
    12.     // 2. 如果firstTask为null，则调用getTask()从队列获取任务。
    13.     // 3. 阻塞队列的特性就是：当队列为空时，当前线程会被阻塞等待
    14.     while (task != null || (task = getTask()) != null) {
    16.       // 这儿对worker进行加锁，是为了达到下面的目的
    17.         // 1. 降低锁范围，提升性能
    18.         // 2. 保证每个worker执行的任务是串行的  
    19.         w.lock();
    20.         // 如果线程池正在停止，则对当前线程进行中断操作,这里要确保执行shutdownNow时，清除中断标志位，中断线程。shutdown不会走这里。
    21.         if ((runStateAtLeast(ctl.get(), STOP) ||
    22.              (Thread.interrupted() &&
    23.               runStateAtLeast(ctl.get(), STOP))) &&
    24.             !wt.isInterrupted())
    25.             wt.interrupt();
    26.         // 执行任务，且在执行前后通过`beforeExecute()`和`afterExecute()`来扩展其功能。
    27.         // 这两个方法在当前类里面为空实现。
    28.         try {
    30.             beforeExecute(wt, task);
    31.             Throwable thrown = null;
    32.             try {
    34.                 task.run();
    35.             } catch (RuntimeException x) {
    37.                 thrown = x; throw x;
    38.             } catch (Error x) {
    40.                 thrown = x; throw x;
    41.             } catch (Throwable x) {
    43.                 thrown = x; throw new Error(x);
    44.             } finally {
    46.                 afterExecute(task, thrown);
    47.             }
    48.         } finally {
    50.             // 帮助gc
    51.             task = null;
    52.             w.completedTasks++;
    53.             w.unlock();
    54.         }
    55.     }
    56.     completedAbruptly = false;
    57. } finally {
    59.     // 自旋操作被退出，说明线程池正在结束
    60.     processWorkerExit(w, completedAbruptly);
    61. }

    }

6. processWorkerExit()

  • processWorkerExit()方法中第一步会先判断Worker是否为异常终止(突然完成)，如果是则减少工作线程。
  • Worker任务完成数+1，然后将Worker从workers中移除。接着调用tryTerminate()方法尝试结束线程池：
    1.当线程池状态是RUNNING状态、小于TIDYING状态 或 线程池状态是SHUTDOWN且任务队列不为空时，直接返回。
    2.当工作线程不为0时，尝试中断一个worker。这里只中断一个worker是因为多线程环境下，可能会存在多个worker并发终止Worker，要确保最后留一个线程用来修改线程池的状态。

  • 最后获取线程池状态，当线程池状态为RUNNING、SHUTDOWN时，判断Worker是否为异常终止，如果是，直接创建一个非核心线程；否者，根据allowCoreThreadTimeOut参数判断出最小线程数min，如果min == 0 并且任务队列不为空，创建一个非核心线程；如果工作线程大于min，则直接返回。

    private void processWorkerExit(Worker w, boolean completedAbruptly) {

    1. // 如果由于用户原因出现异常，直接减少工作线程
    2. if (completedAbruptly) 
    3.     decrementWorkerCount();
    5. final ReentrantLock mainLock = this.mainLock;
    6. mainLock.lock();
    7. try {
    9.     completedTaskCount += w.completedTasks;
    10.     workers.remove(w);
    11. } finally {
    13.     mainLock.unlock();
    14. }
    16. tryTerminate();
    18. int c = ctl.get();
    19. // 线程池状态是运行、或shutdown
    20. if (runStateLessThan(c, STOP)) {
    22.     if (!completedAbruptly) {
    24.         int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
    25.         if (min == 0 && ! workQueue.isEmpty())
    26.             min = 1;
    27.         if (workerCountOf(c) >= min)
    28.             return; // replacement not needed
    29.     }
    30.     // 新建一个非核心线程
    31.     addWorker(null, false);
    32. }

    }

tryTerminate()中断线程方法

方法里通过判断workerCount是否为零来保证只让最后一个线程进行线程池状态斜体样式

  1. final void tryTerminate() {
  3.     for (;;) {
  5.         int c = ctl.get();
  6.         if (isRunning(c) ||
  7.             runStateAtLeast(c, TIDYING) ||
  8.             (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
  9.             return;
  10.         // 保证其他空闲工作线程全部被中断，只让最后一个线程改变线程池状态
  11.         if (workerCountOf(c) != 0) {
  13.             interruptIdleWorkers(ONLY_ONE);
  14.             return;
  15.         }
  17.         final ReentrantLock mainLock = this.mainLock;
  18.         mainLock.lock();
  19.         try {
  21.             if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
  23.                 try {
  25.                     terminated();
  26.                 } finally {
  28.                     ctl.set(ctlOf(TERMINATED, 0));
  29.                     termination.signalAll();
  30.                 }
  31.                 return;
  32.             }
  33.         } finally {
  35.             mainLock.unlock();
  36.         }
  37.         // else retry on failed CAS
  38.     }
  39. }
1）如果线程池中的线程异常终止，线程池的工作线程数量会减少吗？

不会，因为执行线程逻辑的runWorker()中，finally代码块中会执行processWorkerExit()方法，其中如果线程异常终止，会先减少工作线程数量，如果异常中断标志completedAbruptly为true，会直接调用addWorker()方法创建一个非核心线程。实际上addWorker()添加核心线程还是非核心线程，对线程而言并没有区别，只是会判断是否超过核心线程数还是最大线程数。

7. shutdown() 和 shutdownNow()的区别

  • shutdown，不再接收新任务但会执行queue中的任务；shutdownNow不接收新任务也不执行queue中的任务。
  • shutdown将线程状态使用CAS更改为SHUTDOWN；shudownNow更改为STOP。
  • shutdown中断空闲工作线程，shutdownNow中断所有工作线程。

如何判断一个线程是否是空闲线程？
看其在runWorker方法中有没有获取到锁，有则不是。

三、 总结

线程池采用了大量的CAS + 快照操作，确保线程池的状态和工作线程数量的正确变更。

发表评论

表情：
评论列表 （有 0 条评论，90人围观）

还没有评论，来说两句吧...

相关阅读