聊聊 ThreadPoolExecutor 线程池及其源码

我就是我 2023-09-26 10:32 70阅读 0赞

## 什么是线程池 ##

在我们的实际开发中，通常使用线程池创建线程资源。

而线程池本质上是一种池化技术，利用资源复用的思想，复用线程资源。

## 为什么使用线程池 ##

那么，我们为什么要使用线程池呢？主要有以下两点原因：

*  首先，使用线程池可以避免频繁创建和销毁线程所带来的性能开销。因为线程的创建会涉及到 `CPU` 上下文的切换以及内存的分配。
 *  其次，我们可以通过线程池参数控制线程的数量，避免无休止的创建线程带来的资源利用率过高的问题，从而起到资源保护的作用。

## 线程池的状态 ##

![format_png][]

在我们使用线程池时，一共涉及 `5` 种状态，其中包括：

1.  `RUNNING` ：能接受新提交的任务，并且也能处理阻塞队列中的任务；
2.  `SHUTDOWN`：关闭状态，不再接受新提交的任务，但却可以继续处理阻塞队列中已保存的任务。在线程池处于 `RUNNING` 状态时，调用 `shutdown()` 方法会使线程池进入到该状态。（`finalize()` 方法在执行过程中也会调用 `shutdown()` 方法进入该状态）；
3.  `STOP`：不能接受新任务，也不处理队列中的任务，会中断正在处理任务的线程。在线程池处于 `RUNNING` 或 `SHUTDOWN` 状态时，调用 `shutdownNow()` 方法会使线程池进入到该状态；
4.  `TIDYING`：如果所有的任务都已终止了，`workerCount` (有效线程数) 为 `0`，线程池进入该状态后会调用 `terminated()` 方法进入 `TERMINATED` 状态。
5.  `TERMINATED`：在 `terminated()` 方法执行完后进入该状态，默认 `terminated()` 方法中什么也没有做。进入TERMINATED的条件如下：
    
     *  线程池不是 `RUNNING` 状态；
     *  线程池状态不是 `TIDYING` 状态或 `TERMINATED` 状态；
     *  如果线程池状态是 `SHUTDOWN` 并且 `workerQueue` 为空；
     *  `workerCount` 为 `0`；
     *  设置 `TIDYING` 状态成功。

## 线程池的分析及源码 ##

### 核心属性 ###

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    // COUNT_BITS = 32 - 3 = 29
    private static final int COUNT_BITS = Integer.SIZE - 3;
    // 工作线程的最大个数 
    // 00100000 00000000 00000000 00000000 - 1 
    // 000111111111111111111111111111111
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
    // runState is stored in the high-order bits
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;
    private static final int TIDYING    =  2 << COUNT_BITS;
    private static final int TERMINATED =  3 << COUNT_BITS;
    复制代码

其中 `ctl` 属性维护着线程池的两个重要内容：

1.  高 `3` 位维护着线程池的状态；
2.  低 `29` 位维护着工作线程个数；

### ctl 相关方法 ###

// Packing and unpacking ctl
    // 获取线程池状态
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    // 获取工作线程数
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    // 获取运行状态和工作线程数的值
    private static int ctlOf(int rs, int wc) { return rs | wc; }
    复制代码

### ThreadPoolExecutor 构造方法 ###

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
    复制代码

可以看出，在 `ThreadPoolExecutor` 的有参构造器中一共有 `6` 个参数：

*  `corePoolSize`：核心线程数
 *  `maximumPoolSize`：最大线程数
 *  `keepAliveTime`：线程池维护线程所允许的空闲时间，当线程池中的线程数量大于 `corePoolSize` 的时候，如果这时没有新的任务提交，核心线程外的线程不会立即销毁，而是会等待，直到等待的时间超过了 `keepAliveTime`
 *  `unit`：时间单位
 *  `workQueue`：等待队列，当任务提交时，如果线程池中的线程数量大于等于 `corePoolSize` 的时候，把该任务封装成一个 `Worker` 对象放入等待队列
 *  `threadFactory`：它是 `ThreadFactory` 类型的变量，线程工厂，用来创建新的线程
 *  `handler`：它是 `RejectedExecutionHandler` 类型的变量，表示线程池的饱和策略。如果阻塞队列满了并且没有空闲的线程，这时如果继续提交任务，就需要采取一种策略处理该任务。线程池提供了 `4` 种策略：
    
    1.  `AbortPolicy`：直接抛出异常，这是默认策略；
    2.  `CallerRunsPolicy`：用调用者所在的线程来执行任务；
    3.  `DiscardOldestPolicy`：丢弃阻塞队列中靠最前的任务，并执行当前任务；
    4.  `DiscardPolicy`：直接丢弃任务；

### 核心思想 ###

![format_png 1][]

当有新任务提交的时候，首先会执行以下判断：

1.  如果运行的线程少于 `corePoolSize`，则创建新线程来处理任务，即使线程池中的其他线程是空闲的；
2.  如果线程池中的线程数量大于等于 `corePoolSize` 且小于 `maximumPoolSize`，则只有当 `workQueue` 满时才创建新的线程去处理任务；
3.  如果设置的 `corePoolSize` 和 `maximumPoolSize` 相同，则创建的线程池的大小是固定的，这时如果有新任务提交，若 `workQueue` 未满，则将请求放入 `workQueue` 中，等待有空闲的线程去从 `workQueue` 中取任务并处理；
4.  如果运行的线程数量大于等于 `maximumPoolSize`，这时如果 `workQueue` 已经满了，则通过 `handler` 所指定的策略来处理任务；

所以，任务提交时，判断的顺序为 `corePoolSize` –> `workQueue` –> `maximumPoolSize`。

### 源码 ###

#### execute 方法 ####

public void execute(Runnable command) {
        // 首先判断任务是否为null，如果为null，抛出空指针异常
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        // 获取ctl
        int c = ctl.get();
        // workerCountOf方法是获取工作线程数
        // 如果当前工作线程数小于corePoolSize核心线程数
        if (workerCountOf(c) < corePoolSize) {
            // 添加核心线程，其中true是根据corePoolSize判断，false是根据maximumPoolSize判断
            if (addWorker(command, true))
                // 如果添加核心线程成功，直接return
                return;
            // 如果失败，再次获取ctl
            c = ctl.get();
        }
        // 如果当前线程池状态是Running，并且任务添加到workQueue队列中成功
        if (isRunning(c) && workQueue.offer(command)) {
            // 再次获取ctl
            int recheck = ctl.get();
            // 再次判断线程池状态，如果线程池状态不是Running了，需要将刚才添加到workQueue队列中的任务移除
            if (! isRunning(recheck) && remove(command))
                // 移除workQueue队列中的任务
                reject(command);
            // 如果工作线程数为0，则执行addWorker方法
            // 第一个参数为null，表示在线程池中创建一个线程，但不去启动
            // 第二个参数为false，代表是根据maximumPoolSize进行判断
            // 如果判断workerCount大于0，则直接返回，在workQueue中新增的command会在将来的某个时刻被执行。
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        // 如果执行到这里，有两种情况：
        // 1. 线程池已经不是RUNNING状态；
        // 2. 线程池是RUNNING状态，但workerCount >= corePoolSize并且workQueue已满。
        // 这时，再次调用addWorker方法，但第二个参数传入为false，将线程池的有限线程数量的上限设置为maximumPoolSize；
        // 如果失败则拒绝该任务
        else if (!addWorker(command, false))
            reject(command);
    }
    复制代码

我们在执行 `execute()` 方法时，如果线程池状态一直是 `RUNNING` 的话：

1.  如果 `workerCount < corePoolSize`，则创建并启动一个线程来执行新提交的任务；
2.  如果 `workerCount >= corePoolSize`，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中；
3.  如果 `workerCount >= corePoolSize && workerCount < maximumPoolSize`，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务；
4.  如果 `workerCount >= maximumPoolSize`，并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。

这里要注意一下 `addWorker(null, false);`，也就是创建一个线程，但并没有传入任务，因为任务已经被添加到workQueue中了，所以 `worker` 在执行的时候，会直接从 `workQueue` 中获取任务。所以，在 `workerCountOf(recheck) == 0` 时执行 `addWorker(null, false);` 也是为了保证线程池在 `RUNNING` 状态下必须要有一个线程来执行任务。

#### addWorker 方法 ####

`addWorker()` 方法的主要工作是在线程池中创建一个新的线程并执行。其中 `firstTask` 参数为新创建线程第一个需要执行的任务。`core` 参数如果为 `true`，则使用 `corePoolSize` 进行判断，`false` 则使用 `maximumPoolSize`。

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            // 获取ctl
            int c = ctl.get();
            // 获取线程池状态
            int rs = runStateOf(c);
    
            // Check if queue empty only if necessary.
            // 首先，如果rs >= SHUTDOWN，则表示线程池不再接收新任务
            // 之后，再判断后面三个条件，只要有一个不满足，就返回false
            // 如果rs == SHUTDOWN，因为不再接收任务了，firstTask不是null，返回false
            // 如果firstTask是null，并且workQueue为空，也返回false
            // 因为workQueue已经为空了，不需要再创建线程了
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
    
            for (;;) {
                // 获取工作线程
                int wc = workerCountOf(c);
                // 首先判断wc >= CAPACITY，如果大于等于工作线程最大值，返回false
                // 之后，如果core是true，则与corePoolSize比较，是false，则与maximumPoolSize比较
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                // 对工作线程执行+1操作，
                if (compareAndIncrementWorkerCount(c))
                    // 如果+1成功，跳出外层循环
                    // CAS方式，并发操作，只能有一个线程+1成功
                    break retry;
                c = ctl.get();  // Re-read ctl
                // 判断线程池状态是否有变化
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
        // 工作线程是否启动标识
        boolean workerStarted = false;
        // 工作线程是否添加表示标识
        boolean workerAdded = false;
        // Worker就是工作线程
        Worker w = null;
        try {
            // 创建新的工作线程，将firstTask传入Worker对象
            w = new Worker(firstTask);
            // 获取线程
            final Thread t = w.thread;
            if (t != null) {
                // 加锁
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    // 获取线程池状态
                    int rs = runStateOf(ctl.get());
                    // 如果线程池是RUNNING状态，或者线程池是SHUTDOWN状态且firstTask是null
                    // 向线程池中添加线程
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        // workers是HashSet，保存线程池中的所有工作线程
                        workers.add(w);
                        int s = workers.size();
                        // 记录线程池中出现的最大线程数
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        // 将线程添加状态置为true，代表线程添加成功
                        workerAdded = true;
                    }
                } finally {
                    // 释放锁
                    mainLock.unlock();
                }
                // 如果线程添加成功，启动线程，将线程启动状态置为true
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
    复制代码

#### Worker 类 ####

线程池中，每一个线程被封装成一个 `Worker` 对象，`ThreadPool` 维护的其实就是一组 `Worker` 对象。

我们可以看到 `Worker` 类继承了 `AQS`，实现了 `Runnable` 接口。其中包括两个属性，`firstTask` 和 `thread`。其中 `firstTask` 用来保存传入的任务，而 `thread` 是在调用构造方法时通过 `ThreadFactory` 类来创建的线程，用来处理任务。

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        /**
         * This class will never be serialized, but we provide a
         * serialVersionUID to suppress a javac warning.
         */
        private static final long serialVersionUID = 6138294804551838833L;
    
        /** Thread this worker is running in.  Null if factory fails. */
        final Thread thread;
        /** Initial task to run.  Possibly null. */
        Runnable firstTask;
        /** Per-thread task counter */
        volatile long completedTasks;
    
        /**
         * Creates with given first task and thread from ThreadFactory.
         * @param firstTask the first task (null if none)
         */
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
    
        /** Delegates main run loop to outer runWorker  */
        public void run() {
            runWorker(this);
        }
    
        // Lock methods
        //
        // The value 0 represents the unlocked state.
        // The value 1 represents the locked state.
    
        protected boolean isHeldExclusively() {
            return getState() != 0;
        }
    
        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }
    
        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }
    
        public void lock()        { acquire(1); }
        public boolean tryLock()  { return tryAcquire(1); }
        public void unlock()      { release(1); }
        public boolean isLocked() { return isHeldExclusively(); }
    
        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }
    复制代码

#### runWorker 方法 ####

因为 `Worker` 类实现了 `Runnable` 接口，因此，在 `Worker` 对象启动的时候，会调用重新的 `run` 方法，`run` 方法中调用了 `runWorker` 方法。

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        // 获取第一个任务
        Runnable task = w.firstTask;
        w.firstTask = null;
        // 允许中断
        w.unlock(); // allow interrupts
        // 是否因为异常退出循环
        boolean completedAbruptly = true;
        try {
            // 如果task为null，则通过getTask来获取任务
            // 也就是Worker自身有携带任务，直接执行
            // 如果Worker自身没有携带任务，那么就通过getTask方法取工作队列中获取任务
            while (task != null || (task = getTask()) != null) {
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                // 判断线程池状态是否大于等于STOP，如果是，要中断当前线程
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     // DCL
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        // 执行任务
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }
    复制代码

`runWorker` 方法的执行过程：

1.  `while` 循环不断地通过 `getTask()` 方法获取任务；
2.  `getTask()` 方法从阻塞队列中取任务；
3.  如果线程池正在停止，那么要保证当前线程是中断状态，否则要保证当前线程不是中断状态；
4.  调用 `task.run()` 方法执行任务；
5.  如果 `task` 为 `null` 则跳出循环，执行 `processWorkerExit()` 方法；
6.  `runWork()` 方法执行完毕，也代表着 `Worker` 中的 `run` 方法执行完毕，摧毁线程；

#### getTask 方法 ####

`getTask` 方法主要用来在阻塞队列中获取任务。

private Runnable getTask() {
        // timeOut变量的值表示上次从阻塞队列中取任务时是否超时
        boolean timedOut = false; // Did the last poll() time out?
    
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
    
            // Check if queue empty only if necessary.
            // 如果线程池状态rs >= SHUTDOWN，也就是说是非RUNNING状态，那么再判断
            // rs >= STOP，线程池是否正在stop
            // 阻塞队列为空
            // 如果上面两个条件有一个满足，需要工作线程数-1，并返回null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
    
            int wc = workerCountOf(c);
    
            // Are workers subject to culling?
            // 判断核心线程是否允许超时
            // 判断工作线程数时候大于核心线程数
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            // 判断工作线程是否超过最大线程数
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                // 工作线程数-1
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
    
            try {
                // 如果是非核心，走poll，拉取工作队列任务
                // 如果是核心线程，走take一直阻塞，拉取工作队列任务
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }
    复制代码

#### processWorkerExit 方法 ####

private void processWorkerExit(Worker w, boolean completedAbruptly) {
        // 如果completedAbruptly为true，说明线程在执行时出现了异常，需要对工作线程数减1
        // 执行正常的情况下，是在getTask方法中进行减1操作
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
            decrementWorkerCount();
    
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 将当前工作线程完成的任务个数赋值给整个线程池中的任务数
            completedTaskCount += w.completedTasks;
            // 从workers中移除，表示从线程池中移除了一个工作线程
            workers.remove(w);
        } finally {
            mainLock.unlock();
        }
        // 判断线程池是否可以中止，线程池状态是否发生变化
        tryTerminate();
    
        int c = ctl.get();
        // 如果当前线程池状态小于STOP
        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);
        }
    }

[format_png]: https://img-blog.csdnimg.cn/img_convert/fb871e409ac5a4fb75235b45d75362cf.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/a7456e9d93a37cfebc7e569922845925.webp?x-oss-process=image/format,png