16.线程池之线程复用原理与源码解析

忘是亡心i 2023-10-08 19:44 138阅读 0赞

## 线程复用原理 ##

线程池可以把线程和任务进行解耦，线程归线程，任务归任务，摆脱了之前通过 Thread 创建线程时的一个线程必须对应一个任务的限制。

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

所以实现线程复用的逻辑主要在一个不停循环的 while 循环体中。

## 线程复用源码解析 ##

**地址：java.util.concurrent.ThreadPoolExecutor\#execute**  
**我们从 execute 方法开始分析，源码：**

public void execute(Runnable command) {
         
        if (command == null) 
            throw new NullPointerException();
        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);
    }

**开始拆解分析：**

execute 方法中通过 if 语句判断 command ，也就是 Runnable 任务是否等于 null，如果为 null 就抛出异常。

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

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

if (workerCountOf(c) < corePoolSize) {
         
        if (addWorker(command, true)) 
            return;
            c = ctl.get();
    }

**---------addWorker方法作用扩展：**

*  addWorker 方法的主要作用是在线程池中创建一个线程并执行第一个参数传入的任务，它的第二个参数是个布尔值，如果布尔值传入 true 代表增加线程时判断当前线程是否少于 corePoolSize，小于则增加新线程，大于等于则不增加；
 *  如果传入 false 代表增加线程时判断当前线程是否少于 maxPoolSize，小于则增加新线程，大于等于则不增加，所以这里的布尔值的含义是以核心线程数为界限还是以最大线程数为界限进行是否新增线程的判断。
 *  addWorker() 方法如果返回 true 代表添加成功，如果返回 false 代表添加失败。

**---------addWorker作用扩展结束**

继续

代码执行到这里，说明当前线程数大于或等于核心线程数或者 addWorker 失败了，那么就需要通过 if (isRunning© && workQueue.offer(command)) 检查线程池状态是否为 Running。

如果线程池状态是 Running 就把任务放入任务队列中，也就是 workQueue.offer(command)。

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

如果线程池已经不处于 Running 状态，说明线程池被关闭，那么就移除刚刚添加到任务队列中的任务，并执行拒绝策略，如下：

if (! isRunning(recheck) && remove(command)) 
        reject(command);

再来看后一个 else 分支：  
能进入这个 else 说明前面判断到线程池状态为 Running，那么当任务被添加进来之后就需要防止没有可执行线程的情况发生（比如之前的线程被回收了或意外终止了），所以此时如果检查当前线程数为 0，也就是 workerCountOf(recheck) == 0，那就执行 addWorker() 方法新建线程。

else if (workerCountOf(recheck) == 0) 
        addWorker(null, false);

执行下面的判断，说明线程池不是 Running 状态或线程数大于或等于核心线程数并且任务队列已经满了，根据规则，此时需要添加新线程，直到线程数达到“最大线程数”，所以此时就会再次调用 addWorker 方法并将第二个参数传入 false，传入 false 代表增加线程时判断当前线程数是否少于 maxPoolSize，小于则增加新线程，大于等于则不增加，也就是以 maxPoolSize 为上限创建新的 worker；

addWorker 方法如果返回 true 代表添加成功，如果返回 false 代表任务添加失败，说明当前线程数已经达到 maxPoolSize，然后执行拒绝策略 reject 方法。

如果执行到这里线程池的状态不是 Running，那么 addWorker 会失败并返回 false，所以也会执行拒绝策略 reject 方法。

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

**从上面的分析中可以看出，在 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 方法来执行任务，也就是如之前所说的，每个线程都始终在一个大循环中，反复获取任务，然后执行任务，从而实现了线程的复用。**

### 线程回收： ###

这里在扩展下runWorker(Worker w)方法中的 processWorkerExit(w, completedAbruptly);

**线程池线程回收机制就是通过processWorkerExit这样维持线程池核心线程数量 。**

processWorkerExit()方法是为将要终结的Worker做一次清理和数据记录工作，因为processWorkerExit()方法也包裹在runWorker()方法finally代码块中，其实工作线程在执行完processWorkerExit()方法才算真正的终结。

这段代码基本上之前上面我都分析过了，直接看一下就清楚干什么了

private void processWorkerExit(Worker w, boolean completedAbruptly) {
        
            if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
                decrementWorkerCount();
    
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
        
                completedTaskCount += w.completedTasks;
                workers.remove(w);
            } finally {
        
                mainLock.unlock();
            }
    
            tryTerminate();
    
            int c = ctl.get();
            if (runStateLessThan(c, STOP)) {
        
                if (!completedAbruptly) {
        
                    int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                    if (min == 0 && ! workQueue.isEmpty())
                        min = 1;
                    if (workerCountOf(c) >= min)
                        return; // replacement not needed
                }
                addWorker(null, false);
            }
        }