ThreadPoolExecutor ⭐⭐⭐

// ThreadPoolExecutor 的完整构造方法 —— 所有参数一览无余
public ThreadPoolExecutor(
    int corePoolSize,          // 核心线程数：线程池长期维持的最小线程数量
    int maximumPoolSize,       // 最大线程数：线程池允许创建的线程数量上限
    long keepAliveTime,        // 空闲存活时间：非核心线程空闲后等待新任务的最长时间
    TimeUnit unit,             // 时间单位：keepAliveTime 的度量单位
    BlockingQueue<Runnable> workQueue,   // 工作队列：暂存等待执行的任务
    ThreadFactory threadFactory,         // 线程工厂：自定义线程的创建方式
    RejectedExecutionHandler handler     // 拒绝策略：线程池饱和时的应对方案
) { ... }

// 演示：corePoolSize 的懒加载特性
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    3,                      // corePoolSize = 3，核心线程数为 3
    5,                      // maximumPoolSize = 5
    60L,                    // keepAliveTime = 60
    TimeUnit.SECONDS,       // 时间单位为秒
    new LinkedBlockingQueue<>(10)  // 容量为 10 的工作队列
);
 
// 此时线程池刚创建完毕，内部线程数为 0（懒加载，不会预先创建线程）
System.out.println(pool.getPoolSize());  // 输出: 0
 
// 提交第 1 个任务 → 当前线程数(0) < corePoolSize(3) → 创建核心线程 #1
pool.execute(() -> doWork());
 
// 提交第 2 个任务 → 当前线程数(1) < corePoolSize(3) → 创建核心线程 #2
// 注意：即使线程 #1 此刻已空闲，仍然会创建新线程
pool.execute(() -> doWork());
 
// 提交第 3 个任务 → 当前线程数(2) < corePoolSize(3) → 创建核心线程 #3
pool.execute(() -> doWork());
 
// 此后再提交任务 → 当前线程数(3) == corePoolSize(3) → 任务进入 workQueue
pool.execute(() -> doWork());  // 这个任务会被放入队列

// 允许核心线程超时销毁 —— 适用于流量波动大的场景
pool.allowCoreThreadTimeOut(true);
// 开启后，核心线程空闲超过 keepAliveTime 也会被终止
// 线程池可以缩减到 0 个线程（完全空池状态）

// 预启动所有核心线程 —— 适用于启动后立即要承受高并发的场景
int started = pool.prestartAllCoreThreads();  // 返回实际启动的线程数
System.out.println(started);                  // 输出: 3（全部核心线程就位）
System.out.println(pool.getPoolSize());       // 输出: 3

// 直观理解 maximumPoolSize 与 corePoolSize 的关系
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    2,                               // corePoolSize = 2（常驻 2 个核心线程）
    5,                               // maximumPoolSize = 5（最多 5 个线程同时运行）
    30L,                             // 非核心线程空闲 30 秒后回收
    TimeUnit.SECONDS,                // 时间单位
    new ArrayBlockingQueue<>(3)      // 有界队列，容量为 3
);
 
// 最多能容纳的并发任务数 = maximumPoolSize + queueCapacity = 5 + 3 = 8
// 非核心线程数上限 = maximumPoolSize - corePoolSize = 5 - 2 = 3

╔════════════════════════════════════════════════════════════════╗
║               ThreadPoolExecutor 线程数量模型                   ║
╠════════════════════════════════════════════════════════════════╣
║                                                                ║
║  0 ─────── corePoolSize(2) ─────── maximumPoolSize(5)          ║
║  |              |                         |                    ║
║  |◄─ 懒加载区 ─►|◄──── 队列缓冲区 ────►|                    ║
║  |              |     (queue: 3)         |                    ║
║  |              |                         |                    ║
║  |  提交任务时   |  核心满 + 队列满时      |  达到此线程数       ║
║  |  逐步创建    |  开始创建非核心线程       |  再提交 → 拒绝     ║
║  |  核心线程    |                         |                    ║
║                                                                ║
║  总容量 = maximumPoolSize + queueCapacity = 5 + 3 = 8 个任务   ║
╚════════════════════════════════════════════════════════════════╝

// keepAliveTime 的行为演示
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    2,                              // 2 个核心线程（默认不会被回收）
    4,                              // 最多 4 个线程
    60L,                            // 非核心线程空闲 60 秒后被回收
    TimeUnit.SECONDS,               // 时间单位为秒
    new ArrayBlockingQueue<>(5)     // 有界队列容量 5
);
 
// 场景模拟：
// 1. 高峰期：大量任务涌入 → 队列满 → 创建到 4 个线程
// 2. 高峰过后：任务减少 → 2 个非核心线程空闲
// 3. 空闲 60 秒后 → 非核心线程被销毁 → 池中回到 2 个核心线程

// TimeUnit 枚举值一览 —— 从小到大
TimeUnit.NANOSECONDS;    // 纳秒（10⁻⁹ 秒）
TimeUnit.MICROSECONDS;   // 微秒（10⁻⁶ 秒）
TimeUnit.MILLISECONDS;   // 毫秒（10⁻³ 秒）—— 最常用
TimeUnit.SECONDS;        // 秒 —— 最常用
TimeUnit.MINUTES;        // 分钟
TimeUnit.HOURS;          // 小时
TimeUnit.DAYS;           // 天

// 1. 时间转换 —— 比手动乘除更清晰、更不容易出错
long millis = TimeUnit.MINUTES.toMillis(5);     // 5分钟 → 300000毫秒
long seconds = TimeUnit.HOURS.toSeconds(2);     // 2小时 → 7200秒
long nanos = TimeUnit.SECONDS.toNanos(1);       // 1秒 → 1000000000纳秒
 
// 2. 线程睡眠 —— 比 Thread.sleep(millis) 更语义化
TimeUnit.SECONDS.sleep(3);                      // 当前线程睡眠 3 秒
// 等价于 Thread.sleep(3000)，但可读性更强
 
// 3. 线程等待 —— 封装了 Object.wait()
TimeUnit.SECONDS.timedWait(lockObj, 5);         // 在 lockObj 上等待 5 秒
 
// 4. 线程 join —— 封装了 Thread.join()
TimeUnit.SECONDS.timedJoin(thread, 10);         // join 等待 10 秒

// 生产环境常见写法
new ThreadPoolExecutor(
    coreSize, maxSize,
    60L, TimeUnit.SECONDS,        // 60 秒 —— 最常见
    queue, factory, handler
);
 
// 或者对于需要快速回收的场景
new ThreadPoolExecutor(
    coreSize, maxSize,
    0L, TimeUnit.MILLISECONDS,    // 0 毫秒 —— 立即回收非核心线程
    queue, factory, handler
);

// workQueue 在任务流转中的位置
// execute(task) 方法的核心逻辑伪代码：
public void execute(Runnable command) {
    int currentWorkers = workerCount();          // 获取当前工作线程数量
 
    if (currentWorkers < corePoolSize) {
        // 第一步：核心线程未满 → 直接创建核心线程执行任务
        addWorker(command, true);                // true 表示核心线程
    }
    else if (workQueue.offer(command)) {
        // 第二步：核心线程已满 → 尝试将任务放入队列
        // offer() 是非阻塞方法，成功返回 true，失败（队列满）返回 false
    }
    else if (currentWorkers < maximumPoolSize) {
        // 第三步：队列也满了 → 创建非核心线程
        addWorker(command, false);               // false 表示非核心线程
    }
    else {
        // 第四步：全部满了 → 执行拒绝策略
        handler.rejectedExecution(command, this);
    }
}

// ThreadFactory 接口定义 —— 极致简洁
public interface ThreadFactory {
    Thread newThread(Runnable r);  // 根据 Runnable 任务创建并返回一个新线程
}

// 生产级自定义 ThreadFactory 示例
public class NamedThreadFactory implements ThreadFactory {
 
    private final AtomicInteger threadNumber = new AtomicInteger(1);  // 线程编号计数器（原子递增）
    private final String namePrefix;       // 线程名前缀，标识业务模块
    private final boolean daemon;          // 是否设为守护线程
    private final Thread.UncaughtExceptionHandler exceptionHandler;  // 未捕获异常处理器
 
    // 构造方法：传入业务名和守护线程标志
    public NamedThreadFactory(String poolName, boolean daemon) {
        this.namePrefix = poolName + "-worker-";          // 例如 "order-service-worker-"
        this.daemon = daemon;                              // 是否为守护线程
        this.exceptionHandler = (t, e) -> {                // 默认异常处理：记录日志
            System.err.println("Thread " + t.getName()     // 打印出问题的线程名
                + " threw exception: " + e.getMessage());  // 打印异常信息
            e.printStackTrace();                           // 打印完整堆栈
        };
    }
 
    @Override
    public Thread newThread(Runnable r) {
        // 创建新线程，名称格式如：order-service-worker-1, order-service-worker-2, ...
        Thread t = new Thread(r, namePrefix + threadNumber.getAndIncrement());
        t.setDaemon(daemon);                               // 设置守护线程属性
        t.setPriority(Thread.NORM_PRIORITY);               // 设置正常优先级（5）
        t.setUncaughtExceptionHandler(exceptionHandler);   // 绑定异常处理器
        return t;                                          // 返回配置好的线程实例
    }
}
 
// 使用自定义工厂创建线程池
ThreadPoolExecutor orderPool = new ThreadPoolExecutor(
    4, 8, 60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(100),
    new NamedThreadFactory("order-service", false),   // 自定义工厂：线程名带业务标识
    new ThreadPoolExecutor.CallerRunsPolicy()
);

"order-service-worker-1" #15 prio=5 os_prio=0 tid=0x00007f... nid=0x3a03 waiting ...
"order-service-worker-2" #16 prio=5 os_prio=0 tid=0x00007f... nid=0x3a04 runnable ...

// RejectedExecutionHandler 接口定义
public interface RejectedExecutionHandler {
    // r: 被拒绝的任务    executor: 触发拒绝的线程池
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

// 1️⃣ AbortPolicy —— 默认策略：直接抛异常，快速失败
// 适用场景：不允许丢失任务，必须让调用者感知到系统过载
public static class AbortPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 直接抛出 RejectedExecutionException，调用者必须处理
        throw new RejectedExecutionException(
            "Task " + r.toString() +                      // 被拒绝的任务信息
            " rejected from " + e.toString()              // 线程池当前状态信息
        );
    }
}
 
// 2️⃣ CallerRunsPolicy —— 调用者执行：谁提交谁执行
// 适用场景：不希望丢失任务，且能接受调用者线程被阻塞
public static class CallerRunsPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {        // 如果线程池还没有关闭
            r.run();                  // 直接在调用者线程（如 main 线程）中执行任务
            // 注意：这会阻塞调用者线程，形成天然的反压（backpressure）机制
            // 调用者线程忙于执行任务，暂时无法提交新任务，给线程池喘息时间
        }
    }
}
 
// 3️⃣ DiscardPolicy —— 静默丢弃：什么都不做
// 适用场景：允许丢失非关键任务（如日志、统计等）
public static class DiscardPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 空实现 —— 任务被悄悄丢弃，无异常、无日志、无通知
        // ⚠️ 生产环境慎用！任务悄无声息地消失，排查问题极其困难
    }
}
 
// 4️⃣ DiscardOldestPolicy —— 弃老策略：丢弃队列头部最老的任务，然后重新提交当前任务
// 适用场景：新任务比旧任务更重要的场景
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {        // 如果线程池还没有关闭
            e.getQueue().poll();      // 移除队列头部（最老的）任务
            e.execute(r);             // 重新尝试提交当前任务
            // ⚠️ 注意：如果配合 PriorityBlockingQueue 使用，
            // 丢弃的是优先级最高的任务（因为堆顶出队），可能不符合预期！
        }
    }
}

// 5️⃣ 自定义策略 —— 生产级实现
public class LogAndAlertPolicy implements RejectedExecutionHandler {
 
    private static final Logger log = LoggerFactory.getLogger(LogAndAlertPolicy.class);
 
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 1. 记录详细的线程池状态日志
        String msg = String.format(
            "[ThreadPool REJECT] task=%s, poolSize=%d, activeCount=%d, " +  // 当前线程数/活跃数
            "queueSize=%d, completedTask=%d, isShutdown=%s",                // 队列大小/已完成数
            r.toString(),                        // 被拒绝的任务描述
            executor.getPoolSize(),              // 当前池中线程总数
            executor.getActiveCount(),           // 正在执行任务的线程数
            executor.getQueue().size(),          // 队列中等待的任务数
            executor.getCompletedTaskCount(),    // 已完成的任务总数
            executor.isShutdown()                // 线程池是否已关闭
        );
        log.error(msg);                          // 以 ERROR 级别记录日志
 
        // 2. 发送告警通知（钉钉、邮件、短信等）
        AlertService.send("线程池拒绝告警", msg);
 
        // 3. 可选：将任务持久化到数据库或消息队列，后续补偿执行
        // taskBackupService.save(r);
 
        // 4. 最终仍然抛出异常，让调用者感知
        throw new RejectedExecutionException(msg);
    }
}

// ✅ 生产环境线程池配置模板 —— 阿里巴巴规范推荐写法
ThreadPoolExecutor bizPool = new ThreadPoolExecutor(
    Runtime.getRuntime().availableProcessors() * 2,      // corePoolSize: IO密集型推荐 2N
    Runtime.getRuntime().availableProcessors() * 4,      // maximumPoolSize: 核心的 2 倍弹性
    60L,                                                  // keepAliveTime: 非核心线程空闲 60 秒回收
    TimeUnit.SECONDS,                                     // unit: 秒
    new ArrayBlockingQueue<>(1000),                       // workQueue: 有界队列，容量 1000
    new NamedThreadFactory("biz-service", false),         // threadFactory: 自定义线程名
    new LogAndAlertPolicy()                               // handler: 自定义拒绝策略（日志+告警）
);

public void execute(Runnable command) {
    // 空任务校验，防止 NPE
    if (command == null)
        throw new NullPointerException();
 
    // ctl 是一个 AtomicInteger，高3位存储线程池状态，低29位存储工作线程数
    int c = ctl.get();
 
    // 第一步：判断当前工作线程数是否小于 corePoolSize
    if (workerCountOf(c) < corePoolSize) {
        // addWorker 的第二个参数 true 表示以 corePoolSize 为上限创建线程
        // 如果添加成功，直接返回；该新线程会立即执行传入的 command
        if (addWorker(command, true))
            return;
        // 如果添加失败（可能是并发情况下其他线程抢先创建了），重新读取 ctl
        c = ctl.get();
    }
    // ... 后续逻辑
}

public void execute(Runnable command) {
    int c = ctl.get();
 
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
 
    // 第二步：线程池仍在 RUNNING 状态，且任务成功入队
    if (isRunning(c) && workQueue.offer(command)) {
        // 入队成功后进行二次检查（double-check）
        int recheck = ctl.get();
 
        // 如果线程池在入队期间被 shutdown 了，需要把任务从队列中移除并执行拒绝策略
        if (!isRunning(recheck) && remove(command))
            reject(command);
        // 如果线程池还在运行，但工作线程数变为0（所有线程都意外终止了），
        // 则创建一个空任务的 Worker 来消费队列中的任务，确保不会"饿死"
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // ... 后续逻辑
}

// Worker.runWorker() 的简化逻辑
final void runWorker(Worker w) {
    Runnable task = w.firstTask;   // 首次执行时，取出创建 Worker 时传入的任务
    w.firstTask = null;            // 清空引用，允许 GC
 
    while (task != null || (task = getTask()) != null) {
        // 获取到任务后加锁执行
        w.lock();
        try {
            beforeExecute(w.thread, task);   // 钩子方法：执行前回调
            task.run();                       // 核心：执行任务的 run() 方法
            afterExecute(task, null);         // 钩子方法：执行后回调
        } finally {
            task = null;                      // 释放任务引用
            w.completedTasks++;               // 完成任务计数+1
            w.unlock();                       // 解锁
        }
    }
    // 循环退出意味着 getTask() 返回了 null，线程即将终止
    processWorkerExit(w, false);
}

public void execute(Runnable command) {
    int c = ctl.get();
 
    // 第一步：尝试创建核心线程（略）
    // 第二步：尝试入队列（略）
 
    // 第三步：队列已满，尝试以 maximumPoolSize 为上限创建非核心线程
    // addWorker 的第二个参数 false 表示以 maximumPoolSize 为上限
    else if (!addWorker(command, false))
        // 如果创建也失败了（已达最大线程数或线程池已关闭），执行拒绝策略
        reject(command);
}

private Runnable getTask() {
    boolean timedOut = false;
 
    for (;;) {
        int c = ctl.get();
        int wc = workerCountOf(c);
 
        // 判断当前线程是否需要超时回收
        // 两种情况需要超时：
        // 1. allowCoreThreadTimeOut 为 true（核心线程也要超时回收）
        // 2. 当前工作线程数 > corePoolSize（存在非核心线程）
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
        // 如果上一轮 poll 超时了且满足回收条件，返回 null 触发线程退出
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;  // 返回 null，runWorker 的 while 循环退出，线程终止
            continue;
        }
 
        try {
            // 关键分叉点：
            // timed=true  → poll()，超时返回 null
            // timed=false → take()，无限阻塞直到有任务
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
 
            if (r != null)
                return r;      // 获取到任务，返回给 runWorker 执行
            timedOut = true;   // poll 超时，标记为 true，下轮循环决定是否退出
        } catch (InterruptedException retry) {
            timedOut = false;  // 被中断不算超时，重试
        }
    }
}

// addWorker 返回 false，进入 reject 方法
final void reject(Runnable command) {
    // handler 就是构造线程池时传入的 RejectedExecutionHandler
    handler.rejectedExecution(command, this);
}

public void execute(Runnable command) {
    // 【校验】提交的任务不能为 null
    if (command == null)
        throw new NullPointerException();
 
    // 【读取】获取线程池的控制状态字（高3位=状态，低29位=线程数）
    int c = ctl.get();
 
    // ==================== 第一步 ====================
    // 如果当前工作线程数 < corePoolSize，尝试创建核心线程
    if (workerCountOf(c) < corePoolSize) {
        // addWorker(task, true)：true 表示用 corePoolSize 作为上限
        // 创建成功则直接返回，新线程会立即执行 command
        if (addWorker(command, true))
            return;
        // 创建失败（并发竞争或线程池状态变更），重新读取 ctl
        c = ctl.get();
    }
 
    // ==================== 第二步 ====================
    // 线程池处于 RUNNING 状态，且任务成功入队
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 二次检查：入队期间线程池可能被 shutdown
        if (!isRunning(recheck) && remove(command))
            // 线程池已关闭，移除刚入队的任务并拒绝
            reject(command);
        else if (workerCountOf(recheck) == 0)
            // 线程池还在运行但没有活跃线程了（极端情况），
            // 创建一个空任务的 Worker 来消费队列
            addWorker(null, false);
    }
    // ==================== 第三步 & 第四步 ====================
    // 入队失败（队列已满），尝试以 maximumPoolSize 为上限创建非核心线程
    else if (!addWorker(command, false))
        // 创建也失败（已达最大线程数或池已关闭），执行拒绝策略
        reject(command);
}

// LinkedBlockingQueue 核心结构（简化）
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E> {
 
    // 内部链表节点
    static class Node<E> {
        E item;           // 存储的元素
        Node<E> next;     // 指向下一个节点
    }
 
    private final int capacity;          // 队列容量上限
    private final AtomicInteger count;   // 当前元素数量（原子变量，双锁场景下保证可见性）
 
    transient Node<E> head;              // 哨兵头节点（dummy head）
    private transient Node<E> last;      // 尾节点
 
    private final ReentrantLock takeLock; // 出队锁（消费者侧）
    private final ReentrantLock putLock;  // 入队锁（生产者侧）
 
    private final Condition notEmpty;    // 队列非空条件（与 takeLock 关联）
    private final Condition notFull;     // 队列非满条件（与 putLock 关联）
}

// 无参构造 —— 容量为 Integer.MAX_VALUE，近似"无界"
BlockingQueue<Runnable> q1 = new LinkedBlockingQueue<>();
// 等价于：new LinkedBlockingQueue<>(Integer.MAX_VALUE)
 
// 有参构造 —— 指定容量上限，变为"有界"
BlockingQueue<Runnable> q2 = new LinkedBlockingQueue<>(1000);

                    ⚠️ OOM 场景模拟
                    
生产速率: 10,000 tasks/sec
消费速率:    500 tasks/sec  (corePoolSize=5, 每任务 10ms)
 
每秒净积压: 9,500 tasks
每个 Node 对象约占: ~48 bytes (对象头16 + item引用8 + next引用8 + padding)
加上 Runnable 对象本身: ~100-200 bytes/task
 
1分钟积压: 9,500 × 60 = 570,000 tasks ≈ 57-114 MB
1小时积压: 9,500 × 3,600 = 34,200,000 tasks ≈ 3.4-6.8 GB → 💥 OOM!

// Executors.newFixedThreadPool 源码
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(
        nThreads,                          // corePoolSize
        nThreads,                          // maximumPoolSize（与 core 相同）
        0L, TimeUnit.MILLISECONDS,         // 无空闲线程回收
        new LinkedBlockingQueue<Runnable>() // ⚠️ 无界队列！
    );
}

// ✅ 推荐：指定容量
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    4,                                        // corePoolSize
    8,                                        // maximumPoolSize
    60L, TimeUnit.SECONDS,                    // 空闲线程60秒回收
    new LinkedBlockingQueue<>(2000),          // 有界！最多暂存2000个任务
    new ThreadPoolExecutor.CallerRunsPolicy() // 队列满+线程满 → 调用者自己执行
);

// ArrayBlockingQueue 核心结构（简化）
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E> {
 
    final Object[] items;       // 定长数组，存储队列元素
    int takeIndex;              // 下一个出队位置（消费者指针）
    int putIndex;               // 下一个入队位置（生产者指针）
    int count;                  // 当前元素数量
 
    final ReentrantLock lock;   // 唯一的一把锁（单锁设计）
    private final Condition notEmpty;  // 队列非空条件
    private final Condition notFull;   // 队列非满条件
}

ArrayBlockingQueue(capacity=6) 内部环形数组
 
初始状态（空队列）:
┌─────┬─────┬─────┬─────┬─────┬─────┐
│     │     │     │     │     │     │
└─────┴─────┴─────┴─────┴─────┴─────┘
  ↑ takeIndex=0
  ↑ putIndex=0
  count=0
 
入队3个任务后:
┌─────┬─────┬─────┬─────┬─────┬─────┐
│ T1  │ T2  │ T3  │     │     │     │
└─────┴─────┴─────┴─────┴─────┴─────┘
  ↑ takeIndex=0       ↑ putIndex=3
  count=3
 
出队1个任务后（T1 被 Worker 取走）:
┌─────┬─────┬─────┬─────┬─────┬─────┐
│     │ T2  │ T3  │     │     │     │
└─────┴─────┴─────┴─────┴─────┴─────┘
        ↑ takeIndex=1  ↑ putIndex=3
  count=2
 
继续入队4个（putIndex 回绕到数组开头 —— Ring!）:
┌─────┬─────┬─────┬─────┬─────┬─────┐
│ T7  │ T2  │ T3  │ T4  │ T5  │ T6  │
└─────┴─────┴─────┴─────┴─────┴─────┘
        ↑ takeIndex=1
        ↑ putIndex=1  (追上 takeIndex → 队列满！)
  count=6  →  offer() 返回 false!

// ✅ 生产环境推荐配置示例
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    Runtime.getRuntime().availableProcessors(),  // corePoolSize = CPU核心数
    Runtime.getRuntime().availableProcessors() * 2, // maximumPoolSize
    60L, TimeUnit.SECONDS,                       // 空闲线程60秒回收
    new ArrayBlockingQueue<>(500),               // 有界队列，容量500
    new ThreadPoolExecutor.CallerRunsPolicy()    // 背压策略：调用者自己跑
);

// fair=true：等待时间最长的线程优先获取锁
BlockingQueue<Runnable> fairQueue = new ArrayBlockingQueue<>(500, true);
 
// fair=false（默认）：非公平模式，性能更高，但可能导致线程饥饿
BlockingQueue<Runnable> unfairQueue = new ArrayBlockingQueue<>(500, false);

// 非公平模式（默认，性能更优）
SynchronousQueue<Runnable> sq1 = new SynchronousQueue<>();
 
// 公平模式（FIFO 匹配顺序）
SynchronousQueue<Runnable> sq2 = new SynchronousQueue<>(true);

// Executors.newCachedThreadPool 源码
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(
        0,                          // corePoolSize = 0（无核心线程）
        Integer.MAX_VALUE,          // maximumPoolSize = 无上限 ⚠️
        60L, TimeUnit.SECONDS,      // 空闲线程60秒回收
        new SynchronousQueue<Runnable>() // 直接传递队列
    );
}

// ✅ 安全使用 SynchronousQueue 的推荐方式
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    4,                              // 有基础核心线程
    32,                             // 合理上限，而非 Integer.MAX_VALUE
    30L, TimeUnit.SECONDS,          // 空闲30秒回收
    new SynchronousQueue<>(),       // 直接传递
    new ThreadPoolExecutor.CallerRunsPolicy() // 兜底策略
);

// PriorityBlockingQueue 核心结构（简化）
public class PriorityBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E> {
 
    private transient Object[] queue;    // 存储元素的数组（二叉堆）
    private transient int size;          // 当前元素数量
 
    private transient Comparator<? super E> comparator; // 自定义比较器（可选）
 
    private final ReentrantLock lock;    // 单锁
    private final Condition notEmpty;    // 队列非空条件
 
    // ⚠️ 注意：没有 notFull 条件！因为它是无界的，offer() 永远成功
}

                     PriorityBlockingQueue 二叉堆（小顶堆示例）
 
数组下标:    [0]    [1]    [2]    [3]    [4]    [5]    [6]
元素优先级:   1      3      2      7      5      4      6
 
逻辑树结构:
                        1 (index=0)
                      /              \
                 3 (index=1)       2 (index=2)
                /        \        /        \
          7 (index=3)  5 (index=4)  4 (index=5)  6 (index=6)
 
父节点 index = (childIndex - 1) / 2
左子节点 index = parentIndex * 2 + 1
右子节点 index = parentIndex * 2 + 2
 
出队时：取走堆顶 [0]（优先级最高），将末尾元素移到堆顶，执行 siftDown 下沉调整
入队时：将新元素放到数组末尾，执行 siftUp 上浮调整

// 定义带优先级的任务
public class PriorityTask implements Runnable, Comparable<PriorityTask> {
 
    private final int priority;       // 优先级数值，越小优先级越高
    private final String taskName;    // 任务名称
 
    // 构造方法
    public PriorityTask(int priority, String taskName) {
        this.priority = priority;     // 初始化优先级
        this.taskName = taskName;     // 初始化任务名
    }
 
    @Override
    public void run() {
        // 任务执行逻辑
        System.out.println(Thread.currentThread().getName()
            + " 执行任务: " + taskName
            + " [优先级=" + priority + "]");
    }
 
    @Override
    public int compareTo(PriorityTask other) {
        // 优先级数值小的排前面（小顶堆）
        return Integer.compare(this.priority, other.priority);
    }
}

// 创建使用优先级队列的线程池
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    2,                                       // corePoolSize = 2
    2,                                       // maximumPoolSize = 2（与 core 相同）
    0L, TimeUnit.MILLISECONDS,               // 不回收核心线程
    new PriorityBlockingQueue<>()            // 无界优先级队列
);
 
// 提交不同优先级的任务
executor.execute(new PriorityTask(5, "低优先级-清理日志"));      // 优先级 5（最低）
executor.execute(new PriorityTask(1, "高优先级-用户支付"));      // 优先级 1（最高）
executor.execute(new PriorityTask(3, "中优先级-发送通知"));      // 优先级 3（中等）
executor.execute(new PriorityTask(2, "较高优先级-订单处理"));    // 优先级 2（较高）
executor.execute(new PriorityTask(4, "较低优先级-数据同步"));    // 优先级 4（较低）
 
// 当2个核心线程都忙时，后续任务进入 PriorityBlockingQueue
// Worker 线程取任务时，堆顶（优先级最高）的任务最先被取走

public class PriorityTask implements Runnable, Comparable<PriorityTask> {
 
    // 全局递增序列号，用于打破同优先级的排序歧义
    private static final AtomicLong SEQ_GENERATOR = new AtomicLong(0);
 
    private final int priority;          // 任务优先级
    private final long sequenceNumber;   // 提交顺序序列号
    private final Runnable actualTask;   // 实际要执行的任务
 
    public PriorityTask(int priority, Runnable actualTask) {
        this.priority = priority;                         // 设置优先级
        this.sequenceNumber = SEQ_GENERATOR.getAndIncrement(); // 原子递增，获取唯一序列号
        this.actualTask = actualTask;                     // 包装实际任务
    }
 
    @Override
    public int compareTo(PriorityTask other) {
        // 首先按优先级排序
        int res = Integer.compare(this.priority, other.priority);
        if (res != 0) return res;                         // 优先级不同，直接返回
        // 优先级相同时，按提交顺序（序列号小的先执行 = FIFO）
        return Long.compare(this.sequenceNumber, other.sequenceNumber);
    }
 
    @Override
    public void run() {
        actualTask.run();                                 // 委托给实际任务执行
    }
}

new ThreadPoolExecutor(
    4, 8, 60, TimeUnit.SECONDS,
    new SynchronousQueue<>(),
    new ThreadPoolExecutor.AbortPolicy()
);

// 拒绝策略的顶层接口
public interface RejectedExecutionHandler {
    /**
     * @param r        被拒绝的任务（即提交者想要执行的 Runnable）
     * @param executor 拒绝该任务的线程池实例本身
     */
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

// ThreadPoolExecutor 的静态内部类
public static class AbortPolicy implements RejectedExecutionHandler {
 
    // 空构造器
    public AbortPolicy() { }
 
    /**
     * 永远直接抛出 RejectedExecutionException
     * 让调用者（提交任务的线程）立即感知到线程池已饱和
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 异常消息中包含了线程池当前的状态快照，方便排查
        throw new RejectedExecutionException(
            "Task " + r.toString() +            // 被拒绝的任务
            " rejected from " + e.toString()     // 线程池状态（线程数、队列大小等）
        );
    }
}

// 使用 AbortPolicy 的最佳实践：捕获异常并做补偿
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    2, 4, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(10)
    // 未传入 handler，默认就是 AbortPolicy
);
 
try {
    pool.execute(() -> {
        // 业务逻辑：处理订单
        processOrder(order);
    });
} catch (RejectedExecutionException ex) {
    // 补偿策略：将任务写入消息队列，等待后续消费重试
    log.warn("线程池饱和，任务被拒绝，转入 MQ 重试: {}", order.getId(), ex);
    messageQueue.send(order);  // 降级处理
}

public static class CallerRunsPolicy implements RejectedExecutionHandler {
 
    public CallerRunsPolicy() { }
 
    /**
     * 如果线程池未关闭，则直接在调用者线程中同步执行该任务
     * 如果线程池已关闭（isShutdown），则静默丢弃任务
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            // 关键：在当前线程（即调用 execute/submit 的线程）中直接 run
            r.run();  // 注意是 run() 而非 start()，不会创建新线程
        }
        // 如果线程池已 shutdown，任务会被静默丢弃
    }
}

// CallerRunsPolicy 使用示例
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    2,                                    // 核心线程数
    4,                                    // 最大线程数
    60, TimeUnit.SECONDS,                 // 空闲线程存活 60 秒
    new ArrayBlockingQueue<>(100),        // 有界队列，容量 100
    new ThreadPoolExecutor.CallerRunsPolicy()  // 拒绝时由调用者执行
);
 
// 模拟批量提交任务
for (int i = 0; i < 1000; i++) {
    final int taskId = i;
    pool.execute(() -> {
        // 当线程池饱和时，这段代码会在 main 线程中执行
        // main 线程被阻塞，自然减慢了任务提交速度
        System.out.println(Thread.currentThread().getName() + " 执行任务: " + taskId);
        sleepQuietly(100);  // 模拟耗时操作
    });
}

public static class DiscardPolicy implements RejectedExecutionHandler {
 
    public DiscardPolicy() { }
 
    /**
     * 什么都不做。被拒绝的任务 r 直接被忽略，无声无息地消失。
     * 这是一个名副其实的 "黑洞" 策略。
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 方法体为空——nothing happens
    }
}

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
 
    public DiscardOldestPolicy() { }
 
    /**
     * 丢弃队列中最老的任务（队首），然后将当前任务重新提交到线程池
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            // poll() 移除并返回队列头部元素（最老的任务），如果队列为空则返回 null
            e.getQueue().poll();
            // 重新尝试提交当前任务
            // 注意：如果此时线程池仍然饱和，会再次触发拒绝策略（递归效应）
            e.execute(r);
        }
    }
}

// DiscardOldestPolicy 使用示例：实时报价更新
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    1,                                    // 单核心线程处理报价
    1,                                    // 不扩展线程
    0, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(5),          // 最多缓存 5 个待处理报价
    new ThreadPoolExecutor.DiscardOldestPolicy()  // 新报价挤掉旧报价
);
 
// 行情推送：报价不断涌入，旧的报价自动被新的覆盖
for (double price : incomingPrices) {
    pool.execute(() -> {
        // 更新 UI 显示最新报价
        updateDisplay(price);
    });
}

/**
 * 自定义拒绝策略：记录日志 + 监控上报 + 降级持久化
 * 这是生产环境中最推荐的方式
 */
public class CustomRejectedHandler implements RejectedExecutionHandler {
 
    // 使用 SLF4J 日志框架记录被拒绝的任务信息
    private static final Logger log = LoggerFactory.getLogger(CustomRejectedHandler.class);
 
    // 被拒绝任务的计数器（线程安全），用于监控大盘
    private final AtomicLong rejectedCount = new AtomicLong(0);
 
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // ① 累加拒绝计数，可对接 Prometheus / Grafana 等监控系统
        long count = rejectedCount.incrementAndGet();
 
        // ② 记录详细日志，包含线程池运行时状态快照
        log.warn("[线程池饱和] 第 {} 个任务被拒绝 | 任务: {} | 线程池状态: " +
                 "poolSize={}, activeCount={}, queueSize={}, completedTaskCount={}",
                 count,
                 r.toString(),                         // 任务标识
                 executor.getPoolSize(),                // 当前线程数
                 executor.getActiveCount(),             // 正在执行任务的线程数
                 executor.getQueue().size(),            // 队列中等待的任务数
                 executor.getCompletedTaskCount()       // 已完成的任务总数
        );
 
        // ③ 降级处理：将任务序列化后写入消息队列，等待后续异步重试
        try {
            // 尝试将任务持久化（此处示意，实际实现取决于你的中间件选型）
            fallbackToMessageQueue(r);
        } catch (Exception ex) {
            // 如果降级也失败了，记录 ERROR 级别日志并触发告警
            log.error("[严重] 降级持久化失败，任务可能丢失: {}", r.toString(), ex);
            alertService.sendAlert("线程池拒绝策略降级失败", ex);
        }
    }
 
    /**
     * 降级：将被拒绝的任务投递到消息队列
     */
    private void fallbackToMessageQueue(Runnable r) {
        // 实际项目中可能序列化任务参数，发送到 Kafka / RocketMQ / Redis 等
        // messageProducer.send("rejected-tasks-topic", serialize(r));
        log.info("任务已转入消息队列等待重试: {}", r.toString());
    }
 
    /**
     * 提供 getter 方法，便于外部监控系统拉取指标
     */
    public long getRejectedCount() {
        return rejectedCount.get();
    }
}

// 创建线程池并注入自定义拒绝策略
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    4,                                       // 核心线程数
    8,                                       // 最大线程数
    60, TimeUnit.SECONDS,                    // 非核心线程空闲 60 秒后回收
    new ArrayBlockingQueue<>(200),           // 有界队列，容量 200
    new CustomThreadFactory("biz-pool"),     // 自定义线程工厂（设置线程名）
    new CustomRejectedHandler()              // 自定义拒绝策略
);

/**
 * 带超时阻塞的拒绝策略
 * 在直接拒绝之前，先尝试阻塞等待一段时间让队列腾出空间
 * 适用于短暂突发流量的场景
 */
public class BlockingWithTimeoutPolicy implements RejectedExecutionHandler {
 
    // 最大等待时间
    private final long timeout;
    // 等待时间单位
    private final TimeUnit unit;
 
    public BlockingWithTimeoutPolicy(long timeout, TimeUnit unit) {
        this.timeout = timeout;
        this.unit = unit;
    }
 
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        try {
            // 使用 offer(timeout) 阻塞式入队，等待指定时间
            boolean success = executor.getQueue().offer(r, timeout, unit);
            if (!success) {
                // 等待超时后仍无法入队，抛出异常通知调用者
                throw new RejectedExecutionException(
                    "线程池队列已满，等待 " + timeout + " " + unit + " 后仍无法入队"
                );
            }
            // 入队成功，任务将被 Worker 线程正常消费
        } catch (InterruptedException e) {
            // 等待期间被中断，恢复中断标志并抛出异常
            Thread.currentThread().interrupt();
            throw new RejectedExecutionException("等待入队时被中断", e);
        }
    }
}

new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(100),
    new ThreadPoolExecutor.DiscardPolicy());

// java.util.concurrent.ThreadPoolExecutor.Worker（JDK 源码简化）
private final class Worker
    extends AbstractQueuedSynchronizer  // 继承 AQS，自身就是一把锁
    implements Runnable {               // 实现 Runnable，自身就是一个任务
 
    final Thread thread;     // Worker 持有的真实线程（由 ThreadFactory 创建）
    Runnable firstTask;      // 创建 Worker 时携带的第一个任务（可以为 null）
    volatile long completedTasks; // 该 Worker 累计完成的任务数
 
    Worker(Runnable firstTask) {
        setState(-1);                    // 初始 state = -1，阻止中断直到 runWorker
        this.firstTask = firstTask;      // 记录第一个任务
        this.thread = getThreadFactory().newThread(this); // 把 Worker 自身作为 Runnable 传给线程
    }
 
    public void run() {
        runWorker(this);  // 委托给外部类 ThreadPoolExecutor 的 runWorker 方法
    }
    // ... AQS 相关方法（lock / unlock / tryAcquire / tryRelease）
}

┌─────────────────────────────────────────────────────┐
│                     Worker 对象                      │
│                                                     │
│  ┌──────────────┐   ┌──────────────┐                │
│  │  firstTask   │   │   thread     │──► 真实OS线程   │
│  │  (Runnable)  │   │  (Thread)    │                │
│  └──────────────┘   └──────────────┘                │
│                                                     │
│  ├─ extends AQS     → 自身是一把不可重入锁           │
│  ├─ implements Runnable → 自身是线程的执行体          │
│  └─ completedTasks  → 统计完成任务数                 │
└─────────────────────────────────────────────────────┘

// ThreadPoolExecutor.interruptIdleWorkers(boolean onlyOne) 简化源码
private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock; // 线程池全局锁
    mainLock.lock();                              // 加全局锁，保护 workers 集合
    try {
        for (Worker w : workers) {                // 遍历所有 Worker
            Thread t = w.thread;                  // 获取 Worker 持有的线程
            if (!t.isInterrupted() && w.tryLock()) { // 尝试获取 Worker 锁
                // tryLock 成功 → Worker 当前空闲（没有在执行任务）
                try {
                    t.interrupt();                // 中断该空闲线程
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();                   // 释放 Worker 锁
                }
            }
            if (onlyOne)                          // 如果只中断一个，直接跳出
                break;
        }
    } finally {
        mainLock.unlock();                        // 释放全局锁
    }
}

// Worker 的 AQS 实现（源码简化）
// state 含义：
//   -1 : 初始状态（刚创建，还没开始 runWorker）
//    0 : 未锁定（空闲状态）
//    1 : 已锁定（正在执行任务）
 
protected boolean isHeldExclusively() {
    return getState() != 0;       // state != 0 就认为被独占持有
}
 
protected boolean tryAcquire(int unused) {
    // CAS 将 state 从 0 改为 1，只有从 0 才能成功 → 不可重入！
    if (compareAndSetState(0, 1)) {
        setExclusiveOwnerThread(Thread.currentThread()); // 记录持有者线程
        return true;              // 加锁成功
    }
    return false;                 // 加锁失败（state 已经是 1，不可重入）
}
 
protected boolean tryRelease(int unused) {
    setExclusiveOwnerThread(null); // 清除持有者
    setState(0);                   // 恢复为未锁定
    return true;
}
 
public void lock()        { acquire(1); }      // 阻塞式加锁
public boolean tryLock()  { return tryAcquire(1); } // 非阻塞尝试加锁
public void unlock()      { release(1); }      // 释放锁

// 传统方式：每个任务创建一个线程
new Thread(() -> doSomething()).start(); // 任务执行完，线程死亡

// 线程池方式：Worker 充当 Runnable，内部循环获取多个任务
Thread t = threadFactory.newThread(worker); // worker 就是那个 Runnable
t.start(); // 线程启动后，执行 worker.run() → runWorker(this)
// runWorker 内部有个 while 循环，不断取任务执行
// 线程不会在一个任务完成后死亡！

传统方式：
  Thread-1 ──► Task-A ──► 线程销毁 ✗
  Thread-2 ──► Task-B ──► 线程销毁 ✗
  Thread-3 ──► Task-C ──► 线程销毁 ✗
 
线程池方式（Worker 模型）：
  Worker-1.thread ──► Task-A ──► Task-D ──► Task-G ──► ... (循环复用)
  Worker-2.thread ──► Task-B ──► Task-E ──► Task-H ──► ... (循环复用)
  Worker-3.thread ──► Task-C ──► Task-F ──► Task-I ──► ... (循环复用)

public void run() {
    runWorker(this); // 把自身传给 ThreadPoolExecutor 的 runWorker 方法
}

// ThreadPoolExecutor.runWorker() 核心源码（带详尽注释）
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();  // 获取当前线程（即 Worker 持有的 thread）
    Runnable task = w.firstTask;         // 取出 Worker 携带的第一个任务
    w.firstTask = null;                  // 清空引用，帮助 GC
    w.unlock();                          // ★ 将 state 从 -1 设为 0，允许被中断
    boolean completedAbruptly = true;    // 标记是否因异常而退出
 
    try {
        // ★★★ 核心循环：线程复用的关键 ★★★
        // 第一次循环：执行 firstTask（如果有的话）
        // 后续循环：通过 getTask() 从工作队列中获取新任务
        while (task != null || (task = getTask()) != null) {
            w.lock();                    // 加锁，标记 Worker 为"忙碌"状态（state=1）
 
            // 检查线程池状态：如果线程池正在 STOP，确保线程被中断
            // 如果线程池还没 STOP，确保线程没有被中断
            if ((runStateAtLeast(ctl.get(), STOP)          // 线程池 >= STOP？
                    || (Thread.interrupted()                // 或者线程被中断了？
                        && runStateAtLeast(ctl.get(), STOP))) // 再次确认是否 >= STOP
                && !wt.isInterrupted())                     // 且线程还未设置中断标志
                wt.interrupt();                             // 那就中断它
 
            try {
                beforeExecute(wt, task);  // 钩子方法：任务执行前（可覆写）
                try {
                    task.run();           // ★ 直接调用 run()，不是 start()！
                                          // 任务在 Worker 线程中同步执行
                } 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++;       // 已完成任务数 +1
                w.unlock();              // 释放锁，标记 Worker 为"空闲"状态（state=0）
            }
        }
        // while 循环结束 → getTask() 返回了 null → Worker 该退出了
        completedAbruptly = false;        // 正常退出，不是因为异常
    } finally {
        processWorkerExit(w, completedAbruptly); // 清理工作：从 workers 集合移除等
    }
}

// ThreadPoolExecutor.getTask() 核心源码（带详尽注释）
private Runnable getTask() {
    boolean timedOut = false;            // 上一次 poll() 是否超时
 
    for (;;) {                           // 自旋循环
        int c = ctl.get();               // 获取线程池控制状态
 
        // 检查线程池状态：
        // 如果是 SHUTDOWN 且队列为空 → 返回 null（Worker 该退出了）
        // 如果是 STOP 或更高状态 → 返回 null（不管队列里还有没有）
        if (runStateAtLeast(c, SHUTDOWN)
            && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
            decrementWorkerCount();      // 工作线程数 -1
            return null;                 // 返回 null → runWorker 的 while 循环结束 → Worker 退出
        }
 
        int wc = workerCountOf(c);       // 当前工作线程数
 
        // ★ 判断是否需要超时控制 ★
        // allowCoreThreadTimeOut=true → 核心线程也会超时
        // wc > corePoolSize → 当前线程数超过核心数，多出来的是"非核心线程"
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
        // 如果线程数超标或者上次已经超时 → 尝试减少线程数并退出
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c)) // CAS 减少线程数
                return null;             // 返回 null → Worker 退出
            continue;                    // CAS 失败，重新循环
        }
 
        try {
            // ★★ 关键分支 ★★
            // timed=true → 用 poll(keepAliveTime)，超时返回 null
            // timed=false → 用 take()，永久阻塞直到有任务
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : // 超时等待
                workQueue.take();        // 无限等待（核心线程默认走这里）
 
            if (r != null)
                return r;                // 成功获取任务，返回给 runWorker 的 while 循环
            timedOut = true;             // poll 超时了，下一轮循环会尝试退出
        } catch (InterruptedException retry) {
            timedOut = false;            // 被中断了，重置超时标记，重新循环
        }
    }
}

// 线程复用的极简模型
void runWorker() {
    Runnable task;
    // 不断从队列取任务 → 线程不退出 → 线程被复用
    while ((task = getTaskFromQueue()) != null) {
        task.run();  // 直接 run()，不创建新线程
    }
    // getTask 返回 null → 跳出循环 → 线程自然死亡 → 被 GC 回收
}

用户调用 executor.execute(taskA)
    │
    ├─► addWorker(taskA, true)           // 创建新 Worker
    │       │
    │       ├─► new Worker(taskA)        // Worker 构造：firstTask=taskA, state=-1
    │       │       │
    │       │       └─► threadFactory.newThread(worker)  // 创建线程，Runnable=worker
    │       │
    │       └─► worker.thread.start()    // 启动线程 → JVM 回调 Worker.run()
    │               │
    │               └─► runWorker(worker)
    │                       │
    │                       ├─► 第 1 轮循环：task = firstTask(taskA) → taskA.run()
    │                       ├─► 第 2 轮循环：task = getTask() → taskB.run()
    │                       ├─► 第 3 轮循环：task = getTask() → taskC.run()
    │                       ├─► ...（持续复用）
    │                       └─► getTask() 返回 null → 退出循环 → processWorkerExit()
    │
    └─► 或者：workQueue.offer(taskA)     // 队列未满时，任务入队等待被 getTask() 取走

// ThreadPoolExecutor 源码中的 ctl 定义
// ctl 是一个 AtomicInteger，高 3 位存储 runState，低 29 位存储 workerCount
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
 
// Integer.SIZE = 32，因此 COUNT_BITS = 29
// 也就是说，用 29 位来表示线程数量（最大约 5 亿个线程，远超实际需要）
private static final int COUNT_BITS = Integer.SIZE - 3;  // 29
 
// 线程数量的掩码：低 29 位全为 1 → 0001_1111_1111_1111_1111_1111_1111_1111
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
 
// ————— 五种状态常量（高 3 位编码）—————
 
// RUNNING:  111_00000...（高 3 位为 111，是负数，值最小）
private static final int RUNNING    = -1 << COUNT_BITS;
 
// SHUTDOWN: 000_00000...（高 3 位为 000）
private static final int SHUTDOWN   =  0 << COUNT_BITS;
 
// STOP:     001_00000...（高 3 位为 001）
private static final int STOP       =  1 << COUNT_BITS;
 
// TIDYING:  010_00000...（高 3 位为 010）
private static final int TIDYING    =  2 << COUNT_BITS;
 
// TERMINATED: 011_00000...（高 3 位为 011）
private static final int TERMINATED =  3 << COUNT_BITS;
 
// ————— 解包方法 —————
 
// 获取运行状态：ctl 与 ~COUNT_MASK 按位与，只保留高 3 位
private static int runStateOf(int c)     { return c & ~COUNT_MASK; }
 
// 获取工作线程数：ctl 与 COUNT_MASK 按位与，只保留低 29 位
private static int workerCountOf(int c)  { return c & COUNT_MASK; }
 
// 打包方法：将 runState 和 workerCount 合并成一个 int
private static int ctlOf(int rs, int wc) { return rs | wc; }

// 源码中大量使用这种比较方式，非常简洁高效
// 例如：判断线程池是否不在 RUNNING 状态
private static boolean isRunning(int c) {
    return c < SHUTDOWN;  // RUNNING 是负数，小于 0（SHUTDOWN）
}
 
// 判断是否至少已经到了 SHUTDOWN
// runStateAtLeast(c, SHUTDOWN) → runStateOf(c) >= SHUTDOWN