多线程笔记（三）线程池 ThreadPoolExecutor

1. ctl：将 workerCount（工作线程的数量）和线程池的状态放在一个 int 中保存

在 ctolOf 方法中实现

private static int ctlOf(int rs, int wc) { return rs | wc; }

作用：并不是节省空间，而是为了保持线程池状态和运行状态的统一，放在一个变量里，可以保证这两部分同时修改
2. 五种状态，需要三个 bit 来表示，所以高 3 位表示状态，剩下的 29 位表示工作线程的数量，因此线程池中工作线程的最大数量为 $2^29−1$

也就是 ThreadPoolExecutor.CAPACITY（高三位为 0，低 29 位为 1）

private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

计算工作线程的数量：workerCountOf 方法。和 CAPACITY 做 & 运算，由于 CAPACITY 的低 29 位都为 1，所以可以取出工作线程的数量

private static int workerCountOf(int c)  { return c & CAPACITY; }

获取线程池状态：要取 ctl 的高三位的值，因此先把 CAPACITY 取反，然后和 ctl 做 & 运算

private static int runStateOf(int c)     { return c & ~CAPACITY; }

3. execute 方法

 1 public void execute(Runnable command) {
 2     if (command == null)
 3         throw new NullPointerException();
 4 
 5     // 获取当前线程池的状态
 6     int c = ctl.get();
 7     // 当工作线程数 < 核心线程池容量时，添加工作线程去执行任务
 8     if (workerCountOf(c) < corePoolSize) {
 9         if (addWorker(command, true))
10             return;
11         // 重新获取状态
12         c = ctl.get();
13     }
14     // 达到核心池上限，尝试加入缓存队列
15     if (isRunning(c) && workQueue.offer(command)) {
16         int recheck = ctl.get();
17         // 再次检查线程池状态
18         if (! isRunning(recheck) && remove(command))// 线程池没有在运行，将任务移出队列，并执行拒绝策略
19             reject(command);
20         else if (workerCountOf(recheck) == 0)// 线程池中没有线程，需要创建新的线程，传入 null 是因为创建的线程去队列里取任务，不需要指定执行的任务
21             addWorker(null, false);
22     }
23     else if (!addWorker(command, false)) // 缓存队列已满，尝试创建新的线程
24         reject(command);
25 }

4. addWorker 方法

 1 private boolean addWorker(Runnable firstTask, boolean core) {
 2     retry:
 3     for (;;) { // loop1
 4         int c = ctl.get();
 5         int rs = runStateOf(c);
 6 
 7         // Check if queue empty only if necessary.
 8         // 此处 if 判断为 False 只有两种情况：rs 为 running（也就是 < shutdown），当然可以继续添加线程；还有就是虽然线程池已经 shutdown，但是不会添加新的任务（firstTask == null），同时队列里还有任务，也就是(rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())，其他情况都是 True，需要停止添加线程，返回 false
 9         if (rs >= SHUTDOWN &&
10             ! (rs == SHUTDOWN &&
11                firstTask == null &&
12                ! workQueue.isEmpty()))
13             return false;
14 
15         for (;;) {
16             int wc = workerCountOf(c);
17             if (wc >= CAPACITY ||
18                 wc >= (core ? corePoolSize : maximumPoolSize))
19                 return false;
20             // CAS 使 workCount 加 1，成功则退出 loop1，继续下面的代码
21             if (compareAndIncrementWorkerCount(c))
22                 break retry;
23             c = ctl.get();  // Re-read ctl
24             // CAS 操作失败，重新检查状态，如果线程池状态改变，进行外部循环（loop1）
25             if (runStateOf(c) != rs)
26                 continue retry;
27             // else CAS failed due to workerCount change; retry inner loop
28             // 线程池状态不变的话，继续内部循环，尝试使 workCount 加 1
29         }
30     }
31 
32     boolean workerStarted = false;
33     boolean workerAdded = false;
34     Worker w = null;
35     try {
36         // 创建新线程
37         w = new Worker(firstTask);
38         final Thread t = w.thread;
39         if (t != null) {
40             final ReentrantLock mainLock = this.mainLock;
41             mainLock.lock();
42             try {
43                 // Recheck while holding lock.
44                 // Back out on ThreadFactory failure or if
45                 // shut down before lock acquired.
46                 int rs = runStateOf(ctl.get());
47                 
48                 // 检查线程池状态
49                 if (rs < SHUTDOWN ||
50                     (rs == SHUTDOWN && firstTask == null)) {
51                     if (t.isAlive()) // precheck that t is startable
52                         throw new IllegalThreadStateException();
53                     workers.add(w);
54                     int s = workers.size();
55                     if (s > largestPoolSize)
56                         largestPoolSize = s;
57                     workerAdded = true;
58                 }
59             } finally {
60                 mainLock.unlock();
61             }
62             if (workerAdded) {
63                 // 启动线程
64                 t.start();
65                 workerStarted = true;
66             }
67         }
68     } finally {
69         if (! workerStarted)
70             addWorkerFailed(w);
71     }
72     return workerStarted;
73 }