Java线程池源码深度解析：生命周期全流程揭秘

2025年11月26日 互联网

Java线程池源码深度解析：从任务提交到Worker线程的生命周期

Java线程池是并发编程的核心组件，通过复用线程资源提升系统吞吐量，避免频繁创建销毁线程的开销。本文将从源码层面深度解析线程池的任务提交、队列管理、Worker线程创建与销毁的全生命周期，帮助开发者理解其底层机制。

一、任务提交：execute()方法的入口逻辑

线程池的任务提交入口是ThreadPoolExecutor.execute(Runnable command)方法，其核心逻辑分为三步：

1.1 快速失败检查

  1. if (command == null)
  2.     throw new NullPointerException();

源码首先校验任务非空，避免NPE。

1.2 核心参数计算

  1. int c = ctl.get(); // 获取线程池状态和worker数量
  2. if (workerCountOf(c) < corePoolSize) {
  3.     if (addWorker(command, true)) // 尝试创建核心线程
  4.         return;
  5.     c = ctl.get(); // 失败后重新获取状态
  6. }
  • ctl是AtomicInteger，高3位存储运行状态（RUNNING/SHUTDOWN等），低29位存储worker数量。
  • 若当前worker数小于corePoolSize，直接创建新worker执行任务（核心线程不空闲也不会被回收）。

1.3 任务入队与拒绝策略

  1. if (isRunning(c) && workQueue.offer(command)) {
  2.     int recheck = ctl.get();
  3.     if (!isRunning(recheck) && remove(command))
  4.         reject(command); // 线程池关闭时从队列移除任务并拒绝
  5.     else if (workerCountOf(recheck) == 0)
  6.         addWorker(null, false); // 队列非空但无worker时创建非核心线程
  7. } else if (!addWorker(command, false)) // 队列满时尝试创建非核心线程
  8.     reject(command); // 超过最大线程数则触发拒绝策略
  • 任务优先入队（workQueue），若队列满则尝试创建非核心线程（最大到maximumPoolSize）。
  • 拒绝策略通过RejectedExecutionHandler实现，默认抛出RejectedExecutionException。

二、Worker线程的创建与运行

Worker是线程池的核心工作单元，继承AQS并实现Runnable接口：

2.1 Worker的创建（addWorker方法）

  1. private boolean addWorker(Runnable firstTask, boolean core) {
  2.     retry:
  3.     for (;;) {
  4.         int c = ctl.get();
  5.         // 检查线程池状态是否允许添加worker
  6.         if (runStateAtLeast(c, SHUTDOWN)
  7.             && (runStateAtLeast(c, STOP)
  8.                 || firstTask != null
  9.                 || workQueue.isEmpty()))
  10.             return false;
  12.         for (;;) {
  13.             int wc = workerCountOf(c);
  14.             if (wc >= CAPACITY ||
  15.                 wc >= (core ? corePoolSize : maximumPoolSize))
  16.                 return false;
  17.             if (compareAndIncrementWorkerCount(c))
  18.                 break retry; // CAS增加worker数成功则退出循环
  19.             c = ctl.get(); // 失败后重新检查状态
  20.         }
  21.     }
  23.     boolean workerStarted = false;
  24.     boolean workerAdded = false;
  25.     Worker w = null;
  26.     try {
  27.         w = new Worker(firstTask); // 创建Worker对象
  28.         final Thread t = w.thread;
  29.         if (t != null) {
  30.             final ReentrantLock mainLock = this.mainLock;
  31.             mainLock.lock();
  32.             try {
  33.                 // 再次检查线程池状态
  34.                 int rs = runStateOf(ctl.get());
  35.                 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
  36.                     if (t.isAlive()) // 线程未启动则抛异常
  37.                         throw new IllegalThreadStateException();
  38.                     workers.add(w); // 添加到worker集合
  39.                     workerAdded = true;
  40.                 }
  41.             } finally {
  42.                 mainLock.unlock();
  43.             }
  44.             if (workerAdded) {
  45.                 t.start(); // 启动线程
  46.                 workerStarted = true;
  47.             }
  48.         }
  49.     } finally {
  50.         if (!workerStarted)
  51.             addWorkerFailed(w); // 启动失败则清理资源
  52.     }
  53.     return workerStarted;
  54. }
  • 双重检查：先通过CAS增加worker计数，再通过锁保证线程安全地添加到workers集合。
  • Worker构造：传入firstTask作为初始任务，线程启动后执行Worker.run()。

2.2 Worker的运行（runWorker方法）

  1. final void runWorker(Worker w) {
  2.     Thread wt = Thread.currentThread();
  3.     Runnable task = w.firstTask;
  4.     w.firstTask = null;
  5.     w.unlock(); // 释放初始锁，允许中断
  6.     boolean completedAbruptly = true;
  7.     try {
  8.         while (task != null || (task = getTask()) != null) {
  9.             w.lock(); // 获取锁防止被中断
  10.             if ((runStateAtLeast(ctl.get(), STOP)
  11.                  || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP)))
  12.                 && !wt.isInterrupted())
  13.                 wt.interrupt(); // 线程池停止时中断worker
  14.             try {
  15.                 beforeExecute(wt, task); // 钩子方法
  16.                 Throwable thrown = null;
  17.                 try {
  18.                     task.run(); // 执行任务
  19.                 } catch (RuntimeException x) {
  20.                     thrown = x; throw x;
  21.                 } catch (Error x) {
  22.                     thrown = x; throw x;
  23.                 } catch (Throwable x) {
  24.                     thrown = x; throw new Error(x);
  25.                 } finally {
  26.                     afterExecute(task, thrown); // 钩子方法
  27.                 }
  28.             } finally {
  29.                 task = null;
  30.                 w.completedTasks++; // 任务计数+1
  31.                 w.unlock();
  32.             }
  33.         }
  34.         completedAbruptly = false;
  35.     } finally {
  36.         processWorkerExit(w, completedAbruptly); // 清理worker
  37.     }
  38. }
  • 任务获取：优先执行firstTask，后续通过getTask()从队列拉取任务。
  • 中断处理：线程池停止时（STATE >= STOP）会中断worker线程。
  • 钩子方法：beforeExecute和afterExecute允许子类扩展。

2.3 任务获取策略（getTask方法）

  1. private Runnable getTask() {
  2.     boolean timedOut = false; // 是否超时
  3.     for (;;) {
  4.         int c = ctl.get();
  5.         int rs = runStateOf(c);
  6.         // 检查线程池状态
  7.         if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
  8.             decrementWorkerCount(); // 减少worker计数
  9.             return null;
  10.         }
  12.         int wc = workerCountOf(c);
  13.         boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
  14.         if ((wc > maximumPoolSize || (timed && timedOut))
  15.             && (wc > 1 || workQueue.isEmpty())) {
  16.             try {
  17.                 if (compareAndDecrementWorkerCount(c))
  18.                     return null; // 超过最大线程数或超时则退出
  19.                 continue;
  20.             }
  21.         }
  23.         try {
  24.             Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)
  25.                                : workQueue.take(); // 非超时则阻塞获取
  26.             if (r != null)
  27.                 return r;
  28.             timedOut = true;
  29.         } catch (InterruptedException retry) {
  30.             timedOut = false;
  31.         }
  32.     }
  33. }
  • 核心线程保留：若allowCoreThreadTimeOut=false，核心线程会一直存活（即使空闲）。
  • 非核心线程回收：空闲时间超过keepAliveTime的非核心线程会被回收。

三、Worker线程的销毁与资源清理

Worker线程的销毁通过processWorkerExit完成：

  1. private void processWorkerExit(Worker w, boolean completedAbruptly) {
  2.     if (completedAbruptly) // 异常退出时减少worker计数
  3.         decrementWorkerCount();
  5.     final ReentrantLock mainLock = this.mainLock;
  6.     mainLock.lock();
  7.     try {
  8.         completedTaskCount += w.completedTasks; // 更新总任务数
  9.         workers.remove(w); // 从集合移除
  10.     } finally {
  11.         mainLock.unlock();
  12.     }
  14.     tryTerminate(); // 尝试终止线程池
  15.     int c = ctl.get();
  16.     if (runStateLessThan(c, STOP)) // 线程池未停止时补充新worker
  17.         if (!completedAbruptly) {
  18.             int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
  19.             if (min == 0 && !workQueue.isEmpty())
  20.                 min = 1;
  21.             if (workerCountOf(c) >= min)
  22.                 return;
  23.         }
  24.         addWorker(null, false); // 补充非核心线程
  25. }
  • 资源释放：移除worker并更新任务统计。
  • 线程补充：若线程数低于corePoolSize（或1，当队列非空时），会创建新worker。

四、最佳实践与调优建议

  1. 合理配置参数：

    • corePoolSize：根据业务并发量设置，避免频繁创建销毁线程。
    • maximumPoolSize：应对突发流量，但需警惕资源耗尽。
    • keepAliveTime：非核心线程空闲回收时间，默认60秒。

  2. 选择合适的队列：

    • SynchronousQueue：无容量，适合高并发、任务执行快的场景。
    • LinkedBlockingQueue：无界队列，需防止OOM。
    • ArrayBlockingQueue：有界队列，需配合拒绝策略。

  3. 拒绝策略选择：

    • AbortPolicy（默认）：抛异常，适合严格场景。
    • CallerRunsPolicy：调用线程执行任务，降低负载。
    • DiscardOldestPolicy：丢弃最旧任务，尝试提交新任务。

  4. 监控与调优：

    • 通过ThreadPoolExecutor的activeCount()、getQueue().size()等方法监控运行状态。
    • 使用JMX或自定义指标收集线程池指标。

五、总结

Java线程池通过execute()方法接收任务，经由队列管理和Worker线程执行，最终通过processWorkerExit完成资源清理。其核心设计包括：

  • 状态控制：通过ctl原子变量管理线程池状态和worker数量。
  • 任务调度：优先使用核心线程，队列满后创建非核心线程，超量时触发拒绝策略。
  • 线程复用：Worker线程通过循环从队列拉取任务，避免频繁创建销毁。

理解线程池源码有助于开发者优化并发编程性能，避免资源浪费和系统过载。在实际应用中，需结合业务场景合理配置参数，并通过监控动态调整。

最新文章