【Java并发编程】ThreadPoolExecutor 源码解读

淩亂°似流年 2023-01-01 11:53 199阅读 0赞

### 一. 介绍 ###

##### 1.1 池化思想 #####

在说线程池之前先说下池化思想，池化的出现是为了解决资源管理与资源分配的问题，计算机中，基于池化思想诞生了数据库连接池，内存池等，而线程连接池也是池化思想的产物。我们知道，线程的创建与销毁是需要耗费资源的，所以我们通过线程池对线程做统一的管理，可以达到降低资源消耗，提高响应速度，并提供更多的功能。

##### 1.2 类结构 #####

我们先来看看ThreadPoolExecutor的一个UML图  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTgyNjA0_size_16_color_FFFFFF_t_70_pic_center][]

1.  Executor：接口注释告诉我们，它提供了一种思想，即将任务提交和任务执行进行解耦。用户无需关注如何创建线程，如何调度线程来执行任务，用户只需提供Runnable对象，将任务的运行逻辑提交到执行器(Executor)中，由Executor框架完成线程的调配和任务的执行部分。而我们要讨论的ThreadPoolExecutor就很好的实现了这一思想
2.  ExecutorService：接口增加了一些能力，如停止线程池运行等管控线程的方法
3.  AbstractExecutorService：上层的抽象类，实现了ExecutorService接口，对下提供默认实现
4.  ThreadPoolExecutor：最底层的实现类

##### 1.3 运行机制 #####

线程池在内部构建了一个生产消费者模型，将任务与执行任务线程通过阻塞队列进行了解耦，所以我们也将线程池的运行方式按照这个模型分为下面两个部分

1.  任务管理部分：生产者的角色，当任务提交后，线程池会根据状态来决定该任务后续的流转：
    
    （1）申请新线程执行该任务；
    
    （2）放到队列中等待线程执行；
    
    （3）拒绝该任务
2.  线程管理部分：消费者的角色，它们被统一维护在线程池内，根据任务请求进行线程的分配，当线程执行完任务后则会继续获取新的任务去执行，最终当线程获取不到任务的时候，线程就会被回收。

##### 1.4 生命周期 #####

生命周期，在ThreadPoolExecutor里面是通过属性ctl来表示的，这个属性在后面源码分析中会具体阐述，这里我们只要知道线程池主要有几种生命状态，各自代表的含义是什么，状态是如何流转的

1.  RUNNING：可接受新任务，可处理队列中的任务
2.  SHUTDOWN：不接受新任务，可处理队列中的任务
3.  STOP：不接受新任务，不处理队列中的任务，中断正在进行的任务
4.  TIDYING：所有任务都终止，工作线程数为0，且会运行terminated()的钩子方法
5.  TERMINATED：terminated()方法结束

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTgyNjA0_size_16_color_FFFFFF_t_70_pic_center 1][]

### 二.源码解析 ###

##### 2.1 重要属性 #####

线程池中的"运行状态"和"线程的数量"是两个很重要的状态，一般我们都会通过两个变量来进行维护，但是 Doug Lea 却使用一个32位的二进制数"ctl"来维护，其高三位代表"运行状态"，低29位代表"线程的数量"。将本需维护的两个变量减少为一个，这样大大提高了代码的可维护性。

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    // 结果：32-3 = 29
    private static final int COUNT_BITS = Integer.SIZE - 3;
    // 000 11111111111111111111111111111
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
    // 结果： 111 00000000000000000000000000000
    // -1 : 100 00000000000000000000000000001 
    // 取反 : 111 11111111111111111111111111110
    // +1 : 111 11111111111111111111111111111
    // 左移29位 : 111 00000000000000000000000000000
    private static final int RUNNING    = -1 << COUNT_BITS;
    // 结果： 000 00000000000000000000000000000
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    // 结果： 001 00000000000000000000000000000
    private static final int STOP       =  1 << COUNT_BITS;
    // 结果： 010 00000000000000000000000000000
    private static final int TIDYING    =  2 << COUNT_BITS;
    // 结果： 011 00000000000000000000000000000
    private static final int TERMINATED =  3 << COUNT_BITS;
    
    // 跟踪达到的最大池大小。仅在mainLock下访问。
    private int largestPoolSize;

工作状态的判定

// 获取线程池的状态：C & 111 00000000000000000000000000000
    private static int runStateOf(int c)     {  return c & ~CAPACITY; }
    // 获取线程数量：C & 000 11111111111111111111111111111
    private static int workerCountOf(int c)  {  return c & CAPACITY; }
    // 获取ctl的值
    private static int ctlOf(int rs, int wc) {  return rs | wc; }

其他类

// 改队列用来保留任务，并移交给work线程。因为workQueue.isEmpty()并不意味着队列是空的，因此我们只根据isEmpty方法看队列是否为空(在状态从SHUTDOWN变为TIDYING时需要判断)
    private final BlockingQueue<Runnable> workQueue;
    
    // 锁定时要锁定工人位置和相关簿记。虽然我们可以使用某种并发集，但事实证明，通常最好使用锁。 原因之一是它可以对interruptIdleWorkers进行序列化，从而避免了不必要的中断风暴，尤其是在关机期间。 否则，退出线程将同时中断那些尚未中断的线程。 它还简化了一些相关的统计数据，如largePoolSize等。我们还在shutdown和shutdownNow上保留mainLock，以确保工作集稳定，同时分别检查中断许可和实际中断许可
    private final ReentrantLock mainLock = new ReentrantLock();
    
    // 集包含池中的所有worker。一般只有获取mainLock时才能访问。这也是真正的"线程池"
    private final HashSet<Worker> workers = new HashSet<Worker>();
    
    // 等待条件以支持awaitTermination
    private final Condition termination = mainLock.newCondition();
    
    // 新线程的工厂,所有线程都是使用此工厂创建的 (通过addWorker方法)
    private volatile ThreadFactory threadFactory;

##### 2.2 worker类 #####

Worker类主要维护线程运行时的中断状态，以及其他线程相关的参数记录。

其中需要特别注意woker的构造函数中的此段代码 “this.thread = getThreadFactory().newThread(this)”；代码中this指的是woker实例，因此我们创建的线程最终执行的代码是woker实例的run方法

另外，为了抑制直到线程真正开始运行任务之前的中断，我们将锁定状态初始化为负值，并在启动时将其清除（在runWorker中）。

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    { 
        // woker类永远都不会序列化，这里加一个序列化id是为了消除警告
        private static final long serialVersionUID = 6138294804551838833L;
    
        // 此worker类工作的线程
        final Thread thread;
        // worker运行的初始任务，可以为null
        Runnable firstTask;
        // 每个线程完成的任务数
        volatile long completedTasks;
    
      	// 通过firstTask和线程工厂创建的thread来构建一个Worker
        // 这里需要特别注意，其thread的最终传入的是this，即worker实例本身，后续调用thread.start的时候，实际调用的是worker的run方法
        Worker(Runnable firstTask) { 
            setState(-1); // 禁止中断直到runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);//this是Worker实例，即把worker自己作为thread的运行任务
        }
    
        // 运行此线程
        public void run() { 
            runWorker(this); //this指的worker
        }
    
        // 判断是否持有锁
        protected boolean isHeldExclusively() { 
            return getState() != 0;
        }
    		
      	// 重写AQS的tryAcquire方法
        // 如果state置位成功，即抢锁成功，则将独占线程置为自己，返回true；否则返回false
        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) { 
                }
            }
        }
    }

##### 2.3 execute #####

此方法是在将来的某个时间执行给定的任务，即我们最常用的执行任务的方法。针对不同的情况，我们将会使用不同的策略去处理任务command：  
1.如果此时只有少于corePoolSize的线程在运行，尝试使用command启动一个新的线程作为first task。调用addWorker方法会自动检查运行状态(runState)和工作的线程数(workCount)，通过返回false，可以阻止不应该添加线程的错误警报  
2.如果任务能够成功的进入队列，我们仍然需要双重检查是否需要添加线程(因为存在上次检查后死去的线程)或者当进入此方法后线程已经死去的情况。所以我们需要重新检查状态，并且在必要时回滚入队(如果停止)，如果没有线程则新起一个线程  
3.如果我们无法将任务排队，则尝试添加一个新线程。如果失败，我们知道我们已关闭或已饱和，因此拒绝该任务。

我们具体来看下源码

public void execute(Runnable command) { 
        if (command == null)
            throw new NullPointerException();
        // 获取ctl值，如果此时
        int c = ctl.get();
        // 情况1：如果线程数少于线程池核心线程数
        if (workerCountOf(c) < corePoolSize) { 
          	// 调用addWorker方法，
            // 如果返回true,则说明这个任务被分配到了一个工作线程
            // 如果返回false,则说明该任务目前尚未被处理，原因有以下两种
            // 1.则说明在分配过程中线程池状态或者工作线程数发生了变更(并发)，即未生成worker
            // 2.worker生成但worker中的线程启动失败(即t.start()失败，概率很小)，然后又通过addWorkerFailed移除了
            if (addWorker(command, true))
                return;
            // 再获取一次ctl
            c = ctl.get();
        }
        // 情况2：如果线程池是在运行状态，且该任务入队成功
        if (isRunning(c) && workQueue.offer(command)) { 
            // 我们再次获取ctl
            int recheck = ctl.get();
            // 如果线程池是非运行状态，且将刚刚的任务移除队列，如果移除队列成功则执行异常策略
            if (! isRunning(recheck) && remove(command))
                reject(command);
            // 如果此时的工作线程数是0，则增加一个工作线程
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        // 情况3：线程池非运行状态 或 任务入队失败
        else if(!addWorker(command, false))
            reject(command);
    }

##### 2.4 addWorker #####

检查是否可以针对当前池状态和给定的边界（核心数或最大线程数值）添加新的worker，最后启动worker

private boolean addWorker(Runnable firstTask, boolean core) { 
      
        /** 步骤一:状态、工作线程数检查，如果成功工作线程+1 */
        retry:
        for (;;) { 
            int c = ctl.get();
            int rs = runStateOf(c);
            /** 步骤1.1 ：状态rs检查 */
            // 如果线程池的状态rs是不等于RUNNIG的或者当rs为RUNNIG但不满足firstTask为null或者队列不为空时，则返回false
          	// 换过来说：如果rs为running,或者rs为SHUTDOWN且firstTask为null且工作队列不为null时，则不会进入if语句
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
    				
            /** 步骤1.2 ：工作线程数wc检查+增加工作线程数量wc */
            for (;;) { 
                // 获得此时的工作线程数
                int wc = workerCountOf(c);
                // 如果此时wc大于最大容量或者大于我们要比较的corePoolSize或maximumPoolSize，则返回false
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                // 到这一步，说明线程池状态与工作线程数量都检查通过
                // 通过cas操作将工作线程数+1
                // 如果成功，则跳出for自旋
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                // workCount增加失败，再次获取ctl
                // 如果此时的状态和开始的在状态不同，则需要跳到外for循环重新执行一次
                c = ctl.get();  
                if (runStateOf(c) != rs)
                    continue retry;
                // 到了这里，说明状态没变，是上面cas增加工作线程数量失败，原因是并发导致的workCount变化，所以我们需要在再内for循环
            }
        }
        
        /** 步骤二：到这里说明工作线程数量增加成功，我们要新建Woker了 */
        // worker是否开始 与 worker是否添加到wokers中 (HashMap)
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try { 
            // 新建一个worker
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) { 
                // 加锁
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try { 
                    
                    // 获取锁是需要时间的，所以在这期间可能会存在线程池关闭等情况
                    // 所以获取锁后需要重新获取rs,检查状态
                    int rs = runStateOf(ctl.get());
    								
                    // 又是对状态的判断
                    // rs为RUNNING 或者 rs为SHUTDOWN且firstTask为null才会进入
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) { 
                        if (t.isAlive()) // 这里如果t已经start了，则需要抛异常，start是放入works后才start
                            throw new IllegalThreadStateException();
                        // 状态检查成功后，就可以将此worker放入hashMap结构的works中了
                        workers.add(w);
                        // 更新largestPoolSize
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        // 将workerAdded置为true
                        workerAdded = true;
                    }
                } finally { 
                    mainLock.unlock();
                }
                // 如果workerAdded为true，则启动线程，并将workerStarted置为true
                if (workerAdded) { 
                    // 启动线程，执行Woker类中的run方法
                    t.start();
                    workerStarted = true;
                }
            }
        } finally { 
            // 如果线程启动失败，则调用addWorkerFailed方法
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

##### 2.5 runWorker #####

addWorker中的t.start()调用的是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()不为null，则进入while循环
            // 否则跳出while循环
            while (task != null || (task = getTask()) != null) { 
                w.lock();
    						// ( 
                // (运行状态大于等于Stop||(当前线程被中断过 && 运行状态大于等于Stop)) && 当前线程没有被中断过
              	// ) 
                // 如果满足则中断现在的线程
                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 { 
                  	// 回收线程，完成任务数加1，解锁
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            // 到这说明task==null且getTask也为null了，则说明此线程任务完成
            // 到这也说明线程没被异常中断过
            completedAbruptly = false;
        } finally { 
          	// 后续处理操作：维护状态，销毁线程
            processWorkerExit(w, completedAbruptly);
        }
    }

##### 2.6 getTask #####

此方法功能是从阻塞队列获取任务，如果阻塞队列为空，则此线程会阻塞等到有任务为止（或者超时阻塞），源码如下

private Runnable getTask() { 
        boolean timedOut = false; // Did the last poll() time out?
    
        for (;;) { 
            int c = ctl.get();
            int rs = runStateOf(c);
    
            // 线程池状态为SHUTDOWN且队列为空，则不需要获取任务了，直接将wc减一,返回null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 
                decrementWorkerCount();
                return null;
            }
    
            int wc = workerCountOf(c);
    
            // 判断此线程是否是可回收的线程
            // 如果allowCoreThreadTimeOut为true，或者线程数超过了核心线程数，则此线程是超时获取任务的
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    				
            // 重新检查：如果工作线程大于最大线程数 或者 队列中没有任务。则线程数减1且返空
            // 注意并发情况下cas操作是会失败的，失败则重新for自旋
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) { 
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
    
            try { 
                // 如果timed是可回收线程，则限时获取任务，否则阻塞获取
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) { 
                timedOut = false;
            }
        }
    }

##### 2.7 processWorkerExit #####

这个方法主要做善后处理，直接看源码吧

private void processWorkerExit(Worker w, boolean completedAbruptly) { 
        // 如果completedAbruptly为true，说明runWorker非正常退出，workerCount未处理，所以我们减少工作线程
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
            decrementWorkerCount();
    
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try { 
          	// 拿到锁后将完成任务数
            completedTaskCount += w.completedTasks;
            // 将此worker移除
            workers.remove(w);
        } finally { 
            mainLock.unlock();
        }
    		
      	// 判断是否要终止线程
        tryTerminate();
    
        /** 下面步骤主要是根据线程池的状态来判断是否需要新增worker **/
        int c = ctl.get();
        //如果线程池状态大于STOP则直接退出方法
        if (runStateLessThan(c, STOP)) { 
          	// 如果此worker没有异常退出
            if (!completedAbruptly) { 
                // 如果核心线程是带超时的，则 min=0，否则 min=corePoolSize
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                // 如果min为0 且 队列不为空，说明此时queue中还有任务，所以我们至少需要一个worker来完成任务，所以将min置为1
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;
                // 如果当前线程数大于min，则说明不需要新增worker，返回
                if (workerCountOf(c) >= min)
                    return; 
            }
          	// 增加worker去处理任务
            addWorker(null, false);
        }
    }

线程池的主要流程以及相关的方法就讲解完了，关于线程池的其他部分如拒绝策略，异常处理等这里就不详细阐述了，大家有兴趣的可以自己看看。

(完)

欢迎大家关注我的公众号 “程序员进阶之路”，里面记录了一个非科班程序员的成长之路 。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTgyNjA0_size_16_color_FFFFFF_t_70_pic_center 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTgyNjA0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/04fe8cc2dac94dd8b53f83b4e80d8330.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTgyNjA0_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/11f98067a931442c96819602fd60d773.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTgyNjA0_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/3f4122ff305e406e967a0a325742ca02.png