【JUC源码】线程池：ThreadPoolExecutor 万字源码深析（超详细注释）

深藏阁楼爱情的钟 2022-12-11 12:22 190阅读 0赞

在文章开始之前，我们先要明白为啥要有线程池这么个东西。线程是一种稀缺资源，若不加以限制，不仅会占用大量资源，还会影响系统的稳定性。而线程池可以对线程的创建与停止、线程数量等等因素加以控制，使得线程在一种可控的范围内运行，在保证系统稳定运行的同时，还使得性能调优更加方便。另外，每次请求到来时，由于线程的创建已经完成，所以可以直接执行任务，减少了每次创建线程、销毁线程的开销，提高了响应速度。

OK，下面我们就进入正戏，源码…

> PS：本文篇幅较长，对于看着感觉费劲的同学，这里还提供了拆分的版本：
> 
>  *  [【JUC源码】线程池：ThreadPoolExecutor（一）继承关系分析][JUC_ThreadPoolExecutor]
>  *  [【JUC源码】线程池：ThreadPoolExecutor（二）底层结构分析][JUC_ThreadPoolExecutor 1]
>  *  [【JUC源码】线程池：ThreadPoolExecutor（三）Worker 设计思路及源码分析][JUC_ThreadPoolExecutor_Worker]
>  *  [【JUC源码】线程池：ThreadPoolExecutor（四） 任务执行流程源码分析][JUC_ThreadPoolExecutor_]
>  *  [【JUC源码】线程池：ThreadPoolExecutor（五）提炼总结][JUC_ThreadPoolExecutor 2]

# 1.结构 #

ThreadPoolExecutor 核心继承关系，成员变量及主要构造函数：

public class ThreadPoolExecutor extends AbstractExecutorService {
        
        
        // ctl 线程池状态控制字段，由两部分组成：
        // int ctlOf(int rs, int wc) { return rs | wc; }
    	private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
        // 1:workerCount  wc 工作线程数，我们限制 workerCount 最大到(2^29)-1，大概 5 亿个线程
        private static final int COUNT_BITS = Integer.SIZE - 3;// 29
        private static final int CAPACITY   = (1 << COUNT_BITS) - 1;// =(2^29)-1=536870911
        // 2:runState rs 线程池的状态，提供了生命周期的控制，源码中有很多关于状态的校验，状态枚举如下：
        private static final int RUNNING    = -1 << COUNT_BITS;//536870912
        private static final int SHUTDOWN   =  0 << COUNT_BITS;//0
        private static final int STOP       =  1 << COUNT_BITS;//536870912
        private static final int TIDYING    =  2 << COUNT_BITS;//1073741824
        private static final int TERMINATED =  3 << COUNT_BITS;//1610612736
    
        
        // Worker 线程池中任务执行的最小单元
        private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
        ...}
        
        
        // 任务队列，阻塞队列，来保持线程的存活周期
        // 注：这里使用阻塞队列而不是普通容器，是为了worker在取任务时借助它的线程调度能力
        //    比如当线程池中的线程空闲下来了，它就会在队列的另一端等待外部线程投递任务
        //    这里注意，因为将外部线程的任务放入任务队列调用的是offer方法，所以入队时不会出现外部线程阻塞
        private final BlockingQueue<Runnable> workQueue;
    	// 工作线程集合，包含线程池中所有的工作线程
        private final HashSet<Worker> workers = new HashSet<Worker>();
        // 锁，大多数情况下是控制对 workers 的访问权限（如将新worker加入）
        private final ReentrantLock mainLock = new ReentrantLock();
        private final Condition termination = mainLock.newCondition();
        
       
        // 已完成任务的计数
        volatile long completedTasks;
        // 线程池最大容量
        private int largestPoolSize;
        // 已经完成的任务数
        private long completedTaskCount;
        
        //-----------------------------用户可控属性（volatile）-----------------------------
        // 可以使用 threadFactory 创建 thread
        // 创建失败一般不抛出异常，只有在 OutOfMemoryError 时候才会
        private volatile ThreadFactory threadFactory;
        // 线程空闲的最大时间
        private volatile long keepAliveTime;
        // coreSize，allowCoreThreadTimeOut决定是否回收
        private volatile int corePoolSize;
        // maxSize，除核心线程外，空闲就会回收
        private volatile int maximumPoolSize;
        // 饱和或者运行中拒绝任务的 handler 处理类
        private volatile RejectedExecutionHandler handler;
        
        // 默认的拒绝策略
        private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
        
        // 设置 true 的话，核心线程空闲 keepAliveTime 时间后，也会被回收
        // 需要调用allowCoreThreadTimeOut方法进行设置，默认false
        private volatile boolean allowCoreThreadTimeOut;
        
        //-------------------------------构造函数----------------------------------------
        // 构造函数的作用就是设置上面的volatile变量们
        // 注：前五参数个必有，theadFactory与rejected可以没有
        public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        ...}
    }

## 1.1 继承体系 ##

![在这里插入图片描述][20210223031044877.png_pic_center]

### Executor ###

定义 execute 方法来执行任务，入参是 Runnable，无出参

public interface Executor {
        
        void execute(Runnable command);
    }

### ExecutorService ###

Executor 的功能太弱，ExecutorService 丰富了对任务的执行和管理的功能，主要代码如下：

public interface ExecutorService extends Executor {
        
        // 提交有返回值的任务，使用 get 方法可以阻塞等待任务的执行结果返回
        <T> Future<T> submit(Callable<T> task);
        // 提交没有返回值的任务，如果使用 get 方法的话，任务执行完之后得到的是 null 值
        Future<?> submit(Runnable task);
        
        // 给定任务集合，返回已经执行完成的 Future 集合，每个返回的 Future 都是 isDone = true 的状态
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException;
        // 给定任务中有一个执行成功就返回，如果抛异常，其余未完成的任务将被取消
        <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException;
        
        // 关闭，不会接受新的任务，也不会等待未完成的任务
        // 如果需要等待未完成的任务，可以使用 awaitTermination 方法
        void shutdown();
        // 在超时时间内，等待剩余的任务终止
        boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException;
        
        // executor 是否已经关闭了，返回值 true 表示已关闭
        boolean isShutdown();
        // 所有的任务是否都已经终止，是的话，返回 true
        boolean isTerminated();
        
    }

### AbstractExecutorService ###

*  抽象类，封装了 Executor 的很多通用功能
 *  实现了部分ExecutorService方法

public abstract class AbstractExecutorService implements ExecutorService {
        
       
        // 将Callabbe，想要返回值的Runnable转化成Cunnable
        protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        
            return new FutureTask<T>(runnable, value);
        }
        // FutureTask(Runnable,T) -> RunnableFuture -> Runnable
        protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        
            return new FutureTask<T>(callable);
        }
        
        // submit调用的还是execute
        // 只不过将Callable，要返回值的Runnable提前转化成了Runnable
        // 提交无返回值的任务
        public Future<?> submit(Runnable task) {
        
            if (task == null) throw new NullPointerException();
            // ftask 其实是 FutureTask
            RunnableFuture<Void> ftask = newTaskFor(task, null);
            execute(ftask);
            return ftask;
        }
        // 提交有返回值的任务
        public <T> Future<T> submit(Callable<T> task) {
        
            if (task == null) throw new NullPointerException();
            // ftask 其实是 FutureTask
            RunnableFuture<T> ftask = newTaskFor(task);
            execute(ftask);
            return ftask;
        }
    }

## 1.2 相关参数 ##

### 线程池状态 ###

*  RUNNING：可以接收新任务，同时也可以处理阻塞队列里面的任务
 *  SHUTDOWN：不可以接收新任务，但可以处理阻塞队列里面的任务
 *  STOP：不可以接收新任务，也不处理阻塞队列里的任务，同时中断正处理的任务
 *  TIDYING：属于过渡阶段，在这个状态表示所有任务已经结束，当前线程池无有效线程，并将调用terminal方法
 *  TERMINALED：终止状态  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]

### 拒绝策略 ###

*  AbortPolicy（默认）：抛出异常
 *  CallerRunsPolicy：不使用线程池，主线程来执行
 *  DiscardPolicy：直接丢弃任务
 *  DiscardOldestPolicy：丢弃队列中最老任务

## 1.3 Worker ##

线程池中任务执行的最小单元，同它的名字一样是执行任务的工具人。

由于Worker实现了Runnable，所以Worker是本质是一个线程任务。那么请先考虑下面三个问题：

1.  为什么Worker要实现Runnable，而不是创建线程时直接用 firstTask？
    
    答：一句话，为了**线程复用**。说直白点就是当线程new出来之后，它的Runnable就不能变了，所以如果直接拿某一个任务去创建线程，那么它就不能再执行别的新任务了，就无法做到复用。
2.  那实际要执行的任务怎么办，放在那里？
    
    答：放置的位置有两个（这段逻辑可以在runWorker方法中看到）：
    
    1.  在创建worker时，将第一个任务（firstTask）通过构造函数组合进来，执行完之后就删掉（置为null）
    2.  将其余任务全部放到任务队列（workQueue）中，用完就删掉
3.  那要是创建的线程多了，好多线程都没有任务空闲下来了怎么办？
    
    答：若一个线程迟迟等不到任务执行就会被回收，具体回收策略在 getTask 方法中可以看到

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        
    	// 属于当前worker的执行任务的线程
    	final Thread thread;
    	// 实际需要执行的任务。顾名思义只保存第一个执行的任务，第一个任务执行完后=null
        Runnable firstTask;
    	
    	Worker(Runnable firstTask) {
        
    		// 将AQS的状态设置为-1
    		// 从后面的isLocked方法可以看到，state!=0 表示已经被加锁
    	    setState(-1); // inhibit interrupts until runWorker
    	    this.firstTask = firstTask;
    		// ！！！为了线程的复用，Worker本身实现了 Runnable,并且把自己作为任务传递给 thread。非常巧妙的设计！
    	    this.thread = getThreadFactory().newThread(this);
    	}
    	
    	// Thread的启动方法start实际调用的就是run，而这里实际有两个run：
        // 1.Worker 本身是 Runnable，run 方法是 Worker 执行的入口
        // 2.runWorker 是外部的方法，会调用firstTask的run方法
        public void run() {
        
            runWorker(this);
        }
    }

另外，Worker 本身也实现了 AQS，所以其本身也是一个锁，其在执行任务的时候，会锁住自己，任务执行完成之后，会释放自己，保证了在一个线程执行任务时再被丢入别的任务。相关方法如下：

public void lock()        {
         acquire(1); }
    public boolean tryLock()  {
         return tryAcquire(1); }
    public void unlock()      {
         release(1); }
    public boolean isLocked() {
         return isHeldExclusively();  // Lock methods    
                               
    // 尝试加锁，CAS 赋值为 1，表示锁住
    protected boolean tryAcquire(int unused) {
        
        if (compareAndSetState(0, 1)) {
        
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }
    
    // 尝试释放锁，释放锁没有 CAS 校验，可以任意的释放锁
    protected boolean tryRelease(int unused) {
        
        setExclusiveOwnerThread(null);
        setState(0);     
        return true; 
    }
    
    // 0 代表没有锁住，否则代表锁住（-1,1）
    protected boolean isHeldExclusively() {
        
        return getState() != 0;
    }
    
    void interruptIfStarted() {
        
           Thread t;
           if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        
               try {
        
                   t.interrupt();
               } catch (SecurityException ignore) {
        
               }
           }
    }

# 2.方法解析 & api #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]

## 2.1 execute() ##

入口，选择执行策略，分为以下三种情况：

*  情况一：工作线程 < 核心数 ，创建一个线程去执行任务
 *  情况二：工作线程 >= 核心数 且 任务队列未满，加入任务队列（等待核心线程来执行）
    
     *  线程池出现异常，删除当前任务
     *  极限情况：入队时可用线程刚好被回收，新建一个没有任务的线程
 *  情况三：任务队列已满
    
     *  队列已满 && 线程数 < maxSize：创建新的线程来处理任务
     *  队列已满 && 线程数 >= maxSize：使用 RejectedExecutionHandler 类拒绝请求

public void execute(Runnable command) {
        
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get(); // 获取 ctl 
        // 情况一：工作的线程小于核心线程数，创建新的线程，成功返回，失败不抛异常
        if (workerCountOf(c) < corePoolSize) {
        
            if (addWorker(command, true))
                return;
            // 由于 addWorker -> runWorker -> getTask，所以线程池状态可能发生变化
            c = ctl.get();
        }
        // 情况二：工作的线程大于等于核心线程数且任务队列没满
        // 注：isRunning是校验线程池状态是否正常。另外，offer不阻塞而是返回t/f
        if (isRunning(c) && workQueue.offer(command)) {
        
            int recheck = ctl.get();
            // 如果线程池状态异常 尝试从队列中移除任务，可以移除的话就拒绝掉任务
            if (!isRunning(recheck) && remove(command))
                reject(command);
            // 发现可运行的线程数是 0，就初始化一个线程，这里是个极限情况，入队的时候，突然发现可用线程都被回收了
            else if (workerCountOf(recheck) == 0)
                // Runnable是空的，不会影响新增线程，但是线程在 start 的时候不会运行
                // Thread.run() 里面有判断
                addWorker(null, false);
        }
        // 情况三：队列满了，开启线程到 maxSize，如果失败直接拒绝（这段逻辑可以在addWorker方法中看到）
        else if (!addWorker(command, false))
            reject(command);
    }

## 2.2 addWorker() ##

创建woker，返回worker中的线程是否成功启动。大致过程如下：

1.  线程池状态校验
    
     *  失败，返回 false。原因有二：
        
         *  线程池状态异常：SHUTDOWN，STOP，TIDYING，TERMINALED
         *  工作线程数溢出：线程数 >= 容量 或 线程数 >= coreSize or maxSize
     *  成功：CAS使workCount加一
2.  创建Worker
    
    1.  创建两个标识变量：workerAdded，workerStarted
    2.  构造woker，在构造时会通过newThread方法创建出一个新线程
    3.  上锁，将新建的 worker加入管理worker的容器（Set）。锁保证了并发时的线程安全
3.  启动worker中的线程，调用逻辑是：Thread\#start -> Worker\#run -> runWorker

// firstTask 不为空可以直接执行，为空执行不了，Thread.run()方法有判断，Runnable为空不执行
    // core 为 true 表示线程最大新增个数是 coresize，false 表示最大新增个数是 maxsize
    private boolean addWorker(Runnable firstTask, boolean core) {
        
        
    	// break retry 跳到retry处，且不再进入循环
    	// continue retry 跳到retry处，且再次进入循环
        retry:
    --------------------------------------------------------------------------------------------------------------    
        // 1.先是各种状态的校验
        for (;;) {
        
            int c = ctl.get();
            int rs = runStateOf(c); // 获取线程池状态
            
            // 1.1 校验线程池状态，rs>=0:SHUTDOWN，STOP，TIDYING，TERMINALED
            if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) 
                return false;
    
            for (;;) {
        
                int wc = workerCountOf(c); // 得到当前工作线程数，即worker数
                // 1.2 校验工作中的线程数大于等于容量，或者大于等于 coreSize or maxSize
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize)) // 如果core为true就判断是否大于coreSize，否则判断maxSize
                    return false;
                // CAS修改workerCount（+1）
                if (compareAndIncrementWorkerCount(c))
                    // break 结束 retry 的 for 循环
                    break retry;
                // 到这里可能是CAS失败了，重新获取 ctl
                c = ctl.get();  
                // 如果线程池状态被更改
                if (runStateOf(c) != rs)
                    continue retry; // 跳转到retry位置，重新判断
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
    --------------------------------------------------------------------------------------------------------------
    	// 2.创建worker
    	// 2.1 创建标识变量
        boolean workerStarted = false; // woker启动标识
        boolean workerAdded = false; // woker成功加入worker容器标识
        Worker w = null;
        try {
        
        	// 2.2 构造worker。在worker的构造函数中会调用newThread方法创建一个Thread
        	// 注：由于Worker也实现了Runnable，所以在创建线程的时候是newThread(this)。这是一个巧妙的设计
            w = new Worker(firstTask);
            final Thread t = w.thread; // 获取worker中的线程
            // 2.3 将worker加入到worker容器（Set）
            if (t != null) {
        
                final ReentrantLock mainLock = this.mainLock; // 这个mainLock是一个成员变量，作用是控制对worker的操作
                // 加锁是因为，可能有多个线程同时要将worker放入worker容器
                mainLock.lock();
                try {
        
    				// 获取到线程池状态rs
                    int rs = runStateOf(ctl.get());
                    if (rs < SHUTDOWN || // 如果线程池状态是 RUNNING
                        (rs == SHUTDOWN && firstTask == null)) {
         // 线程池是SHUTDOWN且要执行的任务为null
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        // 将当前woker加入到 HashSet<Worker> workers 中
                        workers.add(w); 
                        int s = workers.size(); // 获取到 workers 的大小，即现在有几个worker
                        // 如果worker数已经大于了最大线程池容量
                        if (s > largestPoolSize) 
                            largestPoolSize = s; // 将largestPoolSize设置为worker现在的书香
                        workerAdded = true; // 添加标志设置为成功
                    }
                } finally {
        
                    mainLock.unlock(); // 解锁
                }
    --------------------------------------------------------------------------------------------------------------            
                // 3.启动如果woker中的线程。前提是worker已经添加成功
                if (workerAdded) {
        
                    // 启动刚创建线程：Thread#start -> Worker#run -> runWorker
                    t.start();
                    workerStarted = true; // 线程启动标志置为true
                }
            }
        } finally {
        
        	// 如果线程启动失败
            if (! workerStarted) 
                addWorkerFailed(w);
        }
        // 返回线程是否启动成功
        return workerStarted;
    }

## 2.3 runWorker() ##

首先获取任务，然后让worker去执行任务。该方法大致逻辑如下：

1.  获取任务task，有两个途径
    
     *  firstTask：Worker的初始任务
     *  getTask()：任务队列的任务
2.  上锁，防止线程执行时被再丢入任务
3.  若线程池处于STOP，则中断当前线程
4.  执行 before钩子函数
5.  执行任务，即调用 task.run()
6.  执行 after 钩子函数
7.  删除当前任务，释放锁。while执行下一次任务

这里再注意一点，while 目的是维持当前线程持续执行任务，但线程如果迟迟拿不到 task（getTask方法中会阻塞等待）就会退出循环，即线程生命结束被回收。

final void runWorker(Worker w) {
        
        Thread wt = Thread.currentThread(); // 获取当前线程
        Runnable task = w.firstTask;  // 尝试获取创建worker时的firstTask
        
        w.firstTask = null; // 从这可以看出，只要firstTask执行过一次，就会一直被置为null
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
        
        	// 1.获取任务：如果firstTask已经被执行过了，就从任务队列中获取
        	// 注：通过while维持了线程的存活，并不断获取任务取执行。若迟迟拿不到任务，就会退出while结束线程
            while (task != null || (task = getTask()) != null) {
        
                // 2.锁住 worker，防止worker在执行任务时被丢入另一个任务
                w.lock();
                // 3.判断线程池若处于 stop 中,但线程没有到达中断状态，帮助线程中断
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
        
                    // 4.执行 before 钩子函数
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
        
                        // 5.同步执行任务
                        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 {
        
                        // 6.执行 after 钩子函数
                        // 如果这里抛出异常，会覆盖 catch 的异常，所以这里异常最好不要抛出来
                        afterExecute(task, thrown);
                    }
                } finally {
        
                    // 7.任务执行完成，删除任务，并计算解锁
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
        
            // 做一些抛出异常的善后工作
            processWorkerExit(w, completedAbruptly);
        }
    }

## 2.4 getTask() ##

从阻塞队列中获取任务，若阻塞等待后还没取到任务就会返回null，从而使当前线程在runWorker方法中退出while循环被回收，因为没事干了留着还浪费资源。具体回收策略在源码中，该方法大致流程如下：

1.  第一次判断回收当前线程：线程池SHUTDOWN，且队列空
2.  第二次判断回收当前线程，满足下列任一条件即可：
    
    1.  wc > maximumPoolSize && wc > 1：已有worker数超过线程池最大容量，且回收后线程池最少还有一个线程
    2.  wc > maximumPoolSize && workQueue.isEmpty()：已有worker数超过线程池最大容量，且任务队列为空
    3.  timed && timedOut && wc > 1：允许回收核心线程或已有线程数超过核心数，且当前线程已经超时，且回收后线程池最少还有一个线程
    4.  timed && timedOut && workQueue.isEmpty()：允许回收核心线程或已有线程数超过核心数，且当前线程已经超时，且任务队列为空
3.  从任务队列中获取任务（take或poll），若拿到了就返回，没拿到就将超时（timedOut）设置为true  
    注：只有timed为true才会使用poll然后等待KeepAliveTime时间，否则会使用take一直等待下去

这里再强调一次，核心线程与非核心线程只是概念上的区别，在代码中大家都一样，都是普通Thread。

private Runnable getTask() {
        
    	// 标识是否超时
    	// 默认false，但如果下面自旋中 poll 在 keepAliveTime（线程存活时间） 没等到任务，就会将timedOut置为true
        boolean timedOut = false; // Did the last poll() time out?
    
        for (;;) {
        
            int c = ctl.get();
            int rs = runStateOf(c); // 获取线程池状态
            
    		// 1.第一次判断是否回当前收线程
            // 线程池关闭 && 队列为空，不需要在运行了，直接返回null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
        
                decrementWorkerCount(); // workerCount--
                return null;
            }
    
            int wc = workerCountOf(c); // 获取worker个数
            // timed的作用是决定在阻塞队列中等任务时用 poll 还是 take
            // timed = 核心线程可以被灭亡（默认false） || 运行的线程数大于 coreSize 
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    	   
            // 2. 第二次判断是否回收当前线程。组合后分为4种情况
            if ((wc > maximumPoolSize || (timed && timedOut))  // woker大于线程池最大数量 || (timed && 当前线程已经超时)
                && (wc > 1 || workQueue.isEmpty())) {
         // woker大于1 || 任务队列为空
                // 通过CAS使workerCount--
                if (compareAndDecrementWorkerCount(c)) 
                    return null;
                continue;
            }
    
            try {
        
            	// 3.从阻塞队列中获取任务。timed决定了是使用 poll 还是 take
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : // poll，超时了就返回
                    workQueue.take(); // take，任务队列中没任务会阻塞等待
                // 如果在队列拿到了任务就返回
                if (r != null)
                    return r;
                // 没拿到就将超时timedOut设置为true，表示此时队列没有数据
                timedOut = true;
            } catch (InterruptedException retry) {
        
                timedOut = false;
            }
        }
    }

# 3.提炼总结 #

在文章的最后，我们再对 ThreadPoolExecutor 的关键信息做一些总结：

*  线程池解决两个问题：
    
     *  通过减少任务间的调度开销 (主要是通过线程池中的线程被重复使用的方式)，来提高大量任务时的执行性能
     *  提供了一种方式来管理线程和消费，维护基本数据统计等工作，比如统计已完成的任务数；
 *  线程池容量相关参数：
    
     *  coreSize：当新任务提交时，发现运行的线程数小于 coreSize，一个新的线程将被创建，即使这时候其它工作线程是空闲的，可以通过 getCorePoolSize 方法获得 coreSize
     *  maxSize: 当任务提交时，coreSize < 运行线程数 <= maxSize，但队列没有满时，任务提交到队列中，如果队列满了，在 maxSize 允许的范围内新建线程；
        
        一般来说，coreSize 和 maxSize 在线程池初始化时就已经设定了，但我们也可以通过 setCorePoolSize、setMaximumPoolSize 方法动态的修改这两个值；
 *  Keep-alive times 参数：
    
     *  作用： 如果当前线程池中有超过 coreSize 的线程，并且线程空闲的时间超过 keepAliveTime，当前线程就会被回收，这样可以避免线程没有被使用时的资源浪费；
     *  通过 setKeepAliveTime 方法可以动态的设置 keepAliveTime 的值；
     *  如果设置 allowCoreThreadTimeOut 为 ture 的话，core thread 空闲时间超过 keepAliveTime 的话，也会被回收；
 *  线程池新建时的队列选择有很多，比如：
    
     *  ArrayBlockingQueue，有界队列，可以防止资源被耗尽；
     *  LinkedBlockingQueue，无界队列，未消费的任务可以在队列中等待
     *  SynchronousQueue，为了避免任务被拒绝，要求线程池的 maxSize 无界，缺点是当任务提交的速度超过消费的速度时，可能出现无限制的线程增长
 *  拒绝策略：在 Executor 已经关闭或对最大线程和最大队列都使用饱和时，可以使用 RejectedExecutionHandler 类进行异常捕捉。有如下四种处理策略：
    
     *  AbortPolicy（默认）：抛出异常
     *  CallerRunsPolicy：不使用线程池，主线程来执行
     *  DiscardPolicy：直接丢弃任务
     *  DiscardOldestPolicy：丢弃队列中最老任务
 *  ExecutorService 使用线程池中的线程执行提交的任务，线程池我们可以使用 Executors 进行配置.Executors 为常用的场景设定了可直接初始化线程池的方法，比如:
    
     *  Executors\#newCachedThreadPool 无界的线程池，并且可以自动回收
     *  Executors\#newFixedThreadPool 固定大小线程池
     *  Executors\#newSingleThreadExecutor 单个线程的线程池；
 *  另外，线程池提供了很多可供扩展的钩子函数，比如有：
    
     *  提供在每个任务执行之前 beforeExecute 和执行之后 afterExecute 的钩子方法，主要用于操作执行环境，比如初始化 ThreadLocals、收集统计数据、添加日志条目等
     *  如果在执行器执行完成之后想干一些事情，可以实现 terminated 方法，如果钩子方法执行时发生异常，工作线程可能会失败并立即终止。

[JUC_ThreadPoolExecutor]: https://blog.csdn.net/weixin_43935927/article/details/113966340
[JUC_ThreadPoolExecutor 1]: https://blog.csdn.net/weixin_43935927/article/details/113965321
[JUC_ThreadPoolExecutor_Worker]: https://blog.csdn.net/weixin_43935927/article/details/113966417
[JUC_ThreadPoolExecutor_]: https://blog.csdn.net/weixin_43935927/article/details/113965324
[JUC_ThreadPoolExecutor 2]: https://blog.csdn.net/weixin_43935927/article/details/113965332
[20210223031044877.png_pic_center]: /images/20221123/9a4f636c66364e13847d834e0ead8a3d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/57e9c1a2c5be4ab489821066f3c63ec0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/9c6360c54b194b0a81b693b89c135391.png