深入理解高并发编程 - 通过 ThreadPoolExecutor 类深度解析线程池执行任务

骑猪看日落 2023-10-14 19:56 25阅读 0赞

## 1、核心逻辑 ##

线程池状态管理： ThreadPoolExecutor 使用整数变量 ctl 来表示线程池的状态和工作线程数量。不同状态的定义包括 RUNNING、SHUTDOWN、STOP、TIDYING 和 TERMINATED。通过位运算和原子操作，可以控制状态的转换和工作线程数量的变化。
    
    任务提交和队列管理： 任务通过 execute(Runnable command) 方法提交给线程池。线程池会根据当前状态和工作线程数量来决定是创建新的工作线程执行任务，还是将任务加入工作队列。工作队列（workQueue）可以是不同类型的阻塞队列，用于存储等待执行的任务。
    
    工作线程管理和执行： ThreadPoolExecutor 会维护一组工作线程（Worker 对象），每个工作线程负责从工作队列中获取任务并执行。工作线程的主循环在 runWorker(Worker w) 方法中实现，其中涉及任务的获取、执行、线程状态的控制等。
    
    任务执行前后钩子方法： 在工作线程执行任务之前和之后，可以通过 beforeExecute(Thread t, Runnable r) 和 afterExecute(Runnable r, Throwable t) 方法插入自定义逻辑，例如任务执行统计、日志记录等。
    
    工作线程的终止和状态转换： 工作线程在完成任务后，会进入等待状态，直到新任务到来。线程池会根据需要自动管理工作线程的数量，包括创建和销毁工作线程，以及处理工作线程的异常终止。
    
    任务拒绝策略： 当线程池和工作队列都已满，无法继续接受新任务提交时，会根据预定义的拒绝策略来处理无法执行的任务，如抛出异常、丢弃任务等。
    
    并发控制和线程安全： ThreadPoolExecutor 使用锁、原子操作等机制来保证在多线程环境下的并发控制和线程安全。

## 2、execute(Runnable)方法 ##

execute(Runnable command) 方法是 ThreadPoolExecutor 类中的核心方法之一，用于将一个任务提交给线程池执行。下面是对该方法的源码详细解析：

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

这段源码实现了任务的提交逻辑，主要分为以下几个步骤：

参数检查： 首先，会检查传入的任务对象 command 是否为 null，如果为 null 则抛出 NullPointerException。
    
    核心线程数控制： 通过 ctl 变量获取当前线程池状态和工作线程数量。如果当前工作线程数量小于核心线程数（corePoolSize），则尝试创建一个新的工作线程来执行任务。这是为了确保在任务提交时，始终有足够的核心线程来处理任务，从而减少任务排队和执行的延迟。
    
    任务入队： 如果当前工作线程数量达到核心线程数或创建核心线程的尝试失败，线程池会将任务放入工作队列（workQueue）中，等待工作线程来执行。这个步骤的目标是将任务放入队列，以便工作线程可以从队列中获取任务进行执行。
    
    任务拒绝策略： 如果任务无法放入工作队列，说明工作队列已满，此时线程池会尝试创建一个新的工作线程来执行任务（如果线程数未达到最大线程数）。如果创建工作线程失败，线程池会调用拒绝策略（通过 reject(Runnable command) 方法），处理无法执行的任务，例如抛出异常或记录日志。
    
    状态和工作线程数量检查： 在任务入队或创建工作线程后，线程池会再次检查状态和工作线程数量。如果线程池状态变为非运行状态，或者工作线程数量为零，可能需要进行一些额外的操作，例如移除任务或添加工作线程。

总的来说，execute(Runnable command) 方法负责将任务提交给线程池执行，根据不同情况选择创建工作线程、将任务放入工作队列或处理无法执行的任务。这个方法的实现确保了任务在合适的条件下得到执行，并且在线程池管理下实现了任务的高效调度和执行。

## 3、addWorker(Runnable, boolean)方法 ##

addWorker(Runnable firstTask, boolean core) 方法是 ThreadPoolExecutor 类中的一个关键方法，用于向线程池中添加工作线程。下面是对该方法的源码解析：

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
    
            // 检查线程池状态，如果不是 RUNNING 则不再添加工作线程
            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();  // 重新读取 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 {
                    // 重新检查线程池状态
                    int rs = runStateOf(ctl.get());
    
                    // 只有在 RUNNING 或 SHUTDOWN 状态下才能添加工作线程
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // 检查线程是否已经启动
                            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;
    }

该方法的主要逻辑如下：

状态检查： 首先，检查线程池的状态（runState）是否允许添加工作线程。如果线程池状态不是 RUNNING，那么不再添加工作线程。特殊情况是线程池处于 SHUTDOWN 状态，但工作队列非空，这时允许添加工作线程。
    
    工作线程数量控制： 检查工作线程数量是否已经达到最大线程数。如果超过了最大线程数，则不再添加工作线程。
    
    增加工作线程数量： 通过 CAS 操作增加工作线程数量。如果成功，跳出循环；如果失败，则重新读取线程池状态并继续循环。
    
    创建工作线程： 在成功增加工作线程数量后，创建一个新的工作线程（Worker 对象）。如果工作线程对象创建成功，则将该工作线程添加到线程池管理的集合中（workers）。
    
    启动工作线程： 如果工作线程对象创建并添加成功，则启动工作线程。工作线程会执行 runWorker(Worker w) 方法，从工作队列中获取任务并执行。
    
    异常处理： 在整个过程中，如果发生异常（如线程启动失败），会回滚工作线程数量的增加，确保工作线程数量和状态一致。

总之，addWorker(Runnable firstTask, boolean core) 方法负责创建和添加工作线程，确保线程池的工作线程数量满足要求，并在适当的情况下启动工作线程来执行任务。这个方法的逻辑关键在于通过 CAS 操作增加工作线程数量，以及在多线程环境下对线程池状态和工作线程数量的合理管理。

## 4、addWorkerFailed(Worker)方法 ##

在 ThreadPoolExecutor 类的源码中，addWorkerFailed(Worker w) 方法是一个私有方法，用于在添加工作线程失败时回滚工作线程数量的增加。下面是该方法的源码解析：

private void addWorkerFailed(Worker w) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (w != null)
                workers.remove(w);
            decrementWorkerCount();
            tryTerminate(); // 检查是否需要终止线程池
        } finally {
            mainLock.unlock();
        }
    }

该方法的主要逻辑如下：

移除工作线程： 首先，方法会尝试将指定的工作线程 w 从线程池管理的集合中移除（workers 集合）。这是为了确保工作线程数量和状态的一致性。
    
    减少工作线程数量： 在成功移除工作线程后，会通过调用 decrementWorkerCount() 方法来减少工作线程数量。该方法使用 CAS 操作来降低 ctl 变量中的工作线程数量。
    
    终止线程池检查： 最后，会调用 tryTerminate() 方法来检查是否需要终止线程池。该方法会判断线程池是否满足终止条件，例如线程池状态为 SHUTDOWN 并且工作线程数量为零。如果满足终止条件，会触发线程池的终止操作。

总的来说，addWorkerFailed(Worker w) 方法用于处理添加工作线程失败的情况，回滚工作线程数量的增加，移除失败的工作线程，以及可能触发线程池的终止操作。这样可以保持线程池的一致性和状态管理。

## 5、reject(Runnable)方法 ##

在 ThreadPoolExecutor 类的源码中，reject(Runnable command) 方法用于处理无法执行的任务，即任务提交后无法被线程池接受的情况。下面是该方法的源码解析：

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

这里的 handler 是一个 RejectedExecutionHandler 接口的实例，用于定义当任务无法被线程池接受时的处理逻辑。RejectedExecutionHandler 接口有一个方法 rejectedExecution(Runnable r, ThreadPoolExecutor executor)，在任务被拒绝时被调用，需要用户自行实现。

在 ThreadPoolExecutor 类的构造函数中，可以通过传递一个 RejectedExecutionHandler 实例来指定任务拒绝的处理策略。例如，如果用户未指定任何拒绝策略，默认的 RejectedExecutionHandler 是 AbortPolicy，它会抛出一个 RejectedExecutionException 异常。

以下是一个示例的 RejectedExecutionHandler 实现：

public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            // 自定义任务拒绝处理逻辑，例如记录日志或重试
            System.out.println("Task rejected: " + r.toString());
        }
    }

然后，可以在创建 ThreadPoolExecutor 时传递该拒绝策略实例：

RejectedExecutionHandler handler = new CustomRejectedExecutionHandler();
    ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.SECONDS, workQueue, handler);

这样，当任务被拒绝时，会调用 CustomRejectedExecutionHandler 中的 rejectedExecution 方法来处理，用户可以根据实际需求进行日志记录、任务重试等操作。

## 6、总结 ##

在 ThreadPoolExecutor 类的源码中，我们深入了解了以下几个核心方法和相关概念：

execute(Runnable command): 该方法用于将任务提交给线程池执行。它会根据线程池状态、工作线程数量和工作队列状态来决定任务的执行方式，可以创建新的工作线程执行任务，将任务放入工作队列，或者根据拒绝策略来处理无法执行的任务。
    
    addWorker(Runnable firstTask, boolean core): 这个方法在工作线程不足时尝试添加一个新的工作线程。它会考虑线程池状态、工作线程数量、核心线程数和最大线程数等因素，以决定是否创建新的工作线程来执行任务。
    
    addWorkerFailed(Worker w): 当添加工作线程失败时，这个方法会回滚工作线程数量的增加。它会从线程池管理的工作线程集合中移除失败的工作线程，并递减工作线程数量，同时检查是否需要终止线程池。
    
    reject(Runnable command): 这个方法用于处理无法执行的任务，即任务提交后无法被线程池接受的情况。它会委托给一个 RejectedExecutionHandler 实例来处理任务的拒绝，用户可以自定义拒绝策略。

这些方法共同构成了 ThreadPoolExecutor 类的核心逻辑，实现了任务的提交、工作线程的管理、任务的执行、线程池状态的控制等功能。通过合理配置线程池参数和拒绝策略，可以在多线程环境下实现任务的高效执行和线程资源的合理管理。