Java ThreadPoolExecutor线程池原理详解编程语言

前言

《Java 线程池理解》简要梳理java线程池ThreadPoolExecutor的使用和主要流程，这篇在此基础上继续学习ThreadPoolExecutor的原理，顺带巩固AQS和CAS的知识。

ThreadPoolExecutor构造函数

public ThreadPoolExecutor(int corePoolSize,  
    int maximumPoolSize,  
    long keepAliveTime, 
    TimeUnit unit, 
    BlockingQueue<Runnable> workQueue, 
    ThreadFactory threadFactory,  
    RejectedExecutionHandler handler); 

ThreadPoolExecutor主要参数

1. corePoolSize 核心线程数量
2. maximumPoolSize 最大线程数量
3. unit 超时时间单位 ( TimeUnit.SECONDS 等)
4. keepAliveTime 超时时间，当线程空闲超过keepAlive会被回收，保证线程数量不大于核心线程数量，表示核心线程已经够应对当前的任务了
5. workQueue 任务阻塞队列，用来缓存待执行的任务
6. ThreadFactory 负责创建Thread线程
7. RejectedExecutionHandler 拒绝策略，当任务缓存队列满了，线程也是最大数量了，此时执行拒绝策略

ThreadPoolExecutor执行流程

AQS的使用

在分析ThreadPoolExecutor之前，先学习AbstractQueueSynchronizer，简称AQS，他是实现ReentrantLock、CountDownLatch的重要部分，ThreadPoolExecutor 也使用了AQS实现非重入锁机制。可以参考《ReentrantLock原理从开始到放弃》

简要说一下AQS的实现原理以及优势

AQS 基本数据结构是一个FIFO的双向队列，每个结点Node存储线程和其他信息；

队列的头部结点表示：该结点对应的线程已经处于执行状态，占用了资源；剩下的队列里的线程则被挂起等待唤醒。

AQS实现了一个FIFO的队列，让线程根据规则排列阻塞和争夺资源。

AQS提供了独占方式和共享方式，对应独占锁和共享锁。

AQS提供模板方法需要具体子类实现功能

• isHeldExclusively()：该线程是否正在独占资源
• tryAcquire(int)：独占方式下尝试去获取资源
• tryRelease(int)：独占方式下尝试释放资源
• tryAcquireShared(int)：共享方式下，尝试获取资源。返回值表示剩余资源，负数表示失败
• tryReleaseShared(int)：共享方式

以上方法不需要全部实现，根据获取的锁的种类可以选择实现不同的方法。

支持独占锁的同步器应该实现tryAcquire、 tryRelease、isHeldExclusively

支持共享获取的同步器应该实现tryAcquireShared、tryReleaseShared、isHeldExclusively。

AQS获取、修改state状态

• getState() 获取状态（资源）
• setState() 设置状态（资源）
• compareAndSetState() CAS设置状态（资源）

state可以在不同的场景下表达不同的意义，例如在ReentrantLock表示重入次数，CountDownLatch表示当前计数。

以ReentrantLock为例子：

这里只抽取了ReentrantLock 的公平锁部分核心代码看下：

 static final class FairSync extends Sync {
    
        private static final long serialVersionUID = -3000897897090466540L; 
        final void lock() {
    // 线程调用lock 
            acquire(1); // 调用AQS的acquire()内部会顺序获取 
        } 
        // 具体的尝试获取锁方法，state 
        protected final boolean tryAcquire(int acquires) {
    
            final Thread current = Thread.currentThread(); 
            int c = getState(); // 获取当前状态 
            // 当且state为0时候才可以表示获取锁 
            if (c == 0) {
    
            // hasQueuedPredecessors 表示是否有其他线程排队， 
            // 如果没人排队就不必入队，直接CAS尝试去获取锁 
                if (!hasQueuedPredecessors() && 
                    compareAndSetState(0, acquires)) {
    
                    // 获取锁成功，设置当前线程独占 
                    setExclusiveOwnerThread(current); 
                    return true; 
                } 
            } 
            // 当前线程可重入 
            else if (current == getExclusiveOwnerThread()) {
    
                int nextc = c + acquires; 
                if (nextc < 0) 
                    throw new Error("Maximum lock count exceeded"); 
                setState(nextc); 
                return true; 
            } 
            return false; 
        } 
    } 

线程池状态

线程池每一时刻都会处于以下一种状态。

RUNNING: 可接收新的任务，当前有任务正在排队、执行
SHUTDOWN: 不接收新的任务，当前有任务正在排队、执行
STOP: 不接收新的任务，不允许任务执行、排队，中断正在执行的任务
TIDYING: 所有任务都被终止，无执行任务线程，会执行terminated()
TERMINATED: terminated() 方法执行完成

// 线程池状态 + 数量 
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); 
 
private static final int COUNT_BITS = Integer.SIZE - 3; 
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; 

RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED

ctl有32位bit，其中前3位表示线程状态，后面39位表示线程数量，所以线程数最大为2的39次方-1。

提供几个方法：

// 获取线程池状态 
private static int runStateOf(int c)     {
    return c & ~CAPACITY; } 
// 获取线程数 
private static int workerCountOf(int c)  {
    return c & CAPACITY; } 
// 利用位运算集合线程池状态和数量，计算ctl 
private static int ctlOf(int rs, int wc) {
    return rs | wc; } 

worker

利用worker实现非可重入锁

private final class Worker extends AbstractQueuedSynchronizer implements Runnable  
    /** 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; 
 
    // 执行的第一个任务，后续任务从blockQueue获取 
    Worker(Runnable firstTask) {
    
        // 设置为-1，不允许被线程获取锁，后续到runWorker()内部会解除这个状态 
        setState(-1); // inhibit interrupts until runWorker 
        this.firstTask = firstTask; 
    	// 由定义的ThreadFactory创建线程 
        this.thread = getThreadFactory().newThread(this); 
    } 
 
     /** Delegates main run loop to outer runWorker. */ 
    public void run() {
    
        // 后续详细分析 
        runWorker(this); 
    } 
 
    // 0表示无锁，非0表示锁占用状态 
    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; 
    } 
 
    // 直接acquire，实现非重入锁，对比下ReentrantLock 
    public void lock()        {
    acquire(1); } 
    public boolean tryLock()  {
    return tryAcquire(1); } 
    public void unlock()      {
    release(1); } 
    public boolean isLocked() {
    return isHeldExclusively(); } 
} 

对比下ReentrantLock和Worker的tryAcquire()，Worker中并没有允许同个线程多次获取锁，也没有做过多优化。

runWorker()

再看下runWorker()是我们创建的线程死循环执行取任务、执行任务的过程，简单的看下具体流程。

final void runWorker(Worker w) {
    
       // 运行在创建的子线程中 
        Thread wt = Thread.currentThread(); 
        Runnable task = w.firstTask; 
        w.firstTask = null; 
        // 解锁状态，对应上面 setState(-1);  
        w.unlock(); // allow interrupts 
        // 表示任务执行是否被中断了 
        boolean completedAbruptly = true; 
        try {
    
            // getTask()会从blockQueue取任务，控制线程数量等操作 
            while (task != null || (task = getTask()) != null) {
    
			    // 锁住worker 
                w.lock(); 
                // 这部分是跟线程池状态有关的判断 
                if ((runStateAtLeast(ctl.get(), STOP) || 
                     (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; 
                    // 完成任务数+1 
                    w.completedTasks++; 
                    w.unlock(); 
                } 
            } 
            // 未被中断 
            completedAbruptly = false; 
        } finally {
    
            // 线程结束操作 
            processWorkerExit(w, completedAbruptly); 
        } 
    } 

getTask() 获取任务(返回null代表结束当前线程)

getTask()负责取出要执行的任务，如果返回null就会走到线程结束的操作

private Runnable getTask() {
    
        boolean timedOut = false; //超时 
 
        for (;;) {
    
            int c = ctl.get(); 
            int rs = runStateOf(c); 
            // 检测线程池状态 
            // Check if queue empty only if necessary. 
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
    
                decrementWorkerCount(); 
                return null; 
            } 
            // 当前线程数 
            int wc = workerCountOf(c); 
            // 若阻塞超时的话，是否允许回收掉当前线程 
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; 
            // 当前线程数超过最大线程数，或者，超时了且允许被回收 
            if ((wc > maximumPoolSize || (timed && timedOut)) 
                && (wc > 1 || workQueue.isEmpty())) {
    
                if (compareAndDecrementWorkerCount(c)) 
                    return null; 
                continue; 
            } 
            // 从阻塞队列取任务，或者队列为空被阻塞 
            try {
    
                Runnable r = timed ? 
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : 
                    workQueue.take(); 
                if (r != null) 
                    return r; //获取到任务 
                timedOut = true;// 超时 
            } catch (InterruptedException retry) {
    
                timedOut = false; 
            } 
        } 
    } 

getTask()内部是一个for语句循环，

• 阻塞队列不为空，能取到任务直接返回task；
• 阻塞队列为空，阻塞一段时间超时，如果允许核心线程超时被回收，或者当前线程数 > 核心线程数 或者收，则返回null，让当前线程被结束掉；
• 走到这里的留下的核心线程，只能在死循环 + 阻塞队列阻塞等待任务的到来。

processWorkerExit() 结束线程

processWorkerExit()执行线程结束操作，处理了控制线程数、回收线程资源的操作

    private void processWorkerExit(Worker w, boolean completedAbruptly) {
    
        // 线程被中断了 
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted 
            // 将work数减一 
            decrementWorkerCount(); 
        // 因为要操作全局变量，利用ReentrantLock锁 
        final ReentrantLock mainLock = this.mainLock; 
        mainLock.lock(); 
        try {
    
            // 总的任务完成数 
            completedTaskCount += w.completedTasks; 
            // 移除当前worker 
            workers.remove(w); 
        } finally {
    
            mainLock.unlock(); 
        } 
 
        tryTerminate(); 
 
        int c = ctl.get(); 
        // 如果在没有停止状态 
        if (runStateLessThan(c, STOP)) {
    
            // 不是因为异常中断 
            if (!completedAbruptly) {
     
            // allowCoreThreadTimeOut表示是否允许核心线程超时回收，如果允许就可以不添加worker 
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize; 
                if (min == 0 && ! workQueue.isEmpty()) 
                    min = 1; 
                // worker数量超过core thread数量 
                if (workerCountOf(c) >= min) 
                    return; // replacement not needed 
            } 
            // 若因为异常中断，或者当前线程数<coreThread数量，添加一个worker线程 
            addWorker(null, false); 
        } 
    } 

addWorker() 内部主要做两个操作：

1. 校验当前线程池状态
2. 添加一个worker线程

private boolean addWorker(Runnable firstTask, boolean core) {
    
    w = new Worker(firstTask); 
    final Thread t = w.thread; 
    final ReentrantLock mainLock = this.mainLock; 
    mainLock.lock(); 
    workers.add(w); 
    mainLock.unlock(); 
    t.start(); 
} 

到达这里，我们知道Worker是利用AQS实现一个非可重入锁，但是为什么需要一个非可重入锁呢？

每一个Worker在各自的子线程中运行，取任务前lock，在任务结束之后unlock，一旦获取了独占锁，表示当前线程正在执行任务中。

可是也并没有其他线程争夺Worker，争夺的应该是task，而task利用阻塞队列实现多线程竞争了。

不是很理解为什么需要非可重入锁？

后面发现，使用独占锁来表示线程是否正在执行任务，Worker的线程获取了独占锁就说明它在执行任务，不能被中断。

execute流程

execute流程分成三步：

1. 如果当前worker线程少于核心线程数，则addWorker()添加一个线程执行。
2. 如果worker线程数等于核心线程数，则将任务加入到阻塞队列workQueue
3. 阻塞队列满了且未等到最大线程数之前，增加一个worker线程
4. 否则，执行拒绝策略

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

总结

• 线程池会分核心线程和非核心线程吗？

  不会，ThreadPoolThread是控制线程数来保证核心线程数量

• 线程池在哪里控制数量

  利用阻塞队列阻塞，若超时了并且允许回收，则当前线程会被结束，若不允许结束，则当前线程会死循环并且一段时间阻塞的等待任务。

• worker为什么需要不可重入锁

  Worker继承自AQS，用于判断线程是否空闲以及是否可以被中断

  参考https://www.cnblogs.com/liuzhihu/p/8177371.html