并发编程-ThreadPoolExecutor 源码解析

短命女 2023-06-16 08:08 8阅读 0赞

#  #

**目录**

如何创建线程池：

ThreadPoolExecutor：

构造函数：

线程池的工作原理：

执行流程如下：

源码分析：

execute():

addWorker(Runnable firstTask,boolean core) ：

getTask():

--------------------

前言：

相比重复创建销毁线程来说，线程池有很多的优点：

1.          降低资源的消耗，可以通过重复利用已经创建的线程来执行任务
2.          提高线程的可管理性能，线程池提供统一分配，监控等

# 如何创建线程池： #

**推荐使用ThreadPoolExecutor来创建线程池**，不推荐使用Executor返回线程池对象的方式

因为使用ThreadPoolExecutor可以自定义的设置线程池中的参数，但是Executor中线程池对象参数都是固定的，不一定满足我们项目需求，而且可能会造成OOM溢出（为什么会溢出，后面会介绍）

# 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.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.workQueue = workQueue;
            this.keepAliveTime = unit.toNanos(keepAliveTime);
            this.threadFactory = threadFactory;
            this.handler = handler;
        }

**corePoolSize**

线程池中的核心线程数，当提交一个任务时，线程池创建一个新线程执行任务，直到当前线程数等于corePoolSize，即使其他空闲的线程也可以执行新任务；

如果当前线程数为corePoolSize，继续提交的任务被保存到阻塞队列中，等待被执行；

如果执行了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有核心线程，预加载的感觉。

**maximumPoolSize**

线程池中允许的最大线程数。如果当前阻塞队列满了，且继续提交任务，则创建新的线程执行任务，前提是当前线程数小于maximumPoolSize

**keepAliveTime**

线程空闲时的存活时间，即当线程没有任务执行时，继续存活的时间。默认情况下，该参数只在线程数大于corePoolSize时才有用

**workQueue**

workQueue必须是BlockingQueue阻塞队列。当线程池中的线程数超过它的corePoolSize的时候，线程会进入阻塞队列进行阻塞等待。通过workQueue，线程池实现了阻塞功能

**几种排队的策略：**

（1）不排队，直接提交(相当于传球手)

将任务直接交给线程处理而不保持它们，可使用SynchronousQueue

如果不存在可用于立即运行任务的线程（即线程池中的线程都在工作），则试图把任务加入缓冲队列将会失败，因此会构造一个新的线程来处理新添加的任务，并将其加入到线程池中（corePoolSize-->maximumPoolSize扩容）

Executors.newCachedThreadPool()采用的便是这种策略

（2）无界队列

可以使用LinkedBlockingQueue（基于链表的有界队列，FIFO），理论上是该队列可以对无限多的任务排队

将导致在所有corePoolSize线程都工作的情况下将新任务加入到队列中。这样，创建的线程就不会超过corePoolSize，也因此，maximumPoolSize的值也就无效了

（3）有界队列

可以使用ArrayBlockingQueue（基于数组结构的有界队列，FIFO），并指定队列的最大长度

使用有界队列可以防止资源耗尽，但也会造成超过队列大小和maximumPoolSize后，提交的任务被拒绝的问题，比较难调整和控制。

**threadFactory**

创建线程的工厂，通过自定义的线程工厂可以给每个新建的线程设置一个具有识别度的线程名

**RejectedExecutionHandler（饱和策略）**

线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，线程池提供了4种策略：

（1）AbortPolicy：直接抛出异常，默认策略；

（2）CallerRunsPolicy：用调用者所在的线程来执行任务；

（3）DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；

（4）DiscardPolicy：直接丢弃任务；

# 线程池的工作原理： #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70][]

### 执行流程如下：  ###

1.     当主线程调用线程池的execute，执行任务的时候首先会判断当前线程池中线程数是否小于corepoolSize(核心线程数)，如果小于则直接通过addwork（command，true）来创建核心worker线程，如果创建成功则返回true，否则继续后面的步骤 这个时候可能会有人问，线程数既然小于核心线程数，为什么会创建失败？

a. 因为是多线程工作，这个时候可能有某一个线程对线程池进行了shutdown，那么就无法接收外来的新的工作任务

b. workerCountOf(c) < corePoolSize 判断后，由于并发，别的线程先创建了worker线程，导致workerCount>=corePoolSize

2.     如果线程池还是Runing状态，我们就把当前的工作放入到阻塞队列中进行排队等待，如果添加成功就就行二次检测，否则继续下面的步骤，

什么是二次检测？ 为什么会有二次检测？

如果加入到队列之前线程池已经不是Running状态了就应该拒绝添加新任务，所以从队列中将任务删除。

3.     如果线程池无法入队列，说明阻塞队列已经满了，所以这个时候我们就可以尝试直接开启新的线程，扩容至maxPoolSize

# 源码分析：  #

### execute(): ###

public void execute(Runnable command) {
            //判断任务是否为空
            if (command == null)
                throw new NullPointerException();
            //获取线程池的状态
            int c = ctl.get();
            //用户获取低29位的线程数量和核心线程数做对比如果小于核心线程数
            if (workerCountOf(c) < corePoolSize) {
                //创建新线程执行任务是否成功
                if (addWorker(command, true))
                    //如果失败了就会往队列中放
                    return;
                //因为需要再次用c做判断所以需要重新获取ctl的状态值
                /**
                 * * 失败的原因可能是：
                 *          * 1、线程池已经shutdown，shutdown的线程池不再接收新任务
                 *          * 2、workerCountOf(c) < corePoolSize 判断后，
                 *          由于并发，别的线程先创建了worker线程，导致失败
                 */
                c = ctl.get();
            }
            //线程池处于running状态，并且成功把任务放入workQueue阻塞队列中
            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);
        }

流程图如下：

![image][]

###  addWorker(Runnable firstTask,boolean core) ： ###

firstTask: 加入到线程池中要执行的任务，如果为null说明是要创建一个空的线程

core: true 创建核心线程， false，创建非核心线程

**ddWorker方法有4种传参的方式：**

1、addWorker(command, true)

2、addWorker(command, false)

3、addWorker(null, false)

4、addWorker(null, true)

在execute方法中就使用了前3种，结合这个核心方法进行以下分析  
    第一个：线程数小于corePoolSize时，放一个需要处理的task进Workers Set。如果Workers Set长度超过corePoolSize，就返回false  
    第二个：当队列被放满时，就尝试将这个新来的task直接放入Workers Set，而此时Workers Set的长度限制是maximumPoolSize。如果线程池也满了的话就返回false  
    第三个：放入一个空的task进workers Set，长度限制是maximumPoolSize。这样一个task为空的worker在线程执行的时候会去任务队列里拿任务，这样就相当于创建了一个新的线程，只是没有马上分配任务  
    第四个：这个方法就是放一个null的task进Workers Set，而且是在小于corePoolSize时，如果此时Set中的数量已经达到corePoolSize那就返回false，什么也不干。实际使用中是在prestartAllCoreThreads()方法，这个方法用来为线程池预先启动corePoolSize个worker等待从workQueue中获取任务执行

private boolean addWorker(Runnable firstTask, boolean core) {
        //外层循环，负责判断线程池状态
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c); //状态
     
            // Check if queue empty only if necessary.
            /**
             * 线程池的state越小越是运行状态，runnbale=-1，shutdown=0,stop=1,tidying=2，terminated=3
             * 1、如果线程池state已经至少是shutdown状态了
             * 2、并且以下3个条件任意一个是false
             *   rs == SHUTDOWN         （隐含：rs>=SHUTDOWN）false情况： 线程池状态已经超过shutdown，可能是stop、tidying、terminated其中一个，即线程池已经终止
             *   firstTask == null      （隐含：rs==SHUTDOWN）false情况： firstTask不为空，rs==SHUTDOWN 且 firstTask不为空，return false，场景是在线程池已经shutdown后，还要添加新的任务，拒绝
             *   ! workQueue.isEmpty()  （隐含：rs==SHUTDOWN，firstTask==null）false情况： workQueue为空，当firstTask为空时是为了创建一个没有任务的线程，再从workQueue中获取任务，如果workQueue已经为空，那么就没有添加新worker线程的必要了
             * return false，即无法addWorker()
             */
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
     
            //内层循环，负责worker数量+1
            for (;;) {
                int wc = workerCountOf(c); //worker数量
                 
                //如果worker数量>线程池最大上限CAPACITY（即使用int低29位可以容纳的最大值）
                //或者( worker数量>corePoolSize 或  worker数量>maximumPoolSize )，即已经超过了给定的边界
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                 
                //调用unsafe CAS操作，使得worker数量+1，成功则跳出retry循环
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                 
                //CAS worker数量+1失败，再次读取ctl
                c = ctl.get();  // Re-read ctl
                 
                //如果状态不等于之前获取的state，跳出内层循环，继续去外层循环判断
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
                // else CAS失败时因为workerCount改变了，继续内层循环尝试CAS对worker数量+1
            }
        }
     
        /**
         * worker数量+1成功的后续操作
         * 添加到workers Set集合，并启动worker线程
         */
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            final ReentrantLock mainLock = this.mainLock; 
            w = new Worker(firstTask); //1、设置worker这个AQS锁的同步状态state=-1
                                       //2、将firstTask设置给worker的成员变量firstTask
                                       //3、使用worker自身这个runnable，调用ThreadFactory创建一个线程，并设置给worker的成员变量thread
            final Thread t = w.thread;
            if (t != null) {
                mainLock.lock();
                try {
                    //--------------------------------------------这部分代码是上锁的
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    // 当获取到锁后，再次检查
                    int c = ctl.get();
                    int rs = runStateOf(c);
     
                    //如果线程池在运行running<shutdown 或者 线程池已经shutdown，且firstTask==null（可能是workQueue中仍有未执行完成的任务，创建没有初始任务的worker线程执行）
                    //worker数量-1的操作在addWorkerFailed()
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable   线程已经启动，抛非法线程状态异常
                            throw new IllegalThreadStateException();
                         
                        workers.add(w);//workers是一个HashSet<Worker>
                         
                        //设置最大的池大小largestPoolSize，workerAdded设置为true
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                  //--------------------------------------------
                } 
                finally {
                    mainLock.unlock();
                }
                 
                //如果往HashSet中添加worker成功，启动线程
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            //如果启动线程失败
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

流程图：

![image][]

执行到这里读者可能会有疑惑，既然addworker是创建线程，那么什么时候来执行传递进来的task任务呢？  好像没有地方表现出线程去拿任务。

仔细看addworker里面的代码：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 1][]

是不是突然恍然大悟？ 在Worker中有一个线程，并且在addworker成功之后会去 start 线程，那么就会运行runwoker。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 2][]

### getTask(): ###

private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?
     
        /**
         * 外层循环
         * 用于判断线程池状态
         */
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
     
            // Check if queue empty only if necessary.
            /**
             * 对线程池状态的判断，两种情况会workerCount-1，并且返回null
             * 线程池状态为shutdown，且workQueue为空（反映了shutdown状态的线程池还是要执行workQueue中剩余的任务的）
             * 线程池状态为stop（shutdownNow()会导致变成STOP）（此时不用考虑workQueue的情况）
             */
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount(); //循环的CAS减少worker数量，直到成功
                return null;
            }
     
            boolean timed;      // Are workers subject to culling?
                                // 是否需要定时从workQueue中获取
             
            /**
             * 内层循环
             * 要么break去workQueue获取任务
             * 要么超时了，worker count-1
             */
            for (;;) {
                int wc = workerCountOf(c);
                timed = allowCoreThreadTimeOut || wc > corePoolSize; //allowCoreThreadTimeOut默认为false
                                                                     //如果allowCoreThreadTimeOut为true，说明corePoolSize和maximum都需要定时
                 
                //如果当前执行线程数<maximumPoolSize，并且timedOut 和 timed 任一为false，跳出循环，开始从workQueue获取任务
                if (wc <= maximumPoolSize && ! (timedOut && timed))
                    break;
                 
                /**
                 * 如果到了这一步，说明要么线程数量超过了maximumPoolSize（可能maximumPoolSize被修改了）
                 * 要么既需要计时timed==true，也超时了timedOut==true
                 * worker数量-1，减一执行一次就行了，然后返回null，在runWorker()中会有逻辑减少worker线程
                 * 如果本次减一失败，继续内层循环再次尝试减一
                 */
                if (compareAndDecrementWorkerCount(c))
                    return null;
                 
                //如果减数量失败，再次读取ctl
                c = ctl.get();  // Re-read ctl
                 
                //如果线程池运行状态发生变化，继续外层循环
                //如果状态没变，继续内层循环
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
     
            try {
                //poll() - 使用  LockSupport.parkNanos(this, nanosTimeout) 挂起一段时间，interrupt()时不会抛异常，但会有中断响应
                //take() - 使用 LockSupport.park(this) 挂起，interrupt()时不会抛异常，但会有中断响应
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :    //大于corePoolSize
                    workQueue.take();                                        //小于等于corePoolSize
                 
                //如获取到了任务就返回
                if (r != null)
                    return r;
                 
                //没有返回，说明超时，那么在下一次内层循环时会进入worker count减一的步骤
                timedOut = true;
            } 
            /**
                  * blockingQueue的take()阻塞使用LockSupport.park(this)进入wait状态的，对LockSupport.park(this)进行interrupt不会抛异常，但还是会有中断响应
                  * 但AQS的ConditionObject的await()对中断状态做了判断，会报告中断状态 reportInterruptAfterWait(interruptMode)
                  * 就会上抛InterruptedException，在此处捕获，重新开始循环
                  * 如果是由于shutdown()等操作导致的空闲worker中断响应，在外层循环判断状态时，可能return null
                  */
            catch (InterruptedException retry) { 
                timedOut = false; //响应中断，重新开始，中断状态会被清除
            }
        }
    }

**执行流程：**

1、首先判断是否可以满足从workQueue中获取任务的条件，不满足return null

A、线程池状态是否满足：

（a）shutdown状态 + workQueue为空 或 stop状态，都不满足，因为被shutdown后还是要执行workQueue剩余的任务，但workQueue也为空，就可以退出了

（b）stop状态，shutdownNow()操作会使线程池进入stop，此时不接受新任务，中断正在执行的任务，workQueue中的任务也不执行了，故return null返回

B、线程数量是否超过maximumPoolSize 或 获取任务是否超时

（a）线程数量超过maximumPoolSize可能是线程池在运行时被调用了setMaximumPoolSize()被改变了大小，否则已经addWorker()成功不会超过maximumPoolSize

（b）如果 当前线程数量>corePoolSize，才会检查是否获取任务超时，这也体现了当线程数量达到maximumPoolSize后，如果一直没有新任务，会逐渐终止worker线程直到corePoolSize

2、如果满足获取任务条件，根据是否需要定时获取调用不同方法：

A、workQueue.poll()：如果在keepAliveTime时间内，阻塞队列还是没有任务，返回null

B、workQueue.take()：如果阻塞队列为空，当前线程会被挂起等待；当队列中有任务加入时，线程被唤醒，take方法返回任务

整体的时序图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 3][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210713085733433.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz,size_16,color_FFFFFF,t_70
[image]: https://img-blog.csdnimg.cn/img_convert/40a29d738d3b3be291f91b3f4f0040aa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210713093849334.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/2021071309411435.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20210713094646895.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz,size_16,color_FFFFFF,t_70