Java线程池ThreadPoolExecutor原理分析

简介

线程池可以简单看做是一组线程的集合,通过使用线程池,我们可以方便的复用线程,避免了频繁创建和销毁线程所带来的开销。

类图

如上图,最顶层的接口 Executor 仅声明了一个方法execute。ExecutorService 接口在其父类接口基础上,声明了包含但不限于shutdown、submit、invokeAll、invokeAny 等方法。至于 ScheduledExecutorService 接口,则是声明了一些和定时任务相关的方法,比如 schedule和scheduleAtFixedRate。线程池的核心实现是在 ThreadPoolExecutor 类中,我们使用 Executors 调用newFixedThreadPool、newSingleThreadExecutor和newCachedThreadPool等方法创建线程池均是 ThreadPoolExecutor 类型。

原理分析

  1. 构造函数
    线程池的核心实现即 ThreadPoolExecutor 类。该类包含了几个核心属性,这些属性在可在构造方法进行初始化。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    /**
    * 概念:
    * corePoolSize:核心线程数,当线程数小于该值时,线程池会优先创建新线程来执行新任务
    * maximumPoolSize:线程池所能维护的最大线程数
    * keepAliveTime:空闲线程的存活时间
    * workQueue:任务队列,用于缓存未执行的任务
    * threadFactory:线程工厂。可通过工厂为新建的线程设置更有意义的名字
    * handler:拒绝策略。当线程池和任务队列均处于饱和状态时,使用拒绝策略处理新任务。默认是 AbortPolicy,即直接抛出异常
    *
    * 规则流程:
    * 1.线程数量小于 corePoolSize,直接创建新线程处理新的任务。
    * 2.线程数量大于等于 corePoolSize,workQueue 未满,则缓存新任务。
    * 3.线程数量大于等于 corePoolSize,但小于 maximumPoolSize,且 workQueue 已满。则创建新线程处理新任务
    * 4.线程数量大于等于 maximumPoolSize,且 workQueue 已满,则使用拒绝策略处理新任务
    *
    * 排队策略(当线程数量大于等于 corePoolSize,workQueue 未满时,则缓存新任务):
    * 同步队列(SynchronousQueue):该队列不存储元素,每个插入操作必须等待另一个线程调用移除操作,否则插入操作会一直阻塞
    * 有界队列(ArrayBlockingQueue):基于数组的阻塞队列,按照 FIFO 原则对元素进行排序
    * 无界队列(LinkedBlockingQueue):基于链表的阻塞队列,按照 FIFO 原则对元素进行排序
    * 优先级队列(PriorityBlockingQueue):具有优先级的阻塞队列
    *
    * 拒绝策略(线程数量大于等于 maximumPoolSize,且 workQueue 已满,则使用拒绝策略处理新任务。):
    * AbortPolicy:丢弃新任务,并抛出 RejectedExecutionException
    * DiscardPolicy:不做任何操作,直接丢弃新任务
    * DiscardOldestPolicy:丢弃队列队首的元素,并执行新任务
    * CallerRunsPolicy:由调用线程执行新任务
    */
    public ThreadPoolExecutor(int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
    Executors.defaultThreadFactory(), defaultHandler);
    }

    public ThreadPoolExecutor(int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
    threadFactory, defaultHandler);
    }

    public ThreadPoolExecutor(int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    RejectedExecutionHandler handler) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
    Executors.defaultThreadFactory(), handler);
    }

    public ThreadPoolExecutor(int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
    maximumPoolSize <= 0 ||
    maximumPoolSize < corePoolSize ||
    keepAliveTime < 0)
    throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
    throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
    }
  2. 状态和线程数

线程池状态

  • RUNNING:这个没什么好说的,这是最正常的状态:接受新的任务,处理等待队列中的任务
  • SHUTDOWN:不接受新的任务提交,但是会继续处理等待队列中的任务
  • STOP:不接受新的任务提交,不再处理等待队列中的任务,中断正在执行任务的线程
  • TIDYING:所有的任务都销毁了,workCount 为 0。线程池的状态在转换为 TIDYING 状态时,会执行钩子方法 terminated()
  • TERMINATED:terminated() 方法结束后,线程池的状态就会变成这个

状态转换

  • RUNNING -> SHUTDOWN:当调用了 shutdown() 后,会发生这个状态转换,这也是最重要的
  • (RUNNING or SHUTDOWN) -> STOP:当调用 shutdownNow() 后,会发生这个状态转换,这下要清楚 shutDown() 和 shutDownNow() 的区别了
  • SHUTDOWN -> TIDYING:当任务队列和线程池都清空后,会由 SHUTDOWN 转换为 TIDYING
  • STOP -> TIDYING:当任务队列清空后,发生这个转换
  • TIDYING -> TERMINATED:这个前面说了,当 terminated() 方法结束后
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// 采用一个 32 位的整数来存放线程池的状态和当前池中的线程数,其中高 3 位用于存放线程池状态,低 29 位表示线程数(即使只有 29 位,也已经不小了,大概 500 多万)
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 这里 COUNT_BITS 设置为 29(32-3),意味着前三位用于存放线程状态,后29位用于存放线程数
private static final int COUNT_BITS = Integer.SIZE - 3;
// 000 11111111111111111111111111111
// 这里得到的是 29 个 1,也就是说线程池的最大线程数是 2^29-1=536860911
private static final int CAPACITY = (1 << COUNT_BITS) - 1;

// 我们说了,线程池的状态存放在高 3 位中
// 运算结果为 111跟29个0:111 00000000000000000000000000000
// 备注:负数的二进制用正数的补码表示(https://blog.csdn.net/onewalkingman/article/details/3746154)
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS;
// 000 00000000000000000000000000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 001 00000000000000000000000000000
private static final int STOP = 1 << COUNT_BITS;
// 010 00000000000000000000000000000
private static final int TIDYING = 2 << COUNT_BITS;
// 011 00000000000000000000000000000
private static final int TERMINATED = 3 << COUNT_BITS;

// 将整数 c 的低 29 位修改为 0,就得到了线程池的状态
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 将整数 c 的高 3 为修改为 0,就得到了线程池中的线程数
private static int workerCountOf(int c) { return c & CAPACITY; }
  1. 线程池中做任务的线程Worker

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
    {
    /**
    * This class will never be serialized, but we provide a
    * serialVersionUID to suppress a javac warning.
    */
    private static final long serialVersionUID = 6138294804551838833L;

    //真正的线程
    /** Thread this worker is running in. Null if factory fails. */
    final Thread thread;

    /**
    * 为什么叫 firstTask?因为在创建线程的时候,如果同时指定了这个线程起来以后需要执行的第一个任务,
    * 那么第一个任务就是存放在这里的(线程可不止执行这一个任务)
    * 当然了,也可以为 null,这样线程起来了,自己到任务队列(BlockingQueue)中取任务(getTask 方法)就行了
    */
    /** Initial task to run. Possibly null. */
    Runnable firstTask;
    //用于存放此线程完全的任务数,注意了,这里用了 volatile,保证可见性
    /** Per-thread task counter */
    volatile long completedTasks;

    /**
    * Worker 只有这一个构造方法,传入 firstTask,也可以传 null
    * Creates with given first task and thread from ThreadFactory.
    * @param firstTask the first task (null if none)
    */
    Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    // 调用 ThreadFactory 来创建一个新的线程
    this.thread = getThreadFactory().newThread(this);
    }

    //这里调用了外部类的 runWorker 方法
    /** Delegates main run loop to outer runWorker */
    public void run() {
    runWorker(this);
    }

    //以下是aqs独占锁实现...
    }
  2. execute方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    /**
    * Executes the given task sometime in the future. The task
    * may execute in a new thread or in an existing pooled thread.
    *
    * If the task cannot be submitted for execution, either because this
    * executor has been shutdown or because its capacity has been reached,
    * the task is handled by the current {@code RejectedExecutionHandler}.
    *
    * @param command the task to execute
    * @throws RejectedExecutionException at discretion of
    * {@code RejectedExecutionHandler}, if the task
    * cannot be accepted for execution
    * @throws NullPointerException if {@code command} is null
    */
    public void execute(Runnable command) {
    if (command == null)
    throw new NullPointerException();
    /*
    * Proceed in 3 steps:
    *
    * 1. If fewer than corePoolSize threads are running, try to
    * start a new thread with the given command as its first
    * task. The call to addWorker atomically checks runState and
    * workerCount, and so prevents false alarms that would add
    * threads when it shouldn't, by returning false.
    *
    * 2. If a task can be successfully queued, then we still need
    * to double-check whether we should have added a thread
    * (because existing ones died since last checking) or that
    * the pool shut down since entry into this method. So we
    * recheck state and if necessary roll back the enqueuing if
    * stopped, or start a new thread if there are none.
    *
    * 3. If we cannot queue task, then we try to add a new
    * thread. If it fails, we know we are shut down or saturated
    * and so reject the task.
    */
    //获取运行状态和线程数
    int c = ctl.get();
    // 如果当前线程数少于核心线程数,那么直接添加一个 worker 来执行任务,
    // 创建一个新的线程,并把当前任务 command 作为这个线程的第一个任务(firstTask)
    if (workerCountOf(c) < corePoolSize) {
    // 添加任务成功,那么就结束了。提交任务嘛,线程池已经接受了这个任务,这个方法也就可以返回了
    // 至于执行的结果,到时候会包装到 FutureTask 中。
    // 返回 false 代表线程池不允许提交任务
    if (addWorker(command, true))
    return;
    c = ctl.get();
    }
    // 如果线程池处于 RUNNING 状态,把这个任务添加到任务队列 workQueue 中
    if (isRunning(c) && workQueue.offer(command)) {
    /* 如果任务进入了 workQueue,我们是否需要开启新的线程
    * 因为线程数在 [0, corePoolSize) 是无条件开启新的线程
    * 如果线程数已经大于等于 corePoolSize,那么将任务添加到队列中,然后进到这里
    */
    int recheck = ctl.get();
    // 如果线程池已不处于 RUNNING 状态,那么移除已经入队的这个任务,并且执行拒绝策略
    if (! isRunning(recheck) && remove(command))
    reject(command);
    // 如果线程池还是 RUNNING 的,并且线程数为 0,那么开启新的线程
    else if (workerCountOf(recheck) == 0)
    addWorker(null, false);
    }
    // 如果 workQueue 队列满了,那么进入到这个分支
    // 以 maximumPoolSize 为界创建新的 worker,
    // 如果失败,说明当前线程数已经达到 maximumPoolSize,执行拒绝策略
    else if (!addWorker(command, false))
    reject(command);
    }

    final void reject(Runnable command) {
    // 执行拒绝策略
    handler.rejectedExecution(command, this);
    }
  3. addWorker方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    /**
    * 第一个参数是准备提交给这个线程执行的任务,之前说了,可以为 null
    * 第二个参数为 true 代表使用核心线程数 corePoolSize 作为创建线程的界线,也就说创建这个线程的时候,
    * 如果线程池中的线程总数已经达到 corePoolSize,那么不能响应这次创建线程的请求
    * 如果是 false,代表使用最大线程数 maximumPoolSize 作为界线
    * 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 (;;) {
    int c = ctl.get();
    int rs = runStateOf(c);
    // 如果线程池已关闭,并满足以下条件之一,那么不创建新的 worker:
    // 1. 线程池状态大于 SHUTDOWN,其实也就是 STOP, TIDYING, 或 TERMINATED
    // 2. firstTask != null
    // 3. workQueue.isEmpty()
    // 简单分析下:
    // 还是状态控制的问题,当线程池处于 SHUTDOWN 的时候,不允许提交任务,但是已有的任务继续执行
    // 当状态大于 SHUTDOWN 时,不允许提交任务,且中断正在执行的任务
    // 多说一句:如果线程池处于 SHUTDOWN,但是 firstTask 为 null,且 workQueue 非空,那么是允许创建 worker 的
    // 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;
    // 由于有并发,重新再读取一下 ctl
    c = ctl.get(); // Re-read ctl
    // 正常如果是 CAS 失败的话,进到下一个里层的for循环就可以了
    // 可是如果是因为其他线程的操作,导致线程池的状态发生了变更,如有其他线程关闭了这个线程池
    // 那么需要回到外层的for循环
    if (runStateOf(c) != rs)
    continue retry;
    // else CAS failed due to workerCount change; retry inner loop
    }
    }

    //到这里,我们认为在当前这个时刻,可以开始创建线程来执行任务了
    // worker 是否已经启动
    boolean workerStarted = false;
    // 是否已将这个 worker 添加到 workers 这个 HashSet 中
    boolean workerAdded = false;
    Worker w = null;
    try {
    // 把 firstTask 传给 worker 的构造方法
    w = new Worker(firstTask);
    // 取 worker 中的线程对象,之前说了,Worker的构造方法会调用 ThreadFactory 来创建一个新的线程
    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());
    // 小于 SHUTTDOWN 那就是 RUNNING,这个自不必说,是最正常的情况
    // 如果等于 SHUTDOWN,前面说了,不接受新的任务,但是会继续执行等待队列中的任务
    if (rs < SHUTDOWN ||
    (rs == SHUTDOWN && firstTask == null)) {
    // worker 里面的 thread 可不能是已经启动的
    if (t.isAlive()) // precheck that t is startable
    throw new IllegalThreadStateException();
    // 加到 workers 这个 HashSet 中
    workers.add(w);
    int s = workers.size();
    // largestPoolSize 用于记录 workers 中的个数的最大值
    // 因为 workers 是不断增加减少的,通过这个值可以知道线程池的大小曾经达到的最大值
    if (s > largestPoolSize)
    largestPoolSize = s;
    workerAdded = true;
    }
    } finally {
    mainLock.unlock();
    }
    // 添加成功的话,启动这个线程
    if (workerAdded) {
    t.start();
    workerStarted = true;
    }
    }
    } finally {
    // 如果线程没有启动,需要做一些清理工作,如前面 workCount 加了 1,将其减掉
    if (! workerStarted)
    addWorkerFailed(w);
    }
    // 返回线程是否启动成功
    return workerStarted;
    }
  4. addWorkFailed方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    // workers 中删除掉相应的 worker
    // workCount 减 1
    private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
    if (w != null)
    workers.remove(w);
    decrementWorkerCount();
    // rechecks for termination, in case the existence of this worker was holding up termination
    tryTerminate();
    } finally {
    mainLock.unlock();
    }
    }
  5. runWorker方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    // Worker 类的 run() 方法
    public void run() {
    runWorker(this);
    }

    /**
    * 此方法由 worker 线程启动后调用,这里用一个 while 循环来不断地从等待队列中获取任务并执行
    * 前面说了,worker 在初始化的时候,可以指定 firstTask,那么第一个任务也就可以不需要从队列中获取
    * Main worker run loop. Repeatedly gets tasks from queue and
    * executes them, while coping with a number of issues:
    *
    * 1. We may start out with an initial task, in which case we
    * don't need to get the first one. Otherwise, as long as pool is
    * running, we get tasks from getTask. If it returns null then the
    * worker exits due to changed pool state or configuration
    * parameters. Other exits result from exception throws in
    * external code, in which case completedAbruptly holds, which
    * usually leads processWorkerExit to replace this thread.
    *
    * 2. Before running any task, the lock is acquired to prevent
    * other pool interrupts while the task is executing, and then we
    * ensure that unless pool is stopping, this thread does not have
    * its interrupt set.
    *
    * 3. Each task run is preceded by a call to beforeExecute, which
    * might throw an exception, in which case we cause thread to die
    * (breaking loop with completedAbruptly true) without processing
    * the task.
    *
    * 4. Assuming beforeExecute completes normally, we run the task,
    * gathering any of its thrown exceptions to send to afterExecute.
    * We separately handle RuntimeException, Error (both of which the
    * specs guarantee that we trap) and arbitrary Throwables.
    * Because we cannot rethrow Throwables within Runnable.run, we
    * wrap them within Errors on the way out (to the thread's
    * UncaughtExceptionHandler). Any thrown exception also
    * conservatively causes thread to die.
    *
    * 5. After task.run completes, we call afterExecute, which may
    * also throw an exception, which will also cause thread to
    * die. According to JLS Sec 14.20, this exception is the one that
    * will be in effect even if task.run throws.
    *
    * The net effect of the exception mechanics is that afterExecute
    * and the thread's UncaughtExceptionHandler have as accurate
    * information as we can provide about any problems encountered by
    * user code.
    *
    * @param w the worker
    */
    final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    // 该线程的第一个任务(如果有的话)
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
    // 循环调用 getTask 获取任务
    while (task != null || (task = getTask()) != null) {
    w.lock();
    // 如果线程池状态大于等于 STOP,那么意味着该线程也要中断
    // If pool is stopping, ensure thread is interrupted;
    // if not, ensure thread is not interrupted. This
    // requires a recheck in second case to deal with
    // shutdownNow race while clearing interrupt
    if ((runStateAtLeast(ctl.get(), STOP) ||
    (Thread.interrupted() &&
    runStateAtLeast(ctl.get(), STOP))) &&
    !wt.isInterrupted())
    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 {
    // 也是一个钩子方法,将 task 和异常作为参数,留给需要的子类实现
    afterExecute(task, thrown);
    }
    } finally {
    // 置空 task,准备 getTask 获取下一个任务
    task = null;
    // 累加完成的任务数
    w.completedTasks++;
    // 释放掉 worker 的独占锁
    w.unlock();
    }
    }
    completedAbruptly = false;
    } finally {
    // 如果到这里,需要执行线程关闭:
    // 1. 说明 getTask 返回 null,也就是说,这个 worker 的使命结束了,执行关闭
    // 2. 任务执行过程中发生了异常
    // 第一种情况,已经在代码处理了将 workCount 减 1,这个在 getTask 方法分析中会说
    // 第二种情况,workCount 没有进行处理,所以需要在 processWorkerExit 中处理
    processWorkerExit(w, completedAbruptly);
    }
    }
  6. getTask方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    /**
    * 此方法有三种可能:
    * 1. 阻塞直到获取到任务返回。我们知道,默认 corePoolSize 之内的线程是不会被回收的,
    * 它们会一直等待任务
    * 2. 超时退出。keepAliveTime 起作用的时候,也就是如果这么多时间内都没有任务,那么应该执行关闭
    * 3. 如果发生了以下条件,此方法必须返回 null:
    * - 池中有大于 maximumPoolSize 个 workers 存在(通过调用 setMaximumPoolSize 进行设置)
    * - 线程池处于 SHUTDOWN,而且 workQueue 是空的,前面说了,这种不再接受新的任务
    * - 线程池处于 STOP,不仅不接受新的线程,连 workQueue 中的线程也不再执行
    * Performs blocking or timed wait for a task, depending on
    * current configuration settings, or returns null if this worker
    * must exit because of any of:
    * 1. There are more than maximumPoolSize workers (due to
    * a call to setMaximumPoolSize).
    * 2. The pool is stopped.
    * 3. The pool is shutdown and the queue is empty.
    * 4. This worker timed out waiting for a task, and timed-out
    * workers are subject to termination (that is,
    * {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
    * both before and after the timed wait, and if the queue is
    * non-empty, this worker is not the last thread in the pool.
    *
    * @return task, or null if the worker must exit, in which case
    * workerCount is decremented
    */
    private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
    int c = ctl.get();
    int rs = runStateOf(c);

    // Check if queue empty only if necessary.
    if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
    decrementWorkerCount();
    return null;
    }

    int wc = workerCountOf(c);

    // 允许核心线程数内的线程回收,或当前线程数超过了核心线程数,那么有可能发生超时关闭
    // Are workers subject to culling?
    boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

    if ((wc > maximumPoolSize || (timed && timedOut))
    && (wc > 1 || workQueue.isEmpty())) {
    if (compareAndDecrementWorkerCount(c))
    return null;
    continue;
    }

    try {
    // 到 workQueue 中获取任务
    Runnable r = timed ?
    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
    workQueue.take();
    if (r != null)
    return r;
    timedOut = true;
    } catch (InterruptedException retry) {
    // 如果此 worker 发生了中断,采取的方案是重试
    // 解释下为什么会发生中断,这个读者要去看 setMaximumPoolSize 方法,
    // 如果开发者将 maximumPoolSize 调小了,导致其小于当前的 workers 数量,
    // 那么意味着超出的部分线程要被关闭。重新进入 for 循环,自然会有部分线程会返回 null
    timedOut = false;
    }
    }
    }

    总结

    ThreadPoolExecutor是线程池最底层的实现,需要知道几个参数的含义,还有线程work的工作流程。