【JUC源码】线程池：ThreadPoolExecutor（四）任务执行流程源码分析

╰半橙微兮° 2022-10-31 13:49 219阅读 0赞

> 线程池系列：
> 
>  *  [【JUC源码】线程池：ThreadPoolExecutor（一）继承关系分析][JUC_ThreadPoolExecutor]
>  *  [【JUC源码】线程池：ThreadPoolExecutor（二）底层结构分析][JUC_ThreadPoolExecutor 1]
>  *  [【JUC源码】线程池：ThreadPoolExecutor（三）Worker 设计思路及源码分析][JUC_ThreadPoolExecutor_Worker]
>  *  [【JUC源码】线程池：ThreadPoolExecutor（四） 任务执行流程源码分析][JUC_ThreadPoolExecutor_]
>  *  [【JUC源码】线程池：ThreadPoolExecutor（五）提炼总结][JUC_ThreadPoolExecutor 2]
>  *  [【JUC源码】线程池：创建线程池的参数设置思路&Excutors][JUC_Excutors]
>  *  [【JUC源码】线程池：关于 ThreadPool 的几个问题][JUC_ ThreadPool]

## 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.addWorker() ##

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

1.  线程池状态校验
    
     *  失败，返回 false。原因有二：
        
         *  线程池状态异常：SHUTDOWN，STOP，TIDYING，TERMINALED
         *  工作线程数溢出：线程数 >= 容量 或 在要使用 coreThread 的情况下，线程数 >= 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;
    }

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

## 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一直等待下去

> PS：这里再强调一次，核心线程与非核心线程只是概念上的区别，在代码中大家都一样，都是普通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
            	// keepAliveTime 是线程最大空闲时间，是构造线程池的入参
                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;
            }
        }
    }

## 小结 ##

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

[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
[JUC_Excutors]: https://blog.csdn.net/weixin_43935927/article/details/108889161
[JUC_ ThreadPool]: https://blog.csdn.net/weixin_43935927/article/details/108889102
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221024/92abc40b5f074bbea62ae0aa0bfc2150.png