线程池ThreadPoolExecutor源码剖析

刺骨的言语ヽ痛彻心扉 2023-10-08 18:01 95阅读 0赞

### 一、Java构建线程的方式 ###

*  继承Thread （也实现了Runnable）  
    ![请添加图片描述][4e19255ad7eb476ca44e9c3126203cfe.png]
 *  实现Runnable
 *  实现Callable （与Runnable区别…）
 *  线程池方式 （Java提供了构建线程池的方式）\[可以实现Runnable 和 Callable 功能\]
    
     *  Java提供了Executors可以去创建（规范中不允许使用这种方式创建线程池，这种方式对线程的控制粒度比较低）
     *  推荐手动创建线程池 ThreadPoolExecutor

### 二、线程池的7个参数 ###

public ThreadPoolExecutor(int corePoolSize,  // 核心线程数
                              int maximumPoolSize, // 最大线程数
                              long keepAliveTime, // 最大空闲时间
                              TimeUnit unit, // 时间单位
                              BlockingQueue<Runnable> workQueue, // 阻塞队列
                              ThreadFactory threadFactory, // 线程工厂
                              RejectedExecutionHandler handler) {
         // 拒绝策略
    }

### 三、线程池的执行流程 ###

<table> 
 <thead> 
  <tr> 
   <th align="center">线程池执行流程</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center"><img src="https://img-blog.csdnimg.cn/3769d0dfe1a140a99f341688d92f7ad4.png" alt="[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ygpetg5E-1676901244832)(C:\Users\b\AppData\Roaming\Typora\typora-user-images\image-20230220132010848.png)]"></td> 
  </tr> 
  <tr> 
   <td align="center"></td> 
  </tr> 
 </tbody> 
</table>

为什么要先进阻塞再去尝试创建非核心线程：

饭店（线程池）-- 厨子（线程）-- 人多先排队（阻塞队列）-- 招厨子I(创建最大线程数) – 今日客满（拒绝）

### 四、线程池属性标识 ###

##### 4.1 核心属性 #####

ThreadPoolExecutor 中的属性（核心成员变量）标识

学习课程 ： [https://www.bilibili.com/video/BV1244y1n7bz/?p=4&spm\_id\_from=pageDriver&vd\_source=c81fe4418bb0d2f341abd89cbfa157aahttps://www.bilibili.com/video/BV1244y1n7bz/?p=4&spm\_id\_from=pageDriver&vd\_source=c81fe4418bb0d2f341abd89cbfa157aa][https_www.bilibili.com_video_BV1244y1n7bz_p_4_spm_id_from_pageDriver_vd_source_c81fe4418bb0d2f341abd89cbfa157aahttps_www.bilibili.com_video_BV1244y1n7bz_p_4_spm_id_from_pageDriver_vd_source_c81fe4418bb0d2f341abd89cbfa157aa]

// 是一个int类型的数值，表达了两个意思，1：声明当前线程池的状态，2：声明线程池中的线程数
    // 高3位是：线程池状态  低29位是：线程池中的线程个数
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private static final int COUNT_BITS = Integer.SIZE - 3;  // 29,方便后面做位运算
    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; // 111 代表线程池为RUNNING, 代表正常接收任务
    private static final int SHUTDOWN   =  0 << COUNT_BITS; // 000 代表线程池为SHUTDOWN状态，不接收新任务，但是内部还会处理阻塞队列中的任务，正在进行的任务也正常处理
    private static final int STOP       =  1 << COUNT_BITS; // O01 代表线程池为STOP状态，不接收新任务，也不去处理阻塞队列中的任务，同时会中断正在执行的任务
    private static final int TIDYING    =  2 << COUNT_BITS; // 010 代表线程池为TIDYING状态，过渡的状态，代表当前线程池即将Game Over
    private static final int TERMINATED =  3 << COUNT_BITS; // O11 代表线程池为TERMINATED, 要执行terminated(), 真的凉凉了
    
    // Packing and unpacking ctl
    private static int runStateOf(int c)     {
         return c & ~CAPACITY; } // 得到线程池的状态
    private static int workerCountOf(int c)  {
         return c & CAPACITY; } // 得到当前线程池的线程数量
    private static int ctlOf(int rs, int wc) {
         return rs | wc; }
    
    // 想对位移更掌握，看雪花算法，达到手写的能力，位移方向和各种二进制运算就没问题了。

##### 4.2 线程池状态变化 #####

<table> 
 <thead> 
  <tr> 
   <th align="center">线程池状态变化</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">**<img src="https://img-blog.csdnimg.cn/7121853765eb422a80116df3da8c1a5a.png" alt="[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-i70mMtI0-1676901244832)(C:\Users\b\AppData\Roaming\Typora\typora-user-images\image-20230220145338221.png)]"></td> 
  </tr> 
  <tr> 
   <td align="center">**</td> 
  </tr> 
 </tbody> 
</table>

### 五、线程池的execute方法执行 ###

##### 5.1 从execute方法开始 #####

public void execute(Runnable command) {
        
        // 健壮性判断
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        // 拿到32位的int
        int c = ctl.get();
        // 获取 工作线程数 < 核心线程数
        if (workerCountOf(c) < corePoolSize) {
        
            // 进到if,代表可以创建 核心 线程数
            if (addWorker(command, true))
                // 到这结束
                return;
            // 如果if没进去，代表创建核心线程数失败，重新获取ct1
            c = ctl.get();
        }
        // 判断线程池是不是RUNNING,将任务添加到阻塞队列中
        if (isRunning(c) && workQueue.offer(command)) {
        
            // 再次获取ct1
            int recheck = ctl.get();
            // 再次判断是否是RUNNUING,  如果不是RUNNING,移除任务
            if (! isRunning(recheck) && remove(command))
                reject(command);  // 拒绝策略
            else if (workerCountOf(recheck) == 0) // 如果线程池处在RUNNING状态，BUT工作线程为0
                addWorker(null, false); // 阻塞队列有任务，但是没有工作线程，添加一个任务为空的工作线程处理阻塞队列中的任务
        }
        // 创建非核心线程，处理任务
        else if (!addWorker(command, false))
            reject(command); // 拒绝策略
    }

##### 5.2 通过上述源码，掌握了线程池的执行流程，再次查看addWorker方法内部做了什么处理 #####

private boolean addWorker(Runnable firstTask, boolean core) {
        
        retry:
        // 经过大量的判断，给工作线程数标识+1，
        for (;;) {
        
            // 获取ct1,(32位)
            int c = ctl.get();
            // 获取线程池状态
            int rs = runStateOf(c);
            
            // 除了RUNNING都有可能
            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())
                // rs == SHUTDOWN，如果不是SHUTDOWN,就代表是STOP或者跟高的状态，这时，不需要添加线程处理任务
    			// 任务为空，如果任务为null, 并且线程池状态不是RUNNING，不需要处理
    			// 阻塞队列不为nu11，如果阻塞队列为空，返回false,.外侧的！再次取反，获取true,不需要处理
               )
                // 构建工作线程失败！
                return false;
    
            for (;;) {
        
                // 获取工作线程个数
                int wc = workerCountOf(c);
                // 如果当前线程已经大于线程池最大容量，不去创建了
                // 判断wC是否超过核心线程或者最大线程
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    // 构建工作线程失败！
                    return false;
                
                //将工作线程数+1，采用CAS的方式
                if (compareAndIncrementWorkerCount(c))
                    // 成功就退出外侧for循环
                    break retry;
                
              	// 重新获取ct1
                c = ctl.get();  // Re-read ctl
                // 重新判断线程池状态，如果有变化   如果状态没变化，重新执行内部循环即可
                if (runStateOf(c) != rs)
                    // 结束这次外侧循环，开始下次外侧循环
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
    
        // worker开始 = false
        boolean workerStarted = false;
        // worker添加 = false
        boolean workerAdded = false;
        // Worker就是工作线程
        Worker w = null;
        try {
        
            // 创建Worker,传入任务
            w = new Worker(firstTask);
            // 从Worker中获取线程t
            final Thread t = w.thread;
            // 如果线程t不为nul1
            if (t != null) {
        
                // 获取线程池的全局锁，避免我添加任务时，其他线程干掉了线程池，干掉线程池需要先获取这个锁
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock(); // 加锁
                try {
        
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    
                    // 获取线程池状态
                    int rs = runStateOf(ctl.get());
                    // 是RUNNING状态
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
         // 是SHUTDOV状态，创建空任务工作线程，处理阻塞队列中的任务
                        // 线程是否是运行状态
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        // 将工作线程添加到集合中
                        workers.add(w);
                        // 获取工作线程个数
                        int s = workers.size();
                        // 如果线程工作线程数，大于之前记录的最大工作线程数，就替换一下
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        // workerAdded为true,添加工作线程成功
                        workerAdded = true;
                    }
                } finally {
        
                    // 释放锁
                    mainLock.unlock();
                }
                if (workerAdded) {
        
                    // 启动工作线程
                    t.start();
                    // 启动工作线程成功
                    workerStarted = true;
                }
            }
        } finally {
        
            // 如果启动工作线程失败，调用下面方法
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted; // 返回工作是否启动
    }

### 六、Worker的封装 ###

Worker的封装

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        
        /**
         * This class will never be serialized, but we provide a
         * serialVersionUID to suppress a javac warning.
         */
        private static final long serialVersionUID = 6138294804551838833L;
    
        /** 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;
    
        /**
         * Creates with given first task and thread from ThreadFactory.
         * @param firstTask the first task (null if none)
         */
        Worker(Runnable firstTask) {
        
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
    
        /** Delegates main run loop to outer runWorker  */
        public void run() {
        
            runWorker(this);
        }
    }

看runWorker方法

final void runWorker(Worker w) {
        
        // 获取当前线程
        Thread wt = Thread.currentThread();
        // 拿到任务
        Runnable task = w.firstTask;
        // 先不关注
        w.firstTask = null;
        w.unlock(); // allow interrupts
        
        // 标识为true
        boolean completedAbruptly = true;
        try {
        
            // 任务不空，执行任务。  如果任务为空，通过getTask从阻塞队列中获取任务！
            while (task != null || (task = getTask()) != null) {
        
                w.lock(); // 加锁，避免你shutdown我任务也不会中断
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                
                // 获取当前状态，是否大于等于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 {
        
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
        
            processWorkerExit(w, completedAbruptly);
        }
    }

有时间再去查看getTask方法。processWorkerExit线程执行完毕的后续处理。

[4e19255ad7eb476ca44e9c3126203cfe.png]: https://img-blog.csdnimg.cn/4e19255ad7eb476ca44e9c3126203cfe.png
[https_www.bilibili.com_video_BV1244y1n7bz_p_4_spm_id_from_pageDriver_vd_source_c81fe4418bb0d2f341abd89cbfa157aahttps_www.bilibili.com_video_BV1244y1n7bz_p_4_spm_id_from_pageDriver_vd_source_c81fe4418bb0d2f341abd89cbfa157aa]: https://www.bilibili.com/video/BV1244y1n7bz/?p=4&spm_id_from=pageDriver&vd_source=c81fe4418bb0d2f341abd89cbfa157aahttps://www.bilibili.com/video/BV1244y1n7bz/?p=4&spm_id_from=pageDriver&vd_source=c81fe4418bb0d2f341abd89cbfa157aa