登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 安全 ›JAVA JUC干货之线程池实现原理和源码详解（下） ...

JAVA JUC干货之线程池实现原理和源码详解（下）

寨重 2025-5-30 13:54:47
目录

  • 线程池设计思路
  • 线程池实现原理
  • ThreadPoolExecutor 示例
  • ThreadPoolExecutor 源码详解

    • execute提交任务源码分析
    • addWorker创建线程源码详解
    • runWorker执行任务源码详解
    • processWorkerExit回收线程源码详解
    • getTask获取任务源码详解
    • tryTerminate终止线程源码详解

  • 创建多少线程比较合适
  • 初始化线程池
  • 结束语
  • Reference

摘要：分享JAVA JUC线程池干货，首先描述线程池的基本概念，然后介绍线程工厂和拒绝策略，其次逐步深入线程池实现原理和线程池状态机，最后结合实战讲解源码。
JUC干货系列目录：

  • JAVA JUC干货之线程池状态和状态切换
  • JAVA JUC干货之线程池实现原理和源码详解（上）
  • JAVA JUC干货之线程池实现原理和源码详解（下）
  我在《JAVA JUC干货之线程池实现原理和源码详解（上）》中介绍了线程池的基本功能，本文分享线程池实现原理，并结合案例详解线程池源码。
线程池设计思路

  本节主要参考文章《Java 线程池详解，图文并茂，还有谁不会》，感兴趣的读者可以去看看。有句话叫做艺术来源于生活，编程语言也是如此，很多设计思想能映射到日常生活中，比如封装、继承和抽象等等。今天我们要说的线程池，它同样可以在现实世界找到对应的实体——工厂。先假想一个工厂的生产流程：
   
1.png
   
    工厂生产流程  假如有一个工厂，工厂中有固定的一批工人，称为正式员工，每个正式员工同时只能做一件任务，由这些正式员工完成工厂接收的订单。随着新任务增长的速度远远大于工人做任务的速度，必定出现正式员工忙不过来的场景，工厂会将生产原料暂时堆积在仓库中，等有空闲的正式员工时再处理。因为正式员工清闲自在，空闲了也不会主动处理仓库中的生产任务，所以需要调度员实时调度。仓库堆积满了后，订单还在增加怎么办？
  工厂只能临时扩招一批工人来应对生产高峰，而这批工人在高峰结束后，由于新任务增长的速度放缓慢了，是要辞掉的，所以称为临时工。当临时工也以招满后（受限于工位数量限制导致临时工数量也有上限），后面的订单只能忍痛拒绝了。我们做如下一番映射：
工厂线程池订单任务正式员工核心线程临时工非核心线程仓库阻塞队列工位数量最大线程数调度员getTask()，将任务队列中的任务调度给空闲线程  映射后，形成如下线程池流程图，两者是不是有异曲同工之妙？
   
2.png
   
    工厂-线程池流程映射图  这样，线程池的工作原理或者说流程就很好理解了，在下一章详细介绍。
线程池实现原理

  线程池的执行原理是通过循环地从任务队列中取出任务，然后将任务分配给空闲的工作线程执行。当任务队列为空时，线程池会进入等待状态，直到有新的任务到来。线程池还提供设置线程数和任务队列长度的能力，以控制并发线程的数量。
  下面我们进入正题，看一下在线程池中一个任务从提交到最终执行完毕经历了哪些过程。
  敲黑板划重点，用一句话简述ThreadPoolExecutor线程池实现原理：首先创建核心线程，再把任务放入阻塞队列，其次创建非核心线程，最后抛弃任务。详细流程图如下：
   
3.png
   
    线程池实现原理  从执行流程可知，属性判断顺序如下：corePoolSize -> workQueue -> maxinumPoolSize。线程池执行流程归纳为如下：
1.预热核心线程：提交新任务后，如果线程数没有达到核心线程数 corePoolSize，则创建核心线程执行新任务，而且空闲的核心线程处于阻塞状态不执行新任务直到队列中有任务。即线程池优先填满corePoolSize个核心线程，再复用核心线程处理任务。如果线程数达到核心线程数，则进入下个流程。
2.任务入队：如果线程数达到核心线程数，且任务队列未满，则把新任务放入阻塞队列末尾；这个场景下，核心线程执行完当前任务后自动从任务队列中获取任务来执行。如果任务队列已满，则进入下个流程。
3.创建非核心线程：如果任务队列已满而且最大线程数 maximumPoolSize > 线程数poolSize > corePoolSize，则创建新的普通线程并立刻执行当前的新任务；温馨提示，这个场景下新任务早于队列中的任务执行。如果线程数达到最大线程数，则进入执行拒绝任务的流程。
非核心线程数为 maximumPoolSize - corePoolSize，即为最大线程数减去核心线程数。
4.拒绝任务：如果线程数等于最大线程数且任务队列已满，则根据拒绝策略处理新任务。
5.复用线程：线程执行完任务后去检查任务队列里是否有任务需要执行，若有则马上执行；否则，进入阻塞状态。
6.销毁线程：普通线程如果无事可做的时间超过keepAliveTime，那么就会被回收，以便减少内存占用和资源消耗，实现对系统资源的调优。如果调用了allowCoreThreadTimeOut(true)方法，则会给核心线程数设置存活时间，使得超过keepAliveTime的核心线程也会被销毁，从而最终有可能导致线程池中的线程数为0。
  通常情况下，随着阻塞队列中任务数的减少，线程数最终会收缩到 corePoolSize 的大小，故线程数维持在corePoolSize和maximumPoolSize之间。
  这里有个细节值得大家品味，在线程池里的线程数小于核心线程数的前提下提交任务的时候，虽然核心线程因任务队列空空如也而可能处于闲置状态，但是依然会继续创建核心线程，而非复用已有的、空闲的核心线程。
ThreadPoolExecutor 示例

  下面这个使用execute提交任务的示例用于演示线程池执行流程：
  1. import com.google.common.util.concurrent.ThreadFactoryBuilder;
  3. import java.util.concurrent.*;
  4. import java.util.concurrent.atomic.AtomicInteger;
  6. /**
  7. * @Author 楼兰胡杨
  8. * @Date 2025-05-16
  9. * @Description: 通过调整for循环次数验证线程池执行机制
  10. */
  11. public class ThreadPoolTest {
  12.     public static ThreadPoolExecutor executor;
  14.     public static void main(String[] args) {
  15.         ThreadFactory guavaFactory = new ThreadFactoryBuilder().setNameFormat("pool-Wiener-%d").build();
  16.         // 自定义线程工厂
  17.         ThreadFactory jucFactory = new ThreadFactory() {
  18.             private final AtomicInteger mThreadNum = new AtomicInteger(1);
  20.             @Override
  21.             public Thread newThread(Runnable r) {
  22.                 return new Thread(r, "Wiener-" + mThreadNum.getAndIncrement());
  23.             }
  24.         };
  25.         RejectionImpl rejection = new RejectionImpl();
  26.         executor = new ThreadPoolExecutor(2, 5, 200, TimeUnit.MILLISECONDS,
  27.                 new ArrayBlockingQueue<>(10), guavaFactory, rejection);
  28.         for (int i = 0; i < 25; i++) {
  29.             MyPoolTask myTask = new MyPoolTask(i);
  30.             executor.execute(myTask);
  32.         }
  33.         executor.shutdown();
  34.     }
  36. }
  38. class MyPoolTask implements Runnable {
  39.     private int taskNum;
  40.     ThreadPoolExecutor myExe = ThreadPoolTest.executor;
  42.     public MyPoolTask(int num) {
  43.         this.taskNum = num;
  44.     }
  46.     @Override
  47.     public void run() {
  48. //        System.out.println("正在执行task " + taskNum);
  49.         System.out.println(Thread.currentThread().getName() + "，活跃线程数目：" + myExe.getPoolSize()
  50.                 + "，队列长度：" +
  51.                 myExe.getQueue().size()
  52.                 + "，当前任务是" + taskNum + "，已执行任务数目：" + myExe.getCompletedTaskCount());
  53.         try {
  54.             Thread.currentThread().sleep(300);
  55.         } catch (InterruptedException e) {
  56.             e.printStackTrace();
  57.         }
  58.         System.out.println("执行完毕task " + taskNum);
  59.     }
  61.     @Override
  62.     public String toString() {
  63.         return taskNum + "";
  64.     }
  65. }
  67. class RejectionImpl implements RejectedExecutionHandler {
  68.     @Override
  69.     public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
  70.         System.out.println("被拒绝任务是 " + r.toString());
  71.     }
  72. }
复制代码
  执行结果如下：
  1. pool-Wiener-1，活跃线程数目：3，队列长度：10，当前任务是1，已执行任务数目：0
  2. pool-Wiener-0，活跃线程数目：2，队列长度：9，当前任务是0，已执行任务数目：0
  3. pool-Wiener-2，活跃线程数目：3，队列长度：10，当前任务是12，已执行任务数目：0
  4. pool-Wiener-4，活跃线程数目：5，队列长度：10，当前任务是14，已执行任务数目：0
  5. pool-Wiener-3，活跃线程数目：5，队列长度：10，当前任务是13，已执行任务数目：0
  6. 被拒绝任务是 15
  7. 被拒绝任务是 16
  8. 被拒绝任务是 17
  9. 被拒绝任务是 18
  10. 被拒绝任务是 19
  11. 被拒绝任务是 20
  12. 被拒绝任务是 21
  13. 被拒绝任务是 22
  14. 被拒绝任务是 23
  15. 被拒绝任务是 24
  16. 执行完毕task 0
  17. 执行完毕task 14
  18. pool-Wiener-0，活跃线程数目：5，队列长度：9，当前任务是2，已执行任务数目：1
  19. 执行完毕task 1
  20. pool-Wiener-4，活跃线程数目：5，队列长度：8，当前任务是3，已执行任务数目：2
  21. 执行完毕task 12
  22. pool-Wiener-2，活跃线程数目：5，队列长度：6，当前任务是5，已执行任务数目：4
  23. pool-Wiener-1，活跃线程数目：5，队列长度：7，当前任务是4，已执行任务数目：3
  24. 执行完毕task 13
  25. pool-Wiener-3，活跃线程数目：5，队列长度：5，当前任务是6，已执行任务数目：5
  26. 执行完毕task 3
  27. pool-Wiener-4，活跃线程数目：5，队列长度：4，当前任务是7，已执行任务数目：6
  28. 执行完毕task 2
  29. 执行完毕task 4
  30. pool-Wiener-0，活跃线程数目：5，队列长度：3，当前任务是8，已执行任务数目：7
  31. 执行完毕task 6
  32. pool-Wiener-1，活跃线程数目：5，队列长度：2，当前任务是9，已执行任务数目：8
  33. 执行完毕task 5
  34. pool-Wiener-3，活跃线程数目：5，队列长度：1，当前任务是10，已执行任务数目：9
  35. pool-Wiener-2，活跃线程数目：5，队列长度：0，当前任务是11，已执行任务数目：10
  36. 执行完毕task 9
  37. 执行完毕task 8
  38. 执行完毕task 10
  39. 执行完毕task 11
  40. 执行完毕task 7
复制代码
  从执行结果日志可以看出，当线程池中线程的数目大于5时，便将任务放入任务缓存队列里面，当任务缓存队列满了之后，便创建新的线程。如果上面程序for循环中，把创建的任务数从25个降低到15个，就不会抛出任务被拒绝的异常了。
ThreadPoolExecutor 源码详解

  程序猿为什么需要读源码？阅读源码是程序猿成长过程中非常重要的一个环节。这就像学习语言时需要阅读经典文学作品一样，通过阅读源码，程序员可以学到很多优秀的编程技巧和最佳实践；可以了解多样化的技术选型方案，培养架构设计敏感度，为技术决策提供参考依据。具体来说，读源码有以下几个好处：
  深入了解工具库或框架：当你使用第三方库或者框架时，深入其内部结构有助于更好地掌握它们的功能以及局限性，从而更高效地运用到自己的项目中去。
  增强业务能力和团队协作能力：阅读和理解其他同事敲的代码可以使人深度认识项目，精准了解业务逻辑，还能促进成员间的交流与合作。
  面试准备：在求职过程中，了解常见开源框架的源码实现可以帮助你在面试环节中脱颖而出，因为许多公司会考察候选人对这些技术细节的理解程度。
  拓展技术视野：接触不同项目的源码（如微服务框架、分布式系统）可了解多样化的技术选型方案，培养架构设计敏感度，为技术决策提供参考依据。
  总之，对于任何有志于提升自身技术水平的程序员而言，定期花时间去研究高质量的源码是一项不可或缺的学习活动。你对哪些源码感兴趣？可以评论区留言，一起品尝。下面浅谈我对 ThreadPoolExecutor 源码的理解，不当之处还请在评论区留言，一起翻越ThreadPoolExecutor这座山。
execute提交任务源码分析

  在ThreadPoolExecutor类中，最核心的任务提交方法是execute(Runnable command)方法，虽然通过submit也可以提交任务，但是submit方法底层实现中最终调用的还是execute，所以我们只需要分析execute的源码即可。在JDK 21中，execute(Runnable command)的源代码如下所示，已经添加中文注释，同时保留了原汁原味的英文注释：
  1.     /**
  2.      * Executes the given task sometime in the future.  The task
  3.      * may execute in a new thread or in an existing pooled thread.
  4.      *
  5.      * If the task cannot be submitted for execution, either because this
  6.      * executor has been shutdown or because its capacity has been reached,
  7.      * the task is handled by the current {@link RejectedExecutionHandler}.
  8.      *
  9.      * @param command the task to execute
  10.      * @throws RejectedExecutionException at discretion of
  11.      *         {@code RejectedExecutionHandler}, if the task
  12.      *         cannot be accepted for execution
  13.      * @throws NullPointerException if {@code command} is null
  14.      */
  15.     public void execute(Runnable command) {
  16.         // 判断提交的任务是否为null
  17.         if (command == null)
  18.             throw new NullPointerException();
  19.         /*
  20.          * Proceed in 3 steps（执行过程可以分为如下三个步骤）:
  21.          *
  22.          * 1. If fewer than corePoolSize threads are running, try to
  23.          * start a new thread with the given command as its first
  24.          * task.  The call to addWorker atomically checks runState and
  25.          * workerCount, and so prevents false alarms that would add
  26.          * threads when it shouldn't, by returning false.
  27.          *
  28.          * 2. If a task can be successfully queued, then we still need
  29.          * to double-check whether we should have added a thread
  30.          * (because existing ones died since last checking) or that
  31.          * the pool shut down since entry into this method. So we
  32.          * recheck state and if necessary roll back the enqueuing if
  33.          * stopped, or start a new thread if there are none.
  34.          *
  35.          * 3. If we cannot queue task, then we try to add a new
  36.          * thread.  If it fails, we know we are shut down or saturated
  37.          * and so reject the task.
  38.          */
  39.         // 可以将 ctl 理解成保存了线程数和运行状态等信息的变量
  40.         int c = ctl.get();
  41.         // 判断工作线程总数是否小于核心线程数，小于的时候可以添加核心线程，并且将当前任务作为此线程执行的第一个任务
  42.         if (workerCountOf(c) < corePoolSize) {
  43.            //新增的核心线程成功执行任务，流程结束；addWorker第二个参数为true，表示根据核心线程数判断线程数量
  44.             if (addWorker(command, true))
  45.                 return;
  46.           // addWorker执行报错，需要再次检查变量值
  47.             c = ctl.get();
  48.         }
  49.         //判断线程池的状态是否正常而且任务放入任务队列是否正常
  50.         if (isRunning(c) && workQueue.offer(command)) {
  51.            // 同理，为了防止向任务队列中又提交新的任务造成错误，再次更新变量值
  52.             int recheck = ctl.get();
  53.             if (! isRunning(recheck) && remove(command))
  54.                //当前线程池处于非运行状态且将刚添加的任务成功从任务队列移除，执行拒绝策略
  55.                 reject(command);
  56.             else if (workerCountOf(recheck) == 0)
  57.                // 线程数为0，第一个参数为null，表示创建线程但不启动
  58.                 addWorker(null, false);
  59.        //线程数超过核心线程数而且队列已满，需要新增非核心线程
  60.         } else if (!addWorker(command, false))
  61.             //工作线程已经达到了最大线程数阈值，或者是调用了关闭线程池的方法，触发拒绝策略
  62.             reject(command);
  63.     }
复制代码
  假设线程数达到核心线程数且队列未满，这时新提交的任务可以直接复用空闲的核心线程吗？从源码来看，新提交的任务需要放入任务队列，空闲的核心线程是从任务队列拿任务。
  线程池的本质是对任务和线程的管理，而做到这一点最关键的思想就是借助生产者消费者模式将任务和线程两者解耦，不让两者直接绑定，方便做后续的工作分配。在队列为空时，线程会等待任务进入队列；当队列不满时，任务会等待线程来执行。
  从上述源码不难发现，execute 方法多次通过 addWorker 方法来添加线程处理任务，故我们下面来阅读 addWorker 方法的源码。
addWorker创建线程源码详解
  1.     /**
  2.      * Set containing all worker threads in pool. 线程池中存放线程的集合，维护一组Worker对象
  3.      * Accessed only when holding mainLock.
  4.      */
  5.     private final HashSet<Worker> workers = new HashSet<>();
  6.    
  7.     /*
  8.      * Methods for creating, running and cleaning up after workers
  9.      */
  11.     /**
  12.      * Checks if a new worker can be added with respect to current
  13.      * pool state and the given bound (either core or maximum). If so,
  14.      * the worker count is adjusted accordingly, and, if possible, a
  15.      * new worker is created and started, running firstTask as its
  16.      * first task. This method returns false if the pool is stopped or
  17.      * eligible to shut down. It also returns false if the thread
  18.      * factory fails to create a thread when asked.  If the thread
  19.      * creation fails, either due to the thread factory returning
  20.      * null, or due to an exception (typically OutOfMemoryError in
  21.      * Thread.start()), we roll back cleanly.
  22.      *
  23.      * @param firstTask the task the new thread should run first (or
  24.      * null if none). Workers are created with an initial first task
  25.      * (in method execute()) to bypass queuing when there are fewer
  26.      * than corePoolSize threads (in which case we always start one),
  27.      * or when the queue is full (in which case we must bypass queue).
  28.      * Initially idle threads are usually created via
  29.      * prestartCoreThread or to replace other dying workers.
  30.      *
  31.      * @param core if true use corePoolSize as bound, else
  32.      * maximumPoolSize. (A boolean indicator is used here rather than a
  33.      * value to ensure reads of fresh values after checking other pool
  34.      * state).
  35.      * @return true if successful
  36.      */
  37.     private boolean addWorker(Runnable firstTask, boolean core) {
  38.         // 外层循环：判断线程池状态
  39.         retry:
  40.         for (int c = ctl.get();;) {
  41.             // Check if queue empty only if necessary.
  42.             if (runStateAtLeast(c, SHUTDOWN)
  43.                 && (runStateAtLeast(c, STOP)
  44.                     || firstTask != null
  45.                     || workQueue.isEmpty()))
  46.                 // 线程池已经被停止或者关闭，或者队列为空，终止流程，返回 false
  47.                 return false;
  48.             // 死循环
  49.             for (;;) {
  50.                 // 检查当前工作线程数是否已经达到核心线程数（core 为 true）或最大线程数（core 为 false）
  51.                 if (workerCountOf(c)
  52.                     >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
  53.                     // 达到核心线程数或最大线程数，返回 false
  54.                     return false;
  55.                  //通过cas操作增加线程池的工作线程数量
  56.                 if (compareAndIncrementWorkerCount(c))
  57.                     // 成功原子地增加工作线程数，跳出外层的 for 循环
  58.                     break retry;
  59.                 // 增加线程数量失败，再次读取 ctl 的值
  60.                 c = ctl.get();  // Re-read ctl
  61.                 // 如果当前运行状态不等于
  62.                 if (runStateAtLeast(c, SHUTDOWN))
  63.                     // 线程池处于关闭状态，跳到外层循环重新执行
  64.                     continue retry;
  65.                 // else CAS failed due to workerCount change; retry inner loop
  66.             }
  67.         }
  68.         /**
  69.          * 线程数量+1成功的后续操作：添加新工作线程到工作线程集合，启动工作线程执行firstTask
  70.          */
  71.         boolean workerStarted = false;
  72.         boolean workerAdded = false;
  73.         Worker w = null;
  74.         try {
  75.             //添加新的Worker对象来处理任务firstTask
  76.             w = new Worker(firstTask);
  77.             //获取 Worker 对象中的线程。Worker类是任务线程的包装类，内部封装了一个Thread类型的变量
  78.             final Thread t = w.thread;
  79.             if (t != null) {
  80.                 // 开始加内置锁
  81.                 final ReentrantLock mainLock = this.mainLock;
  82.                 // 基于ReentrantLock的同步块
  83.                 mainLock.lock();
  84.                 try {
  85.                     // Recheck while holding lock.
  86.                     // Back out on ThreadFactory failure or if
  87.                     // shut down before lock acquired.
  88.                     // 再次检查状态，保证线程工厂没有失败或者在获取锁之前线程池没有被关闭
  89.                     int c = ctl.get();
  90.                     //添加线程的前提条件：①线程处于运行状态，②线程处于shutdown状态并且firstTask为null
  91.                     if (isRunning(c) ||
  92.                         (runStateLessThan(c, STOP) && firstTask == null)) {
  93.                         // 检查线程状态，如果不是新建状态就抛异常
  94.                         if (t.getState() != Thread.State.NEW)
  95.                             throw new IllegalThreadStateException();
  96.                         //workers是一个HashSet<Woker> 集合，往里面新增Worker类型的工作线程对象
  97.                         workers.add(w);
  98.                         // 把工作线程添加标志置为 true
  99.                         workerAdded = true;
  100.                         int s = workers.size();
  101.                         // largestPoolSize 表示线程池中出现过的最大线程数
  102.                         if (s > largestPoolSize)
  103.                             // 更新线程池中线程数最大值
  104.                             largestPoolSize = s;
  105.                     }
  106.                 } finally {
  107.                     // 释放锁
  108.                     mainLock.unlock();
  109.                 }
  110.                 if (workerAdded) {
  111.                     // 启动新创建的线程，执行当前任务firstTask
  112.                     container.start(t);
  113.                     // 把线程启动标志置为 true
  114.                     workerStarted = true;
  115.                 }
  116.             }
  117.         } finally {
  118.             //判断线程有没有启动成功，如果没有则调用addWorkerFailed方法
  119.             if (! workerStarted)
  120.                 addWorkerFailed(w);
  121.         }
  122.         // 返回线程启动标志
  123.         return workerStarted;
  124.     }
复制代码
  我们来看看函数addWorker(Runnable firstTask, boolean core) 的入参。firstTask：添加的新线程需要执行的第一任务，执行完成之后才能执行任务队列中其它任务；如果没有任务，则需要设置为null。core的值如果是true，则使用corePoolSize与线程数进行比较，判断线程数是否已经达到核心线程数；否则，使用maximumPoolSize判断线程数是否已经达到最大线程数。
  Worker类继承了AbstractQueuedSynchronizer，实现了Runnable接口。线程池中，每一个线程都被封装成一个Worker对象，ThreadPool维护的其实就是存放在变量HashSet workers中的一组Worker对象。Worker 在执行完任务后，还会无限循环获取工作队列里的任务来执行。我们选择关键源码简单看一下 Worker 类，它是 ThreadPoolExecutor 类的一个内部类：
  1.     private final class Worker extends AbstractQueuedSynchronizer implements Runnable
  2.     {
  3.         /** Thread this worker is running in.  Null if factory fails. 处理任务的线程 */
  4.         @SuppressWarnings("serial") // Unlikely to be serializable
  5.         final Thread thread;
  6.         /** Initial task to run.  Possibly null.保存传入的任务 */
  7.         @SuppressWarnings("serial") // Not statically typed as Serializable
  8.         Runnable firstTask;
  9.         /** Per-thread task counter */
  10.         volatile long completedTasks;
  12.         // TODO: switch to AbstractQueuedLongSynchronizer and move
  13.         // completedTasks into the lock word.
  15.         /**
  16.          * Creates with given first task and thread from ThreadFactory.
  17.          * @param firstTask the first task (null if none)
  18.          */
  19.         Worker(Runnable firstTask) {
  20.             //设置AQS的同步状态
  21.             setState(-1); // inhibit interrupts until runWorker
  22.             // 封装任务
  23.             this.firstTask = firstTask;
  24.             // 新增执行任务的线程，this表示线程，故Worker对象在启动的时候就会调用函数run
  25.             this.thread = getThreadFactory().newThread(this);
  26.         }
  27.         /** Delegates main run loop to outer runWorker. */
  28.         public void run() {
  29.             runWorker(this);
  30.         }
  31.         protected boolean isHeldExclusively() {
  32.             return getState() != 0;
  33.         }
  34.         // 使用AQS实现独占锁，不允许重入
  35.         protected boolean tryAcquire(int unused) {
  36.             // 使用CAS修改状态，不允许重入
  37.             if (compareAndSetState(0, 1)) {
  38.                 setExclusiveOwnerThread(Thread.currentThread());
  39.                 return true;
  40.             }
  41.             return false;
  42.         }
  43.     }
复制代码
  Worker类中的函数run()的实现都在 runWorker() 方法中，故我们看一下runWorker()的源码。
runWorker执行任务源码详解
  1.    /**
  2.      * Main worker run loop.  Repeatedly gets tasks from queue and
  3.      * executes them, while coping with a number of issues:
  4.      *
  5.      * 1. We may start out with an initial task, in which case we
  6.      * don't need to get the first one. Otherwise, as long as pool is
  7.      * running, we get tasks from getTask. If it returns null then the
  8.      * worker exits due to changed pool state or configuration
  9.      * parameters.  Other exits result from exception throws in
  10.      * external code, in which case completedAbruptly holds, which
  11.      * usually leads processWorkerExit to replace this thread.
  12.      *
  13.      * 2. Before running any task, the lock is acquired to prevent
  14.      * other pool interrupts while the task is executing, and then we
  15.      * ensure that unless pool is stopping, this thread does not have
  16.      * its interrupt set.
  17.      *
  18.      * 3. Each task run is preceded by a call to beforeExecute, which
  19.      * might throw an exception, in which case we cause thread to die
  20.      * (breaking loop with completedAbruptly true) without processing
  21.      * the task.
  22.      *
  23.      * 4. Assuming beforeExecute completes normally, we run the task,
  24.      * gathering any of its thrown exceptions to send to afterExecute.
  25.      * We separately handle RuntimeException, Error (both of which the
  26.      * specs guarantee that we trap) and arbitrary Throwables.
  27.      * Because we cannot rethrow Throwables within Runnable.run, we
  28.      * wrap them within Errors on the way out (to the thread's
  29.      * UncaughtExceptionHandler).  Any thrown exception also
  30.      * conservatively causes thread to die.
  31.      *
  32.      * 5. After task.run completes, we call afterExecute, which may
  33.      * also throw an exception, which will also cause thread to
  34.      * die. According to JLS Sec 14.20, this exception is the one that
  35.      * will be in effect even if task.run throws.
  36.      *
  37.      * The net effect of the exception mechanics is that afterExecute
  38.      * and the thread's UncaughtExceptionHandler have as accurate
  39.      * information as we can provide about any problems encountered by
  40.      * user code.
  41.      *
  42.      * @param w the worker
  43.      */
  44.     final void runWorker(Worker w) {
  45.         Thread wt = Thread.currentThread();// 获取当前线程
  46.         // 用创建线程时传入的firstTask初始化任务
  47.         Runnable task = w.firstTask;
  48.         w.firstTask = null;// 清空 Worker 对象的第一个任务
  49.         // 解锁，允许中断
  50.         w.unlock(); // allow interrupts
  51.         // 标记线程是否因为异常退出循环，默认为 true
  52.         boolean completedAbruptly = true;
  53.         try {
  54.             //循环获取任务，在getTask()返回null时跳出while循环，且回收线程
  55.             // 如果firstTask为null，则使用getTask()从任务队列取任务
  56.             while (task != null || (task = getTask()) != null) {
  57.                 w.lock();
  58.                 // If pool is stopping, ensure thread is interrupted;
  59.                 // if not, ensure thread is not interrupted.  This
  60.                 // requires a recheck in second case to deal with
  61.                 // shutdownNow race while clearing interrupt
  62.                 if ((runStateAtLeast(ctl.get(), STOP) ||
  63.                      (Thread.interrupted() &&
  64.                       runStateAtLeast(ctl.get(), STOP))) &&
  65.                     !wt.isInterrupted())
  66.                     //当前线程被打上中断标志
  67.                     wt.interrupt();
  68.                 try {
  69.                     //执行任务前的钩子方法，让继承类做一些统计之类的事情
  70.                     beforeExecute(wt, task);
  71.                     try {
  72.                         // 执行任务
  73.                         task.run();
  74.                         // 执行任务后的钩子方法
  75.                         afterExecute(task, null);
  76.                     } catch (Throwable ex) {
  77.                         afterExecute(task, ex);
  78.                         throw ex;
  79.                     }
  80.                 } finally {
  81.                     // 把执行完的任务设置为null，从而触发getTask()重新获取任务；增加任务数，同时释放锁
  82.                     task = null;
  83.                     // 更新已完成的任务数
  84.                     w.completedTasks++;
  85.                     w.unlock();
  86.                 }
  87.             }
  88.             // 标记线程正常退出
  89.             completedAbruptly = false;
  90.         } finally {
  91.             // 回收/销毁线程
  92.             processWorkerExit(w, completedAbruptly);
  93.         }
  94.     }
复制代码
  getTask结果为null时跳出while循环，执行processWorkerExit，销毁线程；如果执行processWorkerExit的入参completedAbruptly=true，表示线程意外退出，需要减少计数。
  在这个方法中,while (task != null || (task = getTask()) != null)说明线程会循环地使用getTask函数从线程池的任务队列中取任务并执行任务，直到取出的任务对象为null，此时线程池已经关闭或者任务队列为空。接下来就会跳出 while 循环进入 finally 语句块执行processWorkerExit()，尝试回收线程。这行while循环代码也说明如果队列中没有任务且核心线程数小于corePoolSize，则来任务时一定会创建核心线程，而已经创建的核心线程哪怕一直无所事事也不会执行新任务。
  到这里就介绍完线程池脚踏实地干活的Worker类了，简单归纳一下：worker对象同时封装了一个任务Runnable firstTask和一个线程final Thread thread，它的创建依赖于线程状态，任务的执行也是在worker里去处理。这里有个小技巧值得大家细细品味——在函数runWorker中，会把执行完的任务及时设置为null但保留线程，从而触发函数getTask()重新获取任务，实现线程复用。
processWorkerExit回收线程源码详解
  1.    /**
  2.      * Performs cleanup and bookkeeping for a dying worker. Called
  3.      * only from worker threads. Unless completedAbruptly is set,
  4.      * assumes that workerCount has already been adjusted to account
  5.      * for exit.  This method removes thread from worker set, and
  6.      * possibly terminates the pool or replaces the worker if either
  7.      * it exited due to user task exception or if fewer than
  8.      * corePoolSize workers are running or queue is non-empty but
  9.      * there are no workers.
  10.      *
  11.      * @param w the worker
  12.      * @param completedAbruptly if the worker died due to user exception
  13.      */
  14.     private void processWorkerExit(Worker w, boolean completedAbruptly) {
  15.         if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
  16.             decrementWorkerCount();
  18.         final ReentrantLock mainLock = this.mainLock;
  19.         mainLock.lock();
  20.         try {
  21.             completedTaskCount += w.completedTasks;
  22.             // 将线程引用移出线程池
  23.             workers.remove(w);
  24.         } finally {
  25.             mainLock.unlock();
  26.         }
  27.         // 回收线程
  28.         tryTerminate();
  30.         int c = ctl.get();
  31.         if (runStateLessThan(c, STOP)) {
  32.             if (!completedAbruptly) {
  33.                 int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
  34.                 if (min == 0 && ! workQueue.isEmpty())
  35.                     min = 1;
  36.                 if (workerCountOf(c) >= min)
  37.                     return; // replacement not needed
  38.             }
  39.             addWorker(null, false);
  40.         }
  41.     }
复制代码
  事实上，在这个processWorkerExit函数中，将线程引用移出线程池线程集合且调用了tryTerminate()就已经回收了线程。但由于引起线程销毁的可能性有很多，故线程池要判断是什么触发了这次销毁，是否要改变线程池的现阶段状态，是否要根据新状态重新分配线程。
getTask获取任务源码详解

  本节，我们来分析线程怎么通过 getTask() 方法从任务队列中取出任务的。
  1.     /**
  2.      * Performs blocking or timed wait for a task, depending on
  3.      * current configuration settings, or returns null if this worker
  4.      * must exit because of any of:
  5.      * 1. There are more than maximumPoolSize workers (due to
  6.      *    a call to setMaximumPoolSize).
  7.      * 2. The pool is stopped.
  8.      * 3. The pool is shutdown and the queue is empty.
  9.      * 4. This worker timed out waiting for a task, and timed-out
  10.      *    workers are subject to termination (that is,
  11.      *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
  12.      *    both before and after the timed wait, and if the queue is
  13.      *    non-empty, this worker is not the last thread in the pool.
  14.      *
  15.      * @return task, or null if the worker must exit, in which case
  16.      *         workerCount is decremented
  17.      */
  18.     private Runnable getTask() {
  19.         boolean timedOut = false; // Did the last poll() time out?
  20.         // 死循环，此方法要么返回 null，要么返回 Runnable 对象代表取到了任务
  21.         for (;;) {
  22.             int c = ctl.get();
  24.             // Check if queue empty only if necessary.
  25.             if (runStateAtLeast(c, SHUTDOWN)
  26.                 && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
  27.               //线程处于非运行状态时，若满足如下两个条件之一，则线程数减1
  28.               // 1.线程处于 STOP 及以上（STOP、TIDYING、TERMINAL）的状态；2.任务队列为空
  29.                 decrementWorkerCount();
  30.                 return null;
  31.             }
  32.             // 计算线程总数
  33.             int wc = workerCountOf(c);
  35.             // Are workers subject to culling? 可以销毁线程吗
  36.             // 变量timed用于判断是否需要进行超时控制
  37.             boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
  38.            // 如下场景可以销毁线程：①线程数大于最大线程数；
  39.            // ②允许核心线程被回收且空闲时间超过存活时间；
  40.            // ③非核心线程空闲时间超过存活时间
  41.            // TODO wc可以为0吗？
  42.             if ((wc > maximumPoolSize || (timed && timedOut))
  43.                 && (wc > 1 || workQueue.isEmpty())) {
  44.                 // 线程数减一成功时返回null，表名当前线程可以被回收
  45.                 if (compareAndDecrementWorkerCount(c))
  46.                     return null;
  47.                 continue;
  48.             }
  50.             try {
  51.             /**
  52.              *  从阻塞队列中取Runnable类型的任务对象，
  53.              */
  54.                 Runnable r = timed ?
  55.                 // timed 为 true，调用poll方法，如果keepAliveTime纳秒内没有拿到任务，则返回null
  56.                     workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
  57.                 // timed 为 false，调用take方法，此时线程永久阻塞直到有任务对象返回
  58.                     workQueue.take();
  59.                 if (r != null)
  60.                     // 返回从阻塞队列中取到的任务，终止循环
  61.                     return r;
  62.                 // 线程没有拿到任务，标记为已过期等待被销毁
  63.                 timedOut = true;
  64.             } catch (InterruptedException retry) {
  65.                 //当前线程在获取任务时发生中断，则设置为不过期并返回循环重试
  66.                 timedOut = false;
  67.             }
  68.         }
  69.     }
复制代码
tryTerminate终止线程源码详解

在《Java线程池状态和状态切换》中已经介绍过，通过 shutdownNow 与 shutdown可以触发线程池关闭流程，当方法执行完毕后，线程池将会进入 STOP 或者 SHUTDOWN 状态。但是此时线程池并未真正的被关闭，在runWorker方法最后的finally块中，调用了processWorkerExit方法，其逻辑实现中调用了一个 tryTerminate 方法，这个才是正在关闭线程池的钩子方法。楼兰胡杨本节和各位一起看看函数tryTerminate的源码。
  1.     /**
  2.      * Transitions to TERMINATED state if either (SHUTDOWN and pool
  3.      * and queue empty) or (STOP and pool empty).  If otherwise
  4.      * eligible to terminate but workerCount is nonzero, interrupts an
  5.      * idle worker to ensure that shutdown signals propagate. This
  6.      * method must be called following any action that might make
  7.      * termination possible -- reducing worker count or removing tasks
  8.      * from the queue during shutdown. The method is non-private to
  9.      * allow access from ScheduledThreadPoolExecutor.
  10.      */
  11.     final void tryTerminate() {
  12.         for (;;) {
  13.             int c = ctl.get();
  14.             // 如果线程池不处于预停机状态，则不进行停机
  15.             if (isRunning(c) ||
  16.                 runStateAtLeast(c, TIDYING) ||
  17.                 (runStateLessThan(c, STOP) && ! workQueue.isEmpty()))
  18.                 return;
  19.             if (workerCountOf(c) != 0) { // Eligible to terminate
  20.                 // 当前还有工作线程，不停机
  21.                 interruptIdleWorkers(ONLY_ONE);
  22.                 return;
  23.             }
  24.             // 线程处于预关闭状态，开始关闭线程池
  25.             final ReentrantLock mainLock = this.mainLock;
  26.             mainLock.lock();
  27.             try {
  28.                 // 尝试通过 CAS 将线程池状态修改为 TIDYING
  29.                 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
  30.                     try {
  31.                         terminated();
  32.                     } finally {
  33.                         // 尝试通过 CAS 将线程池状态修改为 TERMINATED
  34.                         ctl.set(ctlOf(TERMINATED, 0));
  35.                         termination.signalAll();
  36.                         container.close();
  37.                     }
  38.                     return;
  39.                 }
  40.             } finally {
  41.                 mainLock.unlock();
  42.             }
  43.             // else retry on failed CAS
  44.         }
  45.     }
复制代码
  哦了，再来看看函数 terminated()，是不是瞬间感觉很坑爹？它的方法体里面神！马！也！没！干！淡定，其实它是个钩子方法，允许通过重写在线程池被终止时做一些特殊的业务逻辑，默认的线程池没有什么要做的事情，当然也没有必要写什么啦～
  1.     /**
  2.      * Method invoked when the Executor has terminated.  Default
  3.      * implementation does nothing. Note: To properly nest multiple
  4.      * overridings, subclasses should generally invoke
  5.      * {@code super.terminated} within this method.
  6.      */
  7.     protected void terminated() { }
复制代码
创建多少线程比较合适

  要想合理的配置线程池线程数，就必须首先分析任务特性，可以从以下几个角度来进行分析：

  • 任务的性质：CPU 密集型任务，IO 密集型任务和混合型任务；
  • 任务的优先级：高，中和低；
  • 任务的执行时间；
  • 任务的依赖关系：是否依赖其他系统资源，如数据库连接；
  上面都是需要考虑的因素，至于应该创建多个线程，还是需要压测的，单纯考虑CPU 密集型任务和IO 密集型不合适。另外，建议任务性质不同的任务用不同规模的线程池分开处理，保证业务解耦。
初始化线程池

  默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务之后才会创建线程。在实际中如果需要线程池创建之后立即创建线程，可以通过prestartCoreThreadhe 和 prestartAllCoreThreads两个方法实现，二者都通过调用函数boolean addWorker(Runnable firstTask, boolean core)来实现。
  prestartCoreThread()用于在创建线程池的时候初始化一个核心线程，实现源码如下：
  1.     /**
  2.      * Starts a core thread, causing it to idly wait for work. This
  3.      * overrides the default policy of starting core threads only when
  4.      * new tasks are executed. This method will return {@code false}
  5.      * if all core threads have already been started.
  6.      *
  7.      * @return {@code true} if a thread was started
  8.      */
  9.     public boolean prestartCoreThread() {
  10.         return workerCountOf(ctl.get()) < corePoolSize &&
  11.             //注意传进去的参数firstTask是null
  12.             addWorker(null, true);
  13.     }
复制代码
  prestartAllCoreThreads()：初始线程池时创建所有核心线程，实现源码如下：
  1.     /**
  2.      * Starts all core threads, causing them to idly wait for work. This
  3.      * overrides the default policy of starting core threads only when
  4.      * new tasks are executed.
  5.      *
  6.      * @return the number of threads started
  7.      */
  8.     public int prestartAllCoreThreads() {
  9.         int n = 0;
  10.         while (addWorker(null, true))//注意传入的任务是null
  11.             ++n;
  12.         return n;
  13.     }
复制代码
  注意上面传进boolean addWorker(Runnable firstTask, boolean core)的参数firstTask是null，创建的核心线程会被阻塞在getTask方法中，等待获取任务队列中的任务。
  八股文：当任务数超过核心线程数时，如何直接启用最大线程数maximumPoolSize？
分析：题目中【任务数超过核心线程数】可以理解为【线程数超过核心线程数】。
答：综合【线程池执行流程】所述得知，既然我们的预期是任务数超过核心线程数时新提交的任务不进入任务队列，就需要人为干预第二步 任务入队。这答案就显而易见了——在创建线程池的时候，指定任务队列使用SynchronousQueue。SynchronousQueue是不能存储元素的一个队列，它的特性是每生产一个任务就需要指定一个消费者来处理这个任务；否则，阻塞生产者。
结束语

  至此，已经介绍完线程池核心知识点，预祝各位读者在工作中能够迅速而准确地处理线程池相关需求，就像运斤成风一样。
  在编程这个复杂严峻的环境中，请活得优雅坦然：也许你的钱包空空如也，也许你的工作不够好，也许你正处在困境中，也许你被情所弃。不论什么原因，请你在出门时，一定要把自己打扮地清清爽爽，昂起头，挺起胸，面带微笑，从容自若地面对生活和面对工作。人生就像蒲公英，没事尽量少吹风；只要你自己真正撑起来了一片天地，别人无论如何是压不垮你的，内心的强大才是真正的强大。
Reference


  • JUC线程池: ThreadPoolExecutor详解
  • https://blog.csdn.net/xaiobit_hl/article/details/132281971
  • https://blog.csdn.net/weixin_43918863/article/details/146354499
  • https://zhuanlan.zhihu.com/p/2983855645
  • https://www.cnblogs.com/w08e/p/18410687
  读后有收获，小礼物走一走，请作者喝咖啡。  
4.png
Buy me a coffee. ☕Get red packets.        
5.jpeg
        
6.jpeg
作者：楼兰胡杨链接：https://www.cnblogs.com/east7/p/18903603本文版权归作者和博客园共有，欢迎转载，但请注明原文链接，并保留此段声明，否则保留追究法律责任的权利。
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
寨重
2025-5-30 13:54:47

0

粉丝关注

5

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫