👋hi，我不是一名外包公司的员工，也不会偷吃茶水间的零食，我的梦想是能写高端CRUD

🔥 2025本人正在沉淀中... 博客更新速度++

👍 欢迎点赞、收藏、关注，跟上我的更新节奏

📚欢迎订阅专栏，专栏别名《在2B工作中寻求并发是否搞错了什么》

前言

当我们创建一个ThreadPoolExecutor的时候，你是否会好奇🤔，它到底发生了什么？比如：

我传的拒绝策略、线程工厂是啥时候被使用的？
核心线程数是个啥？最大线程数和它又有什么关系？
线程池，它是怎么调度，我们传入的线程？
...

不要着急，小手手点上关注、点赞、收藏。主播马上从源码的角度带你们探索神秘线程池的世界。

public ThreadPoolExecutor(int corePoolSize,//线程池的核心线程数量
                          int maximumPoolSize,//线程池的最大线程数
                          long keepAliveTime,//当线程数大于核心线程数时，多余的空闲线程存活的最长时间
                          TimeUnit unit,//时间单位
                          BlockingQueue<Runnable> workQueue,//阻塞队列，用来储存等待执行任务的队列
                          ThreadFactory threadFactory,//线程工厂，用来创建线程，一般默认即可
                          RejectedExecutionHandler handler//拒绝策略，当提交的任务过多而不能及时处理时，我们可以定制策略来处理任务
                           ) {

ThreadPoolExecutor构造方法

所有伟大的开始，源于构造方法

我们可以看到，构造方法里，只是对它做了数量的校验和赋值：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
   
    if (corePoolSize < 0 ||                    // 核心线程数是可以为0，但不能小于0
        maximumPoolSize <= 0 ||                // 最大线程数不能小于0
        maximumPoolSize < corePoolSize ||    // 核心线程数小于最大线程数
        keepAliveTime < 0)                    // 存活时间大于0
        throw new IllegalArgumentException();
    // 工作队列、线程工厂、拒绝策略不能为null
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

有没有好奇的uu？构造线程池，有哪些参数可以不传？
下面的构造方法，早已告诉了我们答案 --- 线程工厂和拒绝策略。

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
   
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

如果我们不传线程工厂和拒绝策略，那么就会有默认的线程工厂和拒绝策略：

默认的线程工厂：DefaultThreadFactory

// 默认的线程工厂
static class DefaultThreadFactory implements ThreadFactory {
   
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
   
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    public Thread newThread(Runnable r) {
   
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

默认的拒绝策略：AbortPolicy

// ThreadPoolExecutor默认的拒绝策略
private static final RejectedExecutionHandler defaultHandler =
    new AbortPolicy();

// AbortPolicy拒绝策略
public static class AbortPolicy implements RejectedExecutionHandler {
   

    public AbortPolicy() {
    }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
   
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

提交任务（execute方法）

🤔提交任务的流程是什么勒？源码里，Doug Lea注释里写的很清楚了。

第一步：

* 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.
==========================我是分割线===========================
1、如果运行中的线程数少于corePoolSize，尝试用当前任务作为首个任务启动新线程。
addWorker方法会原子性地检查线程池状态和worker数量，避免非法创建线程。

第二步：

* 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.
==========================我是分割线===========================
2、 如果任务成功入队，仍需二次检查：
   - 是否应补充新线程（例如之前检查后有线程死亡）
   - 线程池是否已关闭
   根据检查结果决定回滚入队操作或创建新线程

第三步：

* 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.
==========================我是分割线===========================
3、如果无法入队（队列已满），尝试创建新线程。
  若失败（线程池关闭或已达最大线程数），执行拒绝策略。

是不是长长的文字不想看，没有关系，一图顶千言👇：

从源码的角度的来看：

    public void execute(Runnable command) {
   
        if (command == null)
            throw new NullPointerException();
        // 获取ctl，ctl是什么？等下会详细的说说，它表示线程池的状态和工作线程数
        int c = ctl.get();
        // 工作线程数小于核心线程数
        if (workerCountOf(c) < corePoolSize) {
   
            if (addWorker(command, true))        // 尝试添加核心线程
                return;
            c = ctl.get();
        }
        // 线程池状态为正在运行 且 成功放入阻塞队列
        if (isRunning(c) && workQueue.offer(command)) {
   
            int recheck = ctl.get();
            // 二次检查的线程状态不为正在运行 且 移除线程成功
            if (! isRunning(recheck) && remove(command))
                reject(command);                    // 触发拒绝策略
            else if (workerCountOf(recheck) == 0)   // 工作线程数为0
                addWorker(null, false);                // 尝试添加非核心线程
        }
        else if (!addWorker(command, false))         // 尝试添加非核心线程
            reject(command);                          // 触发拒绝策略
    }

二次检查状态是为了处理并发场景下的竞态条件，具体检查两个维度：

状态检查：用isRunning(recheck)验证线程池是否被关闭（SHUTDOWN/STOP），若关闭则回滚已入队任务并触发拒绝策略。
线程检查：当workerCount==0时（核心线程被回收/异常终止），需创建非核心线程保证队列任务能被消费。

竞态条件（race condition）指的是两个或者以上进程或者线程并发执行时，其最终的结果依赖于进程或者线程执行的精确时序。竞争条件会产生超出预期的情况。

ctl

刚刚看提交任务源码中埋下的伏笔 --- 什么是ctl？我们下面具体来说说。

Doug Lea用一个变量，来表示2种状态:

线程池当前状态（高3位）
当前线程池工作线程个数（低29位）

// 初始化ctl，线程池状态为正在运行，工作线程数为0
// RUNNING  = 111_00000_00000000_00000000_00000000 (-536870912)
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

// rs: 运行状态   wc:工作线程数量
private static int ctlOf(int rs, int wc) {
    return rs | wc; }


// 用来算线程池状态，左移27位 
private static final int COUNT_BITS = Integer.SIZE - 3;
// CAPACITY = 000_11111_11111111_11111111_11111111
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

线程池状态

private static final int COUNT_BITS = Integer.SIZE - 3;

private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

相当于

// 状态值定义（高3位）
RUNNING    = 111_00000_00000000_00000000_00000000 (-536870912)
SHUTDOWN   = 000_00000_00000000_00000000_00000000 (0)
STOP       = 001_00000_00000000_00000000_00000000 (536870912)
TIDYING    = 010_00000_00000000_00000000_00000000 (1073741824)
TERMINATED = 011_00000_00000000_00000000_00000000 (1610612736)

计算出当前线程池的状态

CAPACITY = 000_11111_11111111_11111111_11111111

// 取高3位（屏蔽低29位）
private static int runStateOf(int c) {
    return c & ~CAPACITY; }

// 状态判断
private static boolean isRunning(int c) {
    return c < SHUTDOWN; }

线程池状态的转变

工作线程数量

CAPACITY = 000_11111_11111111_11111111_11111111

// 取低29位（屏蔽高3位）
private static int workerCountOf(int c) {
    return c & CAPACITY; }

// CAS更新线程数
private boolean compareAndIncrementWorkerCount(int expect) {
   
   return ctl.compareAndSet(expect, expect + 1);
}

尝试向线程池添加线程(addWorker方法)

状态检查与CAS计数更新：通过循环检查线程池状态（是否可接受新任务）和当前线程数（是否超过核心或最大线程数限制），使用CAS原子操作增加workerCount。

// 入参：
// 1、firstTask： 新线程要执行的第一个任务（可能为null）
// 2、core：决定用corePoolSize（true）还是maximumPoolSize（false）作为线程数上限
private boolean addWorker(Runnable firstTask, boolean core：) {
   
    retry:
    for (;;) {
   
        int c = ctl.get();
        // 获取线程池状态
        int rs = runStateOf(c);

        // 通过ctl原子变量（高位存状态，低位存线程数）检查线程池状态，防止在非运行状态下创建新线程
        // 1、rs >= SHUTDOWN 
        //    判断线程池状态是否处于SHUTDOWN或更高状态（STOP, TIDYING, TERMINATED）
        // 2、!(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty())
        // 只有当同时满足以下3个条件时，才会取反为false：
        //   - 状态正好是SHUTDOWN（不是更高状态）
        //   - firstTask为空（表示不是新提交的任务）
        //   - 工作队列非空（还有未处理的任务）
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
   
            // 获取工作线程数
            int wc = workerCountOf(c);
            // 校验是否超过容量限制
            // 1、wc >= CAPACITY：判断工作线程数量是否大等于最大工作线程数量
            // 2、wc >= (core ? corePoolSize : maximumPoolSize))：
            // 判断工作线程数量是否大等于 核心线程数/最大线程数
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            if (compareAndIncrementWorkerCount(c)) // 通过CAS保证线程数增减的原子性，避免并发问题
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
        }
    }
    ...

创建工作线程：创建Worker对象并加锁将其加入线程集合。若线程启动失败，则回滚计数和集合状态。

private boolean addWorker(Runnable firstTask, boolean core) {
   
    ...

    boolean workerStarted = false;    // worker线程是否被启动
    boolean workerAdded = false;    // worker是否被添加到workers集合中  private final HashSet<Worker> workers = new HashSet<Worker>();
    Worker w = null;
    try {
   
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
   
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
   

                int rs = runStateOf(ctl.get());

                // 1、rs < SHUTDOWN：线程池处于RUNNING状态
                // 目的：确保只有在线程池可用时才能创建新线程
                // 2、(rs == SHUTDOWN && firstTask == null)：线程池处于SHUTDOWN状态且没有初始任务（firstTask为null）
                // SHUTDOWN 状态下不允许添加新任务（firstTask != null 会直接拒绝），但允许创建 无初始任务 的线程来消费队列中的残留任务
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
   
                    if (t.isAlive())  // 检查新创建的线程是否已处于活跃状态
                        throw new IllegalThreadStateException();
                    workers.add(w);        // 添加到Worker集合
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
   
                mainLock.unlock();
            }
            if (workerAdded) {
   
                t.start();        // 启动线程
                workerStarted = true;
            }
        }
    } finally {
   
        if (! workerStarted)
            addWorkerFailed(w);  // 失败时的回滚逻辑
    }
    return workerStarted;
}

这里简单说下，失败时候的回滚逻辑

从集合中移除Worker
CAS递减线程数
尝试终止线程池（后面会详细说说🤗）

private void addWorkerFailed(Worker w) {
   
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
   
        if (w != null)                
            workers.remove(w);        // 从集合中移除Worker
        decrementWorkerCount();        // CAS递减线程数
        tryTerminate();                // 尝试终止线程池，后面会详细说说
    } finally {
   
        mainLock.unlock();
    }
}

worker类

刚刚在addWorker方法里，我们创建了Worker类，🤔聪明的你，一定很想知道这个类到底是干什么的吧！
Worker类是ThreadPoolExecutor的核心内部类，它直接负责封装工作线程和管理任务执行。其设计巧妙地将线程、任务和锁机制结合在一起。

private final class Worker
    extends AbstractQueuedSynchronizer // 继承AQS实现锁机制
    implements Runnable {
                // 自身作为线程的Runnable目标
    final Thread thread;              // 实际的工作线程
    Runnable firstTask;               // 初始任务（可能为null）
    volatile long completedTasks;     // 完成的任务计数器
}

构造方法，在addWorker的时候，就会调用到这个构造方法。
怎么样？你是否看到了，自己构造线程池时传入的线程工厂被调用😄？

Worker(Runnable firstTask) {
   
  this.firstTask = firstTask;
  this.thread = getThreadFactory().newThread(this); // this即Worker自身
}

worker类的锁状态
通过继承 AQS（AbstractQueuedSynchronizer）实现了一个不可重入的独占锁，用于精确控制线程的中断行为和状态标识。

注释说了，0是未被锁定，1是被锁定了
// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.

work类的加锁/解锁

// 最终调用AQS的acquire方法。将state从0改为1
public void lock()        {
    acquire(1); }  
public boolean tryLock()  {
    return tryAcquire(1); }

// AQS的tryAcquire实现（Worker内部）
protected boolean tryAcquire(int unused) {
   
    if (compareAndSetState(0, 1)) {
    // CAS操作保证原子性
        setExclusiveOwnerThread(Thread.currentThread());
        return true;
    }
    return false;
}

解锁

// 最终调用AQS的tryRelease实现。将state从1改为0
public void unlock()      {
    release(1); }   

// AQS的tryRelease实现
protected boolean tryRelease(int unused) {
   
    setExclusiveOwnerThread(null);
    setState(0); // 无需CAS，因为只有持有锁的线程能调用此方法
    return true;
}

执行Worker（runWorker方法）

worker类，调用外部类ThreadPoolExecutor的runWorker方法

public void run() {
   
    runWorker(this);   // 调用外部类ThreadPoolExecutor的runWorker方法
}

让我们康康，线程池的runWorker方法吧！

final void runWorker(Worker w) {
   
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;        // 获取初始任务（可能为null）
    w.firstTask = null;                    // 清空初始任务引用
    w.unlock();                         // 允许中断（设置state=0，标记为初始可中断状态）
    boolean completedAbruptly = true;    // 是否因异常退出
    try {
   
        // ---- 核心循环：不断获取任务 ----
        // 执行初始任务，或者获取到的任务，getTask是怎么获取的，下文会介绍
        while (task != null || (task = getTask()) != null) {
   
            w.lock();   // 加锁标记线程为"工作中"
            // 检查线程池状态（若处于STOP需确保线程被中断）
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
   
                // 如果我们写了新类继承了ThreadPoolExector重写了beforeExecute方法，就执行
                // ThreadPoolExector这里没有任何实现: protected void beforeExecute(Thread t, Runnable r) { }
                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 {
   
                    // 如果我们写了新类继承了ThreadPoolExector重写了afterExecute方法，就执行
                    // ThreadPoolExector这里没有任何实现：protected void afterExecute(Runnable r, Throwable t) { }
                    afterExecute(task, thrown);
                }
            } finally {
   
                task = null;                // 清空任务引用
                w.completedTasks++;            // 统计完成的任务数
                w.unlock();                    // 解锁标记线程为“空闲”
            }
        }
        completedAbruptly = false;            // 没有因为异常导致退出
    } finally {
   
        processWorkerExit(w, completedAbruptly);  // 线程退出处理
    }
}

获取任务（getTask方法）

你是否会好奇🤔，刚刚我们在runWork中的task是从哪里获取的？while (task != null || (task = getTask()) != null) 。

private Runnable getTask() {
   
    boolean timedOut = false;         // 标记是否发生poll超时

    for (;;) {
   
        int c = ctl.get();
        int rs = runStateOf(c);        // 解析线程池状态

        // ===== 1. 状态检查：是否需要停止工作 =====
        // 条件1: 线程池状态 >= STOP（立刻停止）
        // 条件2: 线程池状态 >= SHUTDOWN 且 队列为空
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
   
            decrementWorkerCount();        // 减少工作线程数
            return null;                // 触发线程回收
        }

        int wc = workerCountOf(c);        // 当前工作线程数

        // ===== 2. 判断是否允许超时回收 =====
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        // ===== 3. 线程数溢出检查 =====
        // 情况1: 线程数超过maximumPoolSize（可能因配置修改导致）
        // 情况2: 允许超时回收 且 已发生超时
        //                 且
        // 情况1：工作线程数量大于1 或 工作队列为空
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
   
            if (compareAndDecrementWorkerCount(c))    // CAS减少线程数
                return null;
            continue;    // 竞争失败则重试
        }
        // ===== 4. 从队列获取任务 =====
        try {
   
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :  // 限时阻塞等待，超时后返回 null
                workQueue.take();                                       // 永久阻塞（核心线程默认行为）
            if (r != null)
                return r;
            timedOut = true;    // 标记超时
        } catch (InterruptedException retry) {
   
            timedOut = false;    // 中断重试（SHUTDOWN状态可能触发）
        }
    }
}

线程退出（processWorkerExit方法）

我们在runWork方法的时候，调用了processWorkerExit方法，好奇的你一定想看看他是怎么实现的吧！

final void runWorker(Worker w) {
   
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;        // 获取初始任务（可能为null）
    w.firstTask = null;                    // 清空初始任务引用
    w.unlock();                         // 允许中断（设置state=0，标记为初始可中断状态）
    boolean completedAbruptly = true;    // 是否因异常退出
    try {
   
        // ---- 核心循环：不断获取任务 ----
        while (task != null || (task = getTask()) != null) {
   ...}
        completedAbruptly = false;
    } finally {
   
        processWorkerExit(w, completedAbruptly);  // 线程退出处理
    }
}

processWorkerExit方法

private void processWorkerExit(Worker w, boolean completedAbruptly) {
   
    // ===== 1. 参数校验与状态记录 =====
    if (completedAbruptly) // 若因异常退出，需手动减少线程数
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
   
        // ===== 2. 统计完成任务数 =====
        completedTaskCount += w.completedTasks;
        workers.remove(w); // 从Worker集合中移除
    } finally {
   
        mainLock.unlock();
    }

    // ===== 3. 尝试终止线程池 =====
    tryTerminate();

    // ===== 4. 判断是否需要补充线程 =====
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
    // 线程池仍处于RUNNING/SHUTDOWN状态
        if (!completedAbruptly) {
    // 正常退出
            // 计算最小应保持的线程数
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1; // 队列非空时至少保留1个线程处理任务
            if (workerCountOf(c) >= min)
                return; // 当前线程数足够，无需补充
        }
        addWorker(null, false); // 补充新线程（无初始任务）
    }
}

线程池的关闭

之前说了这么多线程池是怎么运行，到这里也该说说线程池是怎么关闭的了🤭之前看源码的时候，经常出现的tryTerminate方法，也会在这里展开说说

线程池对我们开放了下面的3个方法，让我们来关闭线程池。

方法	行为特性	适用场景
shutdown()	温和关闭：停止接受新任务，执行完已提交任务	需要优雅关闭，确保任务不丢失
shutdownNow()	强制关闭：停止接受新任务，尝试中断所有线程，并返回未执行任务列表	需要立即释放资源，容忍任务丢弃
awaitTermination(long timeout, TimeUnit unit)	阻塞等待线程池完全终止（结合shutdown使用）	需要同步等待关闭完成

shutdown方法

shutdown方法的作用：

停止接受新任务：调用 shutdown() 后，线程池不再接受新提交的任务。
执行已提交的任务：线程池会继续执行已经提交但尚未完成的任务。
关闭线程池：在所有任务执行完毕后，线程池会逐步关闭。

public void shutdown() {
   
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
   
        checkShutdownAccess();
        advanceRunState(SHUTDOWN); // CAS更新状态为SHUTDOWN
        interruptIdleWorkers();    // 仅中断空闲线程
    } finally {
   
        mainLock.unlock();
    }
    tryTerminate(); // 尝试推进终止流程
}

interruptIdleWorkers方法
中断那些当前没有执行任务（即空闲）的工作线程。具体来说：

中断空闲线程：它会遍历所有工作线程，并尝试中断那些正在等待任务的线程（即处于空闲状态的线程）。
加速关闭过程：通过中断空闲线程，可以更快地释放资源，从而加速线程池的关闭过程。

private void interruptIdleWorkers(boolean onlyOne) {
   
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
   
        for (Worker w : workers) {
   
            Thread t = w.thread;
            // 关键点：尝试获取Worker锁（判断是否空闲）
            if (!t.isInterrupted() && w.tryLock()) {
   
                try {
   
                    t.interrupt(); // 仅中断空闲线程
                } catch (SecurityException ignore) {
   
                } finally {
   
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
   
        mainLock.unlock();
    }
}

shutdownNow方法

public List<Runnable> shutdownNow() {
   
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
   
        checkShutdownAccess();
        advanceRunState(STOP); // 更新为STOP状态
        interruptWorkers();    // 强制中断所有线程
        tasks = drainQueue();  // 排出未执行任务
    } finally {
   
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

interruptWorkers方法，我们可以看到里面调用了worker类interruptIfStarted

private void interruptWorkers() {
   
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
   
        for (Worker w : workers)
            w.interruptIfStarted(); // 无论是否空闲都尝试中断
    } finally {
   
        mainLock.unlock();
    }
}

// worker类interruptIfStarted方法
void interruptIfStarted() {
   
    Thread t;
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
   
        try {
   
            t.interrupt();
        } catch (SecurityException ignore) {
   
        }
    }
}

tryTerminate方法

tryTerminate() 的核心逻辑是检查线程池的状态和条件，决定是否可以将线程池的状态转换为 TERMINATED，并在适当的时候中断空闲的工作线程。

final void tryTerminate() {
   
    // 检查是否满足终止条件
    if ((runStateOf(c) == SHUTDOWN && !workQueue.isEmpty()) ||
        runStateAtLeast(c, TIDYING) || 
        isRunning(c)) 
        return;

    // 中断最后一个空闲线程（如有）
    if (workerCountOf(c) != 0) {
   
        interruptIdleWorkers(ONLY_ONE);
        return;
    }

    // 推进到TIDYING状态
    if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
   
        try {
   
            terminated(); // 空实现的钩子方法
        } finally {
   
            ctl.set(ctlOf(TERMINATED, 0));
            termination.signalAll(); // 唤醒awaitTermination()
        }
    }
}

后话

👍 线程池源码学习，只是刚刚开始，友友们，点上关注，跟上主播的学习节奏。

话说最近温度回暖了，感觉看源码的速度都变快了！哈哈哈哈哈，开玩笑的。

往期文章推荐

【Java并发】【线程池】带你从0-1入门线程池

【源码】【Java并发】【线程池】邀请您从0-1阅读ThreadPoolExecutor源码

前言

ThreadPoolExecutor构造方法

提交任务（execute方法）

ctl

线程池状态

工作线程数量

worker类

执行Worker（runWorker方法）

获取任务（getTask方法）

线程退出（processWorkerExit方法）

线程池的关闭

shutdown方法

shutdownNow方法

tryTerminate方法

后话

热门文章

最新文章

相关课程

相关电子书

相关实验场景

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

【源码】【Java并发】【线程池】邀请您从0-1阅读ThreadPoolExecutor源码

前言

ThreadPoolExecutor构造方法

提交任务（execute方法）

ctl

线程池状态

工作线程数量

worker类

执行Worker（runWorker方法）

获取任务（getTask方法）

线程退出（processWorkerExit方法）

线程池的关闭

shutdown方法

shutdownNow方法

tryTerminate方法

后话

热门文章

最新文章

相关课程

相关电子书

相关实验场景