signed

QiShunwang

“诚信为本、客户至上”

JAVA并发(四):如何从ThreadPoolExecutor源码细节下手聊线程池实现原理?

2021/6/24 22:15:07   来源:

一、为什么要使用线程池?

  • 降低资源消耗:通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
  • 提高响应速度:当任务到达时,可以不需要等待线程创建就能立即执行。
  • 提高线程的可管理性:线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一的分配,监控和调优。
  • ThreadPoolExecutor可以保存一些基本数据,例如当前线程池完成的任务数。

二、ThreadPoolExecutor源码分析

ThreadPoolExecutor所在JUC包的作者是做C出身的道格-李,而这位大佬的代码风格是:先改状态后做事,做不通再回滚。

下面我们主要从常用变量、构造方法、任务执行过程execute()、addWorker()、runWorker()、processWorkerExit()去分析线程池的实现原理。

1. 常用变量

ThreadPooleExecutor中有一个AtomicInteger类型的成员变量ctl,其高3位用来表示线程池的状态,低29位用来表示线程池的工作线程数量;成员变量COUNT_BITS的值为29,成员变量CAPACITY用来表示worker的最大个数,即2 ^ 29 -1;

// ctl 可以看做一个int类型的数字,高3位表示线程池状态,低29位表示worker数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// COUNT_BITS	Integer.SIZE为32,所以COUNT_BITS为29
private static final int COUNT_BITS = Integer.SIZE - 3;
// 线程允许的最大线程数,1左移29位,然后减1,即2^29 -1
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

线程池状态的获取方式runStateOf()为capacity取反 然后和ctl 做与操作。
线程池工作线程的获取方式workerCountOf()为capacity 和 ctl 做与操作。

// runStateOf() 获取线程池的状态,通过按位与操作,低29位将全部变为0
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// workerCountOf() 获取线程池worker数量,通过按位与操作,高3位将全部变成0
private static int workerCountOf(int c)  { return c & CAPACITY; }

线程池的状态:

  • RUNNING:接受新任务并且处理阻塞队列里的任务;
  • SHUTDOWN:拒绝新任务但是处理阻塞队列里的任务;
  • STOP:拒绝新任务并且抛弃阻塞队列里的任务,同时会中断正在处理的任务;
  • TIDYING:所有任务都执行完(包含阻塞队列里面任务)当前线程池活动线程为 0,将要调用 terminated 方法;
  • TERMINATED:终止状态,terminated方法调用完成以后的状态。

线程池状态转换:

  1. RUNNING -> SHUTDOWN:显式调用 shutdown() 方法;或隐式调用了 finalize(),finalize()里调用了 shutdown() 方法。
  2. RUNNING or SHUTDOWN -> STOP:显式调用 shutdownNow() 方法时。
  3. SHUTDOWN -> TIDYING:当线程池和任务队列都为空时。
  4. STOP -> TIDYING:当线程池为空时。
  5. TIDYING -> TERMINATED:当 terminated() hook 方法执行完成时。
// 线程池有5中状态,按大小排序如下:Running< SHUTDOWN < STOP < TIDING < TERMINATED
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;

// ctlOf(),根据线程池状态和线程池worker数量,生成ctl值
private static int ctlOf(int rs, int wc) { return rs | wc; }

// runStateLessThan(),线程池状态小于xx
private static boolean runStateLessThan(int c, int s) {
    return c < s;
}
// runStateAtLeast(),线程池状态大于等于xx
private static boolean runStateAtLeast(int c, int s) {
    return c >= s;
}

2. 构造方法

我们从最全的构造方法来看其中的参数。

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.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    // 根据传入参数`unit`和`keepAliveTime`,将存活时间转换为纳秒存到变量`keepAliveTime `中
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

从构造中可以看出:核心线程数不能小于0,最大线程数不能小于1,最大线程数不能大于核心线程数,超过核心线程数的线程存活时间不能小于0。

参数介绍:

  1. corePoolSize:线程池核心线程数。
  2. maximunPoolSize:线程池最大线程数。
  3. keepAliveTime:存活时间。如果当前线程池中的线程数量比核心线程数量要多,并且有闲置状态的线程,keepAliveTime表示这些闲置的线程能存活的最大时间。
  4. TimeUnit,存活时间的时间单位。
  5. workQueue:当任务数超过核心线程数时,用于保存等待执行的任务的阻塞队列;比如基于数组的有界 ArrayBlockingQueue,基于链表的无界 LinkedBlockingQueue,最多只有一个元素的同步队列 SynchronousQueue,优先级队列 PriorityBlockingQueue 等。
  6. threadFactory:创建线程的工厂。
  7. handler:拒绝策略,当队列满了并且线程个数达到 maximunPoolSize 后采取的策略。比如 AbortPolicy (直接抛出异常),CallerRunsPolicy(使用调用者所在线程来运行任务),DiscardOldestPolicy(丢弃一个最老的任务,然后执行当前任务),DiscardPolicy(丢失当前任务,没有异常信息)。

3. 任务执行过程execute()

老生常谈,任务的执行过程为:

  1. 首先获取线程池的信息,判断工作线程数是否小于核心线程数,小于则创建核心线程数;
  2. 如果工作线程数超过核心线程数,则将任务添加到阻塞队列中;
  3. 如果阻塞队列满了且工作线程数小于最大线程数,则创建非核心线程;
  4. 如果工作线程数达到最大线程数、线程池状态不是RUNNING状态、或 任务入队失败,则对新添加的任务执行拒绝策略。

这其中有几个细节:

  • 由于其他线程改变线程池的状态导致添加核心线程失败,当前线程重新获取线程池的状态。
  • 多线程环境下,任务进入阻塞队列之后,线程池状态可能会发生改变,所以需要重新获取线程池状态进行recheck。进而判断线程池状态是不是RUNNING状态:
    1.如果不是,说明执行shutdown命令,需要对新加入的任务执行reject()操作;
    2.如果是,还需要判断当前线程池的工作线程数是否为0,如果核心线程数为0,需要创建一个非核心线程来执行任务;因为创建线程池的时候允许核心线程数为0。
  • 如果线程池状态不是RUNNING状态,或 任务进入阻塞队列失败,会尝试创建worker来执行任务,如果创建worker失败,会执行拒绝策略。
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();
        }
        // worker数量超过核心线程数,任务添加到队列中
        if (isRunning(c) && workQueue.offer(command)) {
        // 多线程环境下,任务入队之后,线程池状态也可能发生改变,所以需要重新获取线程池状态进行recheck。
            int recheck = ctl.get();
            // 如果线程池状态不是RUNNING状态,说明执行过shutdown命令,需要对新加入的任务执行reject()操作。
            if (! isRunning(recheck) && remove(command))
                reject(command);
            // 这儿为什么需要判断0值呢?主要是因为线程池构造方法中,核心线程数允许为0。所以当线程数为0的时候,需要使用非核心线程来执行任务
            else if (workerCountOf(recheck) == 0)
                // 添加非核心线程
                addWorker(null, false);
        }
        // 如果线程池不是运行状态,或者任务进入队列失败,则尝试创建worker执行任务。
        // 这儿有3点需要注意:
        // 1. 线程池不是运行状态时,addWorker内部会判断线程池状态
        // 2. addWorker第2个参数表示是否创建核心线程
        // 3. addWorker返回false,则说明任务执行失败,需要执行reject操作
        else if (!addWorker(command, false))
            reject(command);
    }

4. addWorker()方法

1)worker介绍

Worker和和task是有区别的,task是用户提交的线程任务,而worker则是ThreadPoolExecutor自己内部的一个类,主要是对进行中的任务进行中断控制,顺带着对其他行为进行记录,其保存在HashSet中。另外,Worker类继承了AQS,其本身就是一把锁,在runWorker()方法中会用到。


    private final HashSet<Worker> workers = new HashSet<Worker>();

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;

    // 正在跑的线程,如果是null标识factory失败
    final Thread thread;
    // 初始化一个任务以运行,可能为空
    Runnable firstTask;
    // 每个线程的完成任务数
    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;
        // 这儿是Worker的关键所在,使用了线程工厂创建一个线程。传入的参数为当前Worker
        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();}
}

2)addWorker()分析
添加工作线程主要分为两步:判断线程池状态并使用CAS增加工作线程数量、封装Worker并启动工作线程。

第一步:

  • 首先会在retry外层循环中,保存一个CTL的快照,用于后面进行CAS判断线程池的状态是否发生改变。
  • 然后,判断线程池的状态决定是否直接返回false;有三种情况会直接返回FALSE,如下:
    1.当线程池的状态大于SHUTDOWN时;
    2.当线程池的状态等于SHUTDOWN,且firstTask不为null时;
    3.当线程池的状态等于SHUTDOWN,且任务队列为空时;
  • 接着,内循环中获取线程池的工作线程数量,如果core参数为TRUE,即插入核心线程,会判断当前线程数是否大于等于corePoolSize;或者判断当前工作线程数量是否大于等于 工作线程数的最大值Capacity;大于则直接返回FALSE;
  • 然后,尝试使用CAS增加工作线程数,增加成功则退出外城retry循环。
  • 接着,获取线程池的信息ctl,和最初的CTL快照值进行比较,如果线程池状态发生改变,continue 外层循环retry,对外层循环进行自旋。
  • 最后,如果CAS改变工作线程数失败,会进行内循环尝试改变线程池的工作线程数。

第二步:

  • 先定义了一个workerStarted变量,用于判断工作线程是否正常启动;如果启动异常,会调用addWorkerFailed()方法:对workers加独占锁,回滚减少工作线程数量、中断当前工作线程;最后返回FALSE。
  • 接着,将当前任务封装为一个新的Worker;因为Worker的添加必须保证是线程安全的,所以需要加ReentrantLock独占锁。
  • 然后会recheck线程池状态,因为ThreadFactory出现故障 或 在获取锁之前 关闭的话,直接运行finally块中的addWorkerFailed()方法退出即可。
  • 只有当线程池的状态是RUNNING状态 或 线程池的状态是SHUTDOWN且firstTask为null时 才会继续执行线程启动相关后续逻辑:包括check工作线程是否已经启动、添加工作线程到workers,更新largestPoolSize。
  • 最后,释放锁,启动工作线程,将workerStarted变量设置为true并返回。
private boolean addWorker(Runnable firstTask, boolean core) {
    // 1. 判断状态并增加工作线程数量
    retry:
    // 外层自旋
    for (;;) {
        //存一个快照值,后面进行CAS,如果最新的结果不等于快照值,就再跑一遍
        int c = ctl.get();
        int rs = runStateOf(c);

        // 1. 线程池状态大于SHUTDOWN时,直接返回false
        // 2. 线程池状态等于SHUTDOWN,且firstTask不为null,直接返回false
        // 3. 线程池状态等于SHUTDOWN,且队列为空,直接返回false
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        // 内层自旋
        for (;;) {
            int wc = workerCountOf(c);
            // worker数量超过容量,直接返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 使用CAS的方式增加worker数量。
            // 若增加成功,则直接跳出外层循环进入到第二部分
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            // 线程池状态发生变化,对外层循环进行自旋
            if (runStateOf(c) != rs)
                continue retry;
            // 其他情况,直接内层循环进行自旋即可
            // else CAS failed due to workerCount change; retry inner loop
        } 
    }
    
    // 2. 封装worker并启动工作线程
    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;
            // worker的添加必须是串行的,因此需要加锁
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                // 这儿需要重新检查线程池状态
                int rs = runStateOf(ctl.get());

                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    // worker已经调用过了start()方法,则不再创建worker
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    // worker创建并添加到workers成功
                    workers.add(w);
                    // 更新`largestPoolSize`变量
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            // 启动worker线程
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // worker线程启动失败,说明线程池状态发生了变化(关闭操作被执行),需要进行shutdown相关操作
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

5. runWorker()方法

runWorker()方法是Worker的运行逻辑:

  • runWorker会先获取Worker中的firstTask,然后将worker的firstTask引用置为null,接着为了让外部可以中断Worker的运行,会先给Worker解锁。
  • 然后while 自旋调用getTask()方法从任务阻塞队列中获取任务,接着,对Worker进行加锁,保证每个Worker中任务的执行是串行的。
  • 如果线程池正在停止,对当前线程进行中断操作;因为在if条件判断中和代码段中会分别执行一次interrupt()方法,所以会确保执行shutdownNow时,清除中断标志位,中断线程。
  • 然后执行任务,并预留beforeExecute()和afterExecute()两个钩子函数。最后将completedTasks + 1。

此外:runWorker中最外层的try,太容易的是任务的异常,它没有catch,所以监控不到异常。

另外:runWorker()方法中有一个completedAbruptly变量,用来表示任务的执行有没有被异常中断,进而在processWorkerExit()方法保证线程池中的可执行工作线程数量不变。

completedAbruptly为true的三种情况:

  • 调用getTask()从队列获取任务时线程被中断,出现异常。但是这里给包掉了,不会抛出异常。
  • beforeExecute()钩子函数出异常。
  • afterExecute()钩子函数出异常。
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    // 调用unlock()是为了让外部可以中断,在interruptedIdleWorkers方法中会中断空闲线程
    w.unlock();
    // 这个变量用于判断是否进入过自旋(while循环)
    boolean completedAbruptly = true;
    try {
        // 这儿是自旋
        // 1. 如果firstTask不为null,则执行firstTask;
        // 2. 如果firstTask为null,则调用getTask()从队列获取任务。
        // 3. 阻塞队列的特性就是:当队列为空时,当前线程会被阻塞等待
        while (task != null || (task = getTask()) != null) {
          // 这儿对worker进行加锁,是为了达到下面的目的
            // 1. 降低锁范围,提升性能
            // 2. 保证每个worker执行的任务是串行的  
            w.lock();
            // 如果线程池正在停止,则对当前线程进行中断操作,这里要确保执行shutdownNow时,清除中断标志位,中断线程。shutdown不会走这里。
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            // 执行任务,且在执行前后通过`beforeExecute()`和`afterExecute()`来扩展其功能。
            // 这两个方法在当前类里面为空实现。
            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 {
                // 帮助gc
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 自旋操作被退出,说明线程池正在结束
        processWorkerExit(w, completedAbruptly);
    }
}

6. processWorkerExit()

  • processWorkerExit()方法中第一步会先判断Worker是否为异常终止(突然完成),如果是则减少工作线程。
  • Worker任务完成数+1,然后将Worker从workers中移除。接着调用tryTerminate()方法尝试结束线程池:
    1.当线程池状态是RUNNING状态、小于TIDYING状态 活 线程池状态是SHUTDOWN且任务队列不为空时,直接返回。
    2.当工作线程不为0时,尝试中断一个worker。这里只中断一个worker是因为多线程环境下,可能会存在多个worker并发终止Worker,要确保最后留一个线程用来修改线程池的状态。
  • 最后获取线程池状态,当线程池状态为RUNNING、SHUTDOWN时,判断Worker是否为异常终止,如果是,直接创建一个非核心线程;否者,根据allowCoreThreadTimeOut参数判断出最小线程数min,如果min == 0 并且任务队列不为空,创建一个非核心线程;如果工作线程大于min,则直接返回。
private void processWorkerExit(Worker w, boolean completedAbruptly) {
    // 如果由于用户原因出现异常,直接减少工作线程
    if (completedAbruptly) 
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }

    tryTerminate();

    int c = ctl.get();
    // 线程池状态是运行、或shutdown
    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);
    }
}

tryTerminate()中断线程方法

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        // 保证其他空闲工作线程全部被中断,只让最后一个线程改变线程池状态
        if (workerCountOf(c) != 0) {
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

7. shutdown 和 shutdownNow的区别

  • shutdown,不再接收新任务但会执行queue中的任务;shutdownNow不接收新任务也不执行queue中的任务。
  • shutdown将线程状态使用CAS更改为SHUTDOWN;shudownNow更改为STOP
  • shutdown中断空闲工作线程,shutdownNow中断所有工作线程。

如何判断一个线程是否是空闲线程?
看其在runWorker方法中有没有获取到锁,有则不是。

三、 总结

线程池采用了大量的CAS + 快照操作,确保线程池的状态和工作线程数量的正确变更。