详解(一)-ThreadPollExecutor-并发编程(Java)

1 前言

前面通过实现自定义线程池-并发编程(Java)及详解-ThreadPollExecutor-并发编程(Java)，我们对线程池有了初步了解。下面，我们深入JDK代码底层，更深入的学习

2 TheadPoolExcutor 基础

ThreadPoolExcutor就是JDK给我们提供的线程池实现类。继承关系如下图1-1所示：[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-1nqJF2BS-1668310655374)(L:\study\java\concurrent\thread-pool\20221111-threadpool-inheritance.png)]

2.1 状态

先看下源码关于状态的定义：

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// 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;
// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

• COUNT_BITS：值为int位数32减去3即29
• ctl：这个原子整数使用int的高3位来表示线程状态；低29位表示线程数，即线程最大容量$2^{29}-1$；
• 对于有对ctl的合成 与取状态线程数不明白的可以复习下按位运算。

线程的5种状态如下表2.1-1所示：

• 注：那么问题来了，为什么它要用一个原子整数存储2个值：状态和线程数，难道用2个变量不是更好操作吗？

我们这个是线程池，主要任务自然用多线程处理多个任务，不可避免涉及访问控制。使用一个原子整数，不用对2个变量进行原子操作；且ctl的合成和取状态和线程数不复杂。

2.2 状态转换

• 线程池状态转换如下表2.1-2所示

2 构造方法

2.1 ThreadPoolExcutor构造方法

构造方法有重载，我们下面展示一个最全的构造方法：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)
1
2
3
4
5
6
7

2.2 corePoolSize和maximumPoolSize

• corePoolSize：核心线程数
• maximumPoolSize：最大线程数

核心线程数和最大线程数作用，当有新任务时，如果当前线程数小于核心线程数，创建新的工作线程；大于等于核心线程数时，新任务放入阻塞队列；如果阻塞队列满了但是不超过最大线程数，创建非核心线程；如果超过最大线程数，执行拒绝策略。

非核心线程有的地方叫做救急线程，具体的代码和详细解析在下面线程池执行流程处讲解。

2.3 keepAliveTime和unit

• keepAliveTime：存活时间，指非核心线程或者设置了允许核心线程超时的核心线程
• unit：时间单位

看下面代码2.3-1:

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 {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}
private void decrementWorkerCount() {
    do {} while (! compareAndDecrementWorkerCount(ctl.get()));
}
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

上述代码为工作线程从阻塞队列获取任务执行的方法，执行流程图如下图2.3-1所示：

其中获取阻塞队列任务的时候，用到这两个变量。下面我们来分析下满足那些条件，会执行到这里。

• 第一个主要分支判断：rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())，我们需要的结果是false,满足false的条件为一下之一即条件关系是或的关系
  • 线程池状态是RUNNING
  • SHUTDOWN<=线程池状态
• 第二个主要分支判断：(wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())，我们需要的结果仍然是false,满足false的条件也是下面列出的条件之一
  • wc <= maximumPoolSize && (!timed || !timedOut) ：继续拆解，条件关系为且
    • wc <= maximumPoolSize：工作现场数小于等于最大的工作现场数
    • !timed || !timedOut：拆解，条件关系为或
      • !timed：timed = allowCoreThreadTimeOut || wc > corePoolSize，允许核心线程设置超时否且工作现场数线程数小于等于核心线程数
      • !timedOut：没有超时
• 第三个分支判断：timed
  • true：允许核心线程超时设置为真或者线程数大于核心线程数且小于等于最大工作现场数，此时执行从阻塞队列有时限的等待获取任务

1.3 阻塞队列 BlockingQueue

BlockingQueue阻塞队列主要功能就是当队列为空时，从队列取元素会阻塞；当队列满时，存入队列会阻塞。具体的实现，这里不详述，等以后有用到在讲解。

1.4 线程工厂ThreadFactory

线程工厂顾名思义就是使用工厂模式实现生成线程的，比起我们自己手动创建线程，它很好的封装了实现的细节。线程池代码中生成线程的地方，如下代码片段1.4-1：

Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}
1
2
3
4
5

即在工作线程构造方法中，具体的实现类同上不详述。

1.5 拒绝策略 RejectedExecutionHandler

拒绝策略在把任务放入阻塞队列失败或者添加工作线程失败时触发，定义了4种拒绝策略实现：

• AbortPolicy：抛异常,线程池默认拒绝策略

  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
      throw new RejectedExecutionException("Task " + r.toString() +
                                           " rejected from " +
                                           e.toString());
  }
  1
  2
  3
  4
  5

• DiscardPolicy：丢弃任务，什么也不做

  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  }
  1
  2

• DiscardOldestPolicy：丢弃最早放入阻塞队列的任务，执行当前任务

  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
              if (!e.isShutdown()) {
                  e.getQueue().poll();
                  e.execute(r);
              }
          }
  1
  2
  3
  4
  5
  6

• CallerRunsPolicy：任务自己执行

  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
              if (!e.isShutdown()) {
                  r.run();
              }
          }
  1
  2
  3
  4
  5

2.2 Excutors

我们常用的有3种类型的线程池，具体介绍可以查看这篇博客详解-ThreadPollExecutor-并发编程(Java)或者自行查阅相关文档。一般建议不直接new而是使用Excutors的静态方法来创建相应的线程池。

3 线程池执行任务流程

3.1 execute()

3.1.1 execute执行流程

讲解之前，先看下源码如下3.1-1所示：

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();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}
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

代码中也有很清晰的解释，不过是英文👊。这里简单划个的流程图，如下流程图3.1-1所示：

下面我们继续分析addWorker()方法：加入工作集合

3.1.2 addWorker()分析

方法addWorker()主要任务就是构建工作线程，加入工作线程集合，执行当前任务；如果失败直接返回false或者把构建的工作线程移除。

下上JDK源码如下所示：

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;
}
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

下面详细分析下执行流程：

• 首先带一个标志的双层for(;;)循环

  • 因为是双层for循环，如果内层循环想结束循环(不是结束方法),只能通过break或者continue循环标志。
  • 外层循环主要检测线程池状态
    • 如果状态>=SHUDOWN且(状态不是SHUTDOWN或者当前任务不为空或者阻塞队列为空)，直接返回false，结束方法、
      • 状态>=SHUTDOWN且不为SHUTDOWN，此时不在执行新任务也不去阻塞队列获取任务
  • 内层循环用于通过cas方法给工作线程计数加1
    • 如果线程数大于等于容量或者core为true核心线程false最大线程数，直接放回false（不满足用于创建执行任务的线程的条件了）
      • core当前要创建的工作线程是否为核心线程的标志
    • 条件满足，cas对工作线程数加1，成功直接跳出外循环；失败说明有竞争。
      • 重新获取原子整数，判断线程状态是否改变。没改变说明是线程数变化引起的竞争，重新开始内循环；改变了继续外循环。

• 执行完双层for循环之后，程序没结束，这里开始执行创建工作线程，加入工作线程集合以及开始执行任务的操作，当然过程不一定一帆风顺

  • 设置标志
    • workerStarted：工作线程是否以启动标志
    • workerAdded：工作行程是否加入工作行程结合标志
  • 不保证过程中不发生异常，try{}finally{}
    • 创建工作线程，获取线程工厂创建的线程，判断不为空，继续向下执行
      • 线程工厂创建线程，除非硬件故障或者服务崩溃等严重问题，一般情况下不会为空
    • 后续操作需要加锁，try{加锁}finally{释放锁}
      • 判断线程池状态如果是RUNNING或者(是SHUTDOWN且当前任务为空)
        • 线程池状态SHUTDOWN且当前任务为空的情况这里暂时不分析，后面遇到在讲解
      • 如果线程工厂刚创建的线程存活直接抛异常，可能性极小
      • 工作线程加入工作线程集合，且设置workerAdded为true
    • 判断workerAdded为真
      • 启动工作线程，设置workerStarted为true
    • finally模块判断如果workerStarted:false,执行addWorkerFailed()方法
  • 返回workerStarted

• 说明：

  • Worker工作线程类自AbstractQueuedSynchronizer，一种同步器，等后面学习之后再讲解，此处略过。
  • 关于该方法的流程图，有空的时候在画。
  • addWorkerFailed()方法流程相对清晰，不在详述。

5 关闭线程池

• shutdown()方法

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}
1
2
3
4
5
6
7
8
9
10
11
12
13

• shutdownNow()方法

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(STOP);
        interruptWorkers();
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

• 比较

  关闭线程池方法	状态设置	终止的工作线程类型	后续处理
  shutdown()	SHUTDOWN	打断空闲的工作线程	tryTerminate()
  shutdownNow()	STOP	启动的工作线程	tryTerminate()；且返回阻塞队列中的任务集合

• tryTerminate()方法：

  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
      }
  }
  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

  • 执行终止的状态需满足条件：RUNNING
  • 工作线程计数为0
    • 不为0中断空闲线程返回
  • cas尝试设置状态为TIDYING
    • 成功，执行terminated()
      • terminated()默认实现为空即啥也没干
      • 最终设置状态为TERMINATED,结束
    • 失败 循环方法

• tips：上面讲解了线程池一些主要方法及执行流程，其他知识盲点或者后续用到的时候在讲解。

  • submit()、invokeAll()、invokeAny()涉及Future相关知识，等后面我们会单独讲解。

6 后记

如有问题，欢迎交流讨论。

❓QQ:806797785

⭐️源代码仓库地址：https://gitee.com/gaogzhen/concurrent