乐趣区

关于多线程:深入分析3种线程池执行任务的逻辑方法

jiezi

摘要： 联合 ThreadPoolExecutor 类的源码深度剖析线程池执行工作的整体流程。

本文分享自华为云社区《【高并发】通过 ThreadPoolExecutor 类的源码深度解析线程池执行工作的外围流程》，作者：冰 河。

ThreadPoolExecutor 类中存在一个 workers 工作线程汇合，用户能够向线程池中增加须要执行的工作，workers 汇合中的工作线程能够间接执行工作，或者从工作队列中获取工作后执行。ThreadPoolExecutor 类中提供了整个线程池从创立到执行工作，再到沦亡的整个流程办法。本文，就联合 ThreadPoolExecutor 类的源码深度剖析线程池执行工作的整体流程。

在 ThreadPoolExecutor 类中，线程池的逻辑次要体现在 execute(Runnable) 办法，addWorker(Runnable, boolean) 办法，addWorkerFailed(Worker) 办法和回绝策略上，接下来，咱们就深入分析这几个外围办法。

execute(Runnable) 办法

execute(Runnable) 办法的作用是提交 Runnable 类型的工作到线程池中。咱们先看下 execute(Runnable) 办法的源码，如下所示。

public void execute(Runnable command) {
    // 如果提交的工作为空，则抛出空指针异样
    if (command == null)
        throw new NullPointerException();
    // 获取线程池的状态和线程池中线程的数量
    int c = ctl.get();
    // 线程池中的线程数量小于 corePoolSize 的值
    if (workerCountOf(c) < corePoolSize) {
        // 从新开启线程执行工作
        if (addWorker(command, true))
            return;
        c = ctl.get();}
    // 如果线程池处于 RUNNING 状态，则将工作增加到阻塞队列中
    if (isRunning(c) && workQueue.offer(command)) {
        // 再次获取线程池的状态和线程池中线程的数量，用于二次查看
        int recheck = ctl.get();
        // 如果线程池没有未处于 RUNNING 状态，从队列中删除工作
        if (! isRunning(recheck) && remove(command))
            // 执行回绝策略
            reject(command);
        // 如果线程池为空，则向线程池中增加一个线程
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 工作队列已满，则新增 worker 线程，如果新增线程失败，则执行回绝策略
    else if (!addWorker(command, false))
        reject(command);
}

整个工作的执行流程，咱们能够简化成下图所示。

接下来，咱们拆解 execute(Runnable) 办法，具体分析 execute(Runnable) 办法的执行逻辑。

（1）线程池中的线程数是否小于 corePoolSize 外围线程数，如果小于 corePoolSize 外围线程数，则向 workers 工作线程汇合中增加一个外围线程执行工作。代码如下所示。

// 线程池中的线程数量小于 corePoolSize 的值
if (workerCountOf(c) < corePoolSize) {
    // 从新开启线程执行工作
    if (addWorker(command, true))
        return;
    c = ctl.get();}

（2）如果线程池中的线程数量大于 corePoolSize 外围线程数，则判断以后线程池是否处于 RUNNING 状态，如果处于 RUNNING 状态，则增加工作到待执行的工作队列中。留神：这里向工作队列增加工作时，须要判断线程池是否处于 RUNNING 状态，只有线程池处于 RUNNING 状态时，能力向工作队列增加新工作。否则，会执行回绝策略。代码如下所示。

if (isRunning(c) && workQueue.offer(command))

（3）向工作队列中增加工作胜利，因为其余线程可能会批改线程池的状态，所以这里须要对线程池进行二次查看，如果以后线程池的状态不再是 RUNNING 状态，则须要将增加的工作从工作队列中移除，执行后续的回绝策略。如果以后线程池依然处于 RUNNING 状态，则判断线程池是否为空，如果线程池中不存在任何线程，则新建一个线程增加到线程池中，如下所示。

// 再次获取线程池的状态和线程池中线程的数量，用于二次查看
int recheck = ctl.get();
// 如果线程池没有未处于 RUNNING 状态，从队列中删除工作
if (! isRunning(recheck) && remove(command))
    // 执行回绝策略
    reject(command);
// 如果线程池为空，则向线程池中增加一个线程
else if (workerCountOf(recheck) == 0)
    addWorker(null, false);

（4）如果在步骤（3）中向工作队列中增加工作失败，则尝试开启新的线程执行工作。此时，如果线程池中的线程数量曾经大于线程池中的最大线程数 maximumPoolSize，则不能再启动新线程。此时，示意线程池中的工作队列已满，并且线程池中的线程已满，须要执行回绝策略，代码如下所示。

// 工作队列已满，则新增 worker 线程，如果新增线程失败，则执行回绝策略
else if (!addWorker(command, false))
    reject(command);

这里，咱们将 execute(Runnable) 办法拆解，联合流程图来了解线程池中工作的执行流程就比较简单了。能够这么说，execute(Runnable) 办法的逻辑基本上就是个别线程池的执行逻辑，了解了 execute(Runnable) 办法，就根本了解了线程池的执行逻辑。

留神：无关 ScheduledThreadPoolExecutor 类和 ForkJoinPool 类执行线程池的逻辑，在【高并发专题】系列文章中的后文中会具体阐明，了解了这些类的执行逻辑，就根本全面把握了线程池的执行流程。

在剖析 execute(Runnable) 办法的源码时，咱们发现 execute(Runnable) 办法中多处调用了 addWorker(Runnable, boolean) 办法，接下来，咱们就一起剖析下 addWorker(Runnable, boolean) 办法的逻辑。

addWorker(Runnable, boolean) 办法

总体上，addWorker(Runnable, boolean) 办法能够分为三局部，第一局部是应用 CAS 平安的向线程池中增加工作线程；第二局部是创立新的工作线程；第三局部则是将工作通过平安的并发形式增加到 workers 中，并启动工作线程执行工作。

接下来，咱们看下 addWorker(Runnable, boolean) 办法的源码，如下所示。

private boolean addWorker(Runnable firstTask, boolean core) {
    // 标记重试的标识
    retry:
    for (;;) {int c = ctl.get();
        int rs = runStateOf(c);

        // 查看队列是否在某些特定的条件下为空
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;
        // 上面循环的次要作用为通过 CAS 形式减少线程的个数
        for (;;) {
            // 获取线程池中的线程数量
            int wc = workerCountOf(c);
            // 如果线程池中的线程数量超出限度，间接返回 false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 通过 CAS 形式向线程池新增线程数量
            if (compareAndIncrementWorkerCount(c))
                // 通过 CAS 形式保障只有一个线程执行胜利，跳出最外层循环
                break retry;
            // 从新获取 ctl 的值
            c = ctl.get();  
            // 如果 CAS 操作失败了，则须要在内循环中从新尝试通过 CAS 新增线程数量
            if (runStateOf(c) != rs)
                continue retry;
        }
    }
 
    // 跳出最外层 for 循环，阐明通过 CAS 新增线程数量胜利
    // 此时创立新的工作线程
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 将执行的工作封装成 worker
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            // 独占锁，保障操作 workers 时的同步
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 此处须要从新查看线程池状态
                // 起因是在取得锁之前可能其余的线程扭转了线程池的状态
                int rs = runStateOf(ctl.get());
 
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {if (t.isAlive())
                        throw new IllegalThreadStateException();
                    // 向 worker 中增加新工作
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    // 将是否增加了新工作的标识设置为 true
                    workerAdded = true;
                }
            } finally {
                // 开释独占锁
                mainLock.unlock();}
            // 增加新任胜利，则启动线程执行工作
            if (workerAdded) {t.start();
                // 将工作是否曾经启动的标识设置为 true
                workerStarted = true;
            }
        }
    } finally {// 如果工作未启动或启动失败，则调用 addWorkerFailed(Worker) 办法
        if (! workerStarted)
            addWorkerFailed(w);
    }
    // 返回是否启动工作的标识
    return workerStarted;
}

乍一看，addWorker(Runnable, boolean) 办法还蛮长的，这里，咱们还是将 addWorker(Runnable, boolean) 办法进行拆解。

（1）查看工作队列是否在某些特定的条件下为空，代码如下所示。

// 查看队列是否在某些特定的条件下为空
if (rs >= SHUTDOWN &&
    ! (rs == SHUTDOWN &&
       firstTask == null &&
       ! workQueue.isEmpty()))
    return false;

（2）在通过步骤（1）的校验后，则进入内层 for 循环，在内层 for 循环中通过 CAS 来减少线程池中的线程数量，如果 CAS 操作胜利，则间接退出双重 for 循环。如果 CAS 操作失败，则查看以后线程池的状态是否产生了变动，如果线程池的状态产生了变动，则通过 continue 关键字从新通过外层 for 循环校验工作队列，测验通过再次执行内层 for 循环的 CAS 操作。如果线程池的状态没有发生变化，此时上一次 CAS 操作失败了，则持续尝试 CAS 操作。代码如下所示。

for (;;) {
    // 获取线程池中的线程数量
    int wc = workerCountOf(c);
    // 如果线程池中的线程数量超出限度，间接返回 false
    if (wc >= CAPACITY ||
        wc >= (core ? corePoolSize : maximumPoolSize))
        return false;
    // 通过 CAS 形式向线程池新增线程数量
    if (compareAndIncrementWorkerCount(c))
        // 通过 CAS 形式保障只有一个线程执行胜利，跳出最外层循环
        break retry;
    // 从新获取 ctl 的值
    c = ctl.get();  
    // 如果 CAS 操作失败了，则须要在内循环中从新尝试通过 CAS 新增线程数量
    if (runStateOf(c) != rs)
        continue retry;
}

（3）CAS 操作胜利后，示意向线程池中胜利增加了工作线程，此时，还没有线程去执行工作。应用全局的独占锁 mainLock 来将新增的工作线程 Worker 对象平安的增加到 workers 中。

总体逻辑就是：创立新的 Worker 对象，并获取 Worker 对象中的执行线程，如果线程不为空，则获取独占锁，获取锁胜利后，再次查看线线程的状态，这是防止在获取独占锁之前其余线程批改了线程池的状态，或者敞开了线程池。如果线程池敞开，则须要开释锁。否则将新减少的线程增加到工作汇合中，开释锁并启动线程执行工作。将是否启动线程的标识设置为 true。最初，判断线程是否启动，如果没有启动，则调用 addWorkerFailed(Worker) 办法。最终返回线程是否起送的标识。

// 跳出最外层 for 循环，阐明通过 CAS 新增线程数量胜利
// 此时创立新的工作线程
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
    // 将执行的工作封装成 worker
    w = new Worker(firstTask);
    final Thread t = w.thread;
    if (t != null) {
        // 独占锁，保障操作 workers 时的同步
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 此处须要从新查看线程池状态
            // 起因是在取得锁之前可能其余的线程扭转了线程池的状态
            int rs = runStateOf(ctl.get());
 
            if (rs < SHUTDOWN ||
                (rs == SHUTDOWN && firstTask == null)) {if (t.isAlive())
                    throw new IllegalThreadStateException();
                // 向 worker 中增加新工作
                workers.add(w);
                int s = workers.size();
                if (s > largestPoolSize)
                    largestPoolSize = s;
                // 将是否增加了新工作的标识设置为 true
                workerAdded = true;
            }
        } finally {
            // 开释独占锁
            mainLock.unlock();}
        // 增加新任胜利，则启动线程执行工作
        if (workerAdded) {t.start();
            // 将工作是否曾经启动的标识设置为 true
            workerStarted = true;
        }
    }
} finally {// 如果工作未启动或启动失败，则调用 addWorkerFailed(Worker) 办法
    if (! workerStarted)
        addWorkerFailed(w);
}
// 返回是否启动工作的标识
return workerStarted;

addWorkerFailed(Worker) 办法

在 addWorker(Runnable, boolean) 办法中，如果增加工作线程失败或者工作线程启动失败时，则会调用 addWorkerFailed(Worker) 办法，上面咱们就来看看 addWorkerFailed(Worker) 办法的实现，如下所示。

private void addWorkerFailed(Worker w) {
    // 获取独占锁
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 如果 Worker 工作不为空
        if (w != null)
            // 将工作从 workers 汇合中移除
            workers.remove(w);
        // 通过 CAS 将工作数量减 1
        decrementWorkerCount();
        tryTerminate();} finally {
        // 开释锁
        mainLock.unlock();}
}

addWorkerFailed(Worker) 办法的逻辑就比较简单了，获取独占锁，将工作从 workers 中移除，并且通过 CAS 将工作的数量减 1，最初开释锁。

回绝策略

咱们在剖析 execute(Runnable) 办法时，线程池会在适当的时候调用 reject(Runnable) 办法来执行相应的回绝策略，咱们看下 reject(Runnable) 办法的实现，如下所示。

final void reject(Runnable command) {handler.rejectedExecution(command, this);
}

通过代码，咱们发现调用的是 handler 的 rejectedExecution 办法，handler 又是个什么鬼，咱们持续跟进代码，如下所示。

private volatile RejectedExecutionHandler handler;

再看看 RejectedExecutionHandler 是个啥类型，如下所示。

package java.util.concurrent;

public interface RejectedExecutionHandler {void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

能够发现 RejectedExecutionHandler 是个接口，定义了一个 rejectedExecution(Runnable, ThreadPoolExecutor) 办法。既然 RejectedExecutionHandler 是个接口，那咱们就看看有哪些类实现了 RejectedExecutionHandler 接口。

看到这里，咱们发现 RejectedExecutionHandler 接口的实现类正是线程池默认提供的四种回绝策略的实现类。

至于 reject(Runnable) 办法中具体会执行哪个类的回绝策略，是依据创立线程池时传递的参数决定的。如果没有传递回绝策略，则默认会执行 AbortPolicy 类的回绝策略。否则会执行传递的类的回绝策略。

在创立线程池时，除了可能传递 JDK 默认提供的回绝策略外，还能够传递自定义的回绝策略。如果想应用自定义的回绝策略，则只须要实现 RejectedExecutionHandler 接口，并重写 rejectedExecution(Runnable, ThreadPoolExecutor) 办法即可。例如，上面的代码。

public class CustomPolicy implements RejectedExecutionHandler {public CustomPolicy() { }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {System.out.println("应用调用者所在的线程来执行工作")
            r.run();}
    }
}

应用如下形式创立线程池。

new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                       60L, TimeUnit.SECONDS,
                       new SynchronousQueue<Runnable>(),
                       Executors.defaultThreadFactory(),
               new CustomPolicy());

至此，线程池执行工作的整体外围逻辑剖析完结。

点击关注，第一工夫理解华为云陈腐技术~

退出移动版