Java线程池如何实现线程复用

古城微笑少年丶 2023-09-26 16:37 94阅读 0赞

线程池把线程和任务进行解耦，线程归线程，任务归任务，摆脱了通过 Thread 创建线程时“一个线程必须对应一个任务”的限制。在线程池中，同一个线程可以从 BlockingQueue 中不断提取新任务来执行，其核心原理在于线程池对 Thread 进行了封装，并不是每次执行任务都会调用 Thread.start() 来创建新线程，而是让每个线程去执行一个“循环任务”，在这个“循环任务”中，不停地检查是否还有任务等待被执行，如果有则直接去执行这个任务，也就是调用任务的 run 方法，把 run 方法当作和普通方法一样的地位去调用，相当于把每个任务的 run() 方法串联了起来，所以线程数量并不增加。

如图是线程池的工作流程图：

![a7cfc5312b96a7b7967e121be9eea922.png][]

我们可以查看线程池实现的execute()方法源码来进行分析：

public void execute(Runnable command) {   //对Runnable判空    if (command == null)         throw new NullPointerException();  //获取ctl的value值    int c = ctl.get();  //判断线程数量和核心线程数    if (workerCountOf(c) < corePoolSize) {     //为true以核心线程数为限制添加线程        if (addWorker(command, true))             Return;    //更新c的值        c = ctl.get();    }  //检查是否为running并提交到任务队列    if (isRunning(c) && workQueue.offer(command)) {
            //更新ctl的值        int recheck = ctl.get();    //检查线程池状态        if (! isRunning(recheck) && remove(command))       reject(command);    //检查是否有线程可用        else if (workerCountOf(recheck) == 0)             addWorker(null, false);    }   //为false以最大线程数为限制添加线程    else if (!addWorker(command, false))         reject(command);}

对照着代码、注释可以看出，完全实现了线程池的工作流程，我们再来分步进行完整的分析与解读。（可以在idea中和ThreadPoolExecutor源码配合一起阅读以下内容）

在ThreadPoolExecutor源码中，定义了很多“魔法值”，以下三个方法都是根据“魔法值”的含义来实现的。

int c = ctl.get();workerCountOf(c);isRunning(c);

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//-1000 ... ...0000private static final int COUNT_BITS = Integer.SIZE - 3;      // 29private static final int COUNT_MASK = (1 << COUNT_BITS) - 1; // 00011111 ... ... 11111111
    // 状态在高位存储private static final int RUNNING    = -1 << COUNT_BITS;      // 11100000 ... ... 00000000private static final int SHUTDOWN   =  0 << COUNT_BITS;      // 00000000 ... ... 00000000private static final int STOP       =  1 << COUNT_BITS;      // 00100000 ... ... 00000000private static final int TIDYING    =  2 << COUNT_BITS;      // 01000000 ... ... 00000000private static final int TERMINATED =  3 << COUNT_BITS;      // 01100000 ... ... 00000000
    private static int ctlOf(int rs, int wc) { return rs | wc; }
    private static int workerCountOf(int c) {     return c & CAPACITY; }
    private static boolean isRunning(int c) {
           return c < SHUTDOWN;}

来看下ctl的定义：ctl 是一个 AtomicInteger 的类，就是让保存的 int 变量的更新都是原子操作，保证线程安全。ctlOf 方法就是组合运行状态和工作线程数量。可以看到，ctlOf 方法是通过按位或的方式来实现的。这里把一个 int 变量拆成两部分来用。前面3位用来表示状态，后面29位用来表示工作线程数量。所以，工作线程数量最大不能超过 2^29-1 ，ThreadPoolExecutor 的设计者也是考虑不太可能超过这个数，暂时就用了29位。

workerCountOf方法：CAPACITY的值为00011111 ... ... 11111111，按位与之后便是工作线程数量。

isRunning方法：初始状态，ctl 的值是11100000 ... … 00000000，表示 RUNNING 状态，和0个工作线程。后面，每创建一个新线程，都把 ctl 加一。在 RUNNING 状态下，ctl 始终是负值，而 SHUTDOWN 是0，所以可以通过直接比较 ctl 的值来确定状态。

思考一下为什么要这样设计？

使用位运算可以给我们节省很多空间，但是这里就算使用两个值分开表示状态和工作线程数量，也才8个字节，显然不是这个原因。我了解到的可能是为了保证在并发情况下，保证运行状态和工作线程数量的一致性，把两个值放在一个int里面，然后用原子类进行存储和读写，始终可以保持一致性。

搞清楚这些之后，我们来分步看下execute方法。

①其中首先对Runnable进行判空

if (command == null)         throw new NullPointerException();

②接下来判断当前线程数是否小于核心线程数，如果小于核心线程数就调用 addWorker() 方法增加一个 Worker，这里的 Worker 就可以理解为一个线程

if (workerCountOf(c) < corePoolSize) {         if (addWorker(command, true))             return;        c = ctl.get();//这里更新工作线程数量c的值    }

addWorker中的第二个参数如果为true，代表增加线程时判断当前线程是否小于corePoolSize，小于则增加新线程，大于则不增加；为false则代表是否小于maxPoolSize，同理。如果添加线程成功返回true，失败返回false。

③接着往下看，能运行到这个if的话说明线程数大于或等于核心线程数或者addwork失败

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

那么就需要通过 if (isRunning(c) && workQueue.offer(command)) 检查线程池状态是否为 Running，如果线程池状态是 Running 就把任务放入任务队列中，也就是 workQueue.offer(command)。如果任务可以成功排队，仍然需要仔细检查我们是否应该添加一个线程（因为现有线程可能自上次检查以来已死亡），或者在进入此方法后，线程池可能关闭，需要执行拒绝策略。因此，我们需要重新检查状态。如果线程池已经不处于 Running 状态，说明线程池被关闭，那么就移除刚刚添加到任务队列中的任务，并执行拒绝策略；如果进入 else if 说明前面判断到线程池状态为 Running，那么当任务被添加进来之后就需要防止没有可执行线程的情况发生（比如之前的线程被回收了或意外终止了），所以此时如果检查当前线程数为 0，也就是 workerCountOf(recheck) == 0，那就执行 addWorker() 方法新建线程。

④看最后一个else if

else if (!addWorker(command, false))         reject(command);

执行到这里，说明线程池不是 Running 状态或线程数大于或等于核心线程数并且任务队列已经满了，根据规则，此时需要添加新线程，直到线程数达到“最大线程数”。所以addWork第二个参数传入false以 maxPoolSize 为上限创建新的 worker；addWorker 方法如果返回 true 代表添加成功，如果返回 false 代表任务添加失败，说明当前线程数已经达到 maxPoolSize，然后执行拒绝策略 reject 方法。如果执行到这里线程池的状态不是 Running，那么addWorker 会失败并返回 false，所以也会执行拒绝策略 reject 方法。

由以上代码分析得到，在 execute 方法中，多次调用 addWorker 方法把任务传入，addWorker 方法会添加并启动一个 Worker，这里的 Worker 可以理解为是对 Thread 的包装，Worker 内部有一个 Thread 对象，是最终真正执行任务的线程，所以一个 Worker 就对应线程池中的一个线程，addWorker 就代表增加线程。线程复用的逻辑实现主要在 Worker 类中的 run 方法里执行的 runWorker 方法中，简化后的 runWorker 方法代码如下所示。

runWorker(Worker w) {
           Runnable task = w.firstTask;    while (task != null || (task = getTask()) != null) {
               try {
                   task.run();        } finally {
                   task = null;        }    }}

可以看出，实现线程复用的逻辑主要在一个不停循环的 while 循环体中。通过取 Worker 的 firstTask 或者通过 getTask 方法从 workQueue 中获取待执行的任务。直接调用 task 的 run 方法来执行具体的任务（而不是新建线程）。在这里，我们找到了最终的实现，通过取 Worker 的 firstTask 或者 getTask方法从 workQueue 中取出了新任务，并直接调用 Runnable 的 run 方法来执行任务，也就是如之前所说的，每个线程都始终在一个大循环中，反复获取任务，然后执行任务，从而实现了线程的复用。

[a7cfc5312b96a7b7967e121be9eea922.png]: https://img-blog.csdnimg.cn/img_convert/a7cfc5312b96a7b7967e121be9eea922.png