线程池 ThreadPoolExecutor：7 大参数与 4 种拒绝策略

// 当时的"合理"配置
ExecutorService pool = new ThreadPoolExecutor(
    10,                       // corePoolSize
    10,                       // maximumPoolSize —— 和 core 一样！
    0L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>() // 无界队列！默认容量 Integer.MAX_VALUE
);

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();                        // ctl 是 AtomicInteger，高3位=状态，低29位=线程数
    int workerCount = workerCountOf(c);

    // 第一步：线程数 < corePoolSize → 创建核心线程
    if (workerCount < corePoolSize) {
        if (addWorker(command, true))         // 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)
            addWorker(null, false);           // 兜底：确保至少有一个线程消费队列
        return;
    }

    // 第三步：队列满 → 创建非核心线程
    if (!addWorker(command, false))           // false = 创建非核心线程
        reject(command);                       // 第四步：线程也满了 → 拒绝
}

// 推荐：有界队列 + CallerRunsPolicy = 天然背压
new ThreadPoolExecutor(
    20,                                       // core
    50,                                       // max
    60L, TimeUnit.SECONDS,                    // 非核心线程空闲 60s 回收
    new ArrayBlockingQueue<>(1000),          // 有界队列，容量 1000
    new ThreadFactoryBuilder()                // Guava 工具
        .setNameFormat("order-pool-%d")
        .setDaemon(true)
        .build(),
    new ThreadPoolExecutor.CallerRunsPolicy()  // 队列满 + 线程满 → 调用者线程自己跑
);

// 1. AbortPolicy（默认）—— 直接抛异常
public static class AbortPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        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();    // 在调用者线程直接执行，不抛异常，不丢任务
        }
    }
}

// 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);          // 重新尝试提交当前任务
        }
    }
}

/*
 * 场景：订单查询服务，每次请求要调 3 个 RPC + 2 次 DB 查询
 * 机器：8 核
 * 观测：单次请求中 CPU 计算约 10ms，IO 等待约 40ms
 *       → W/C = 40/10 = 4
 *
 * 套用公式：corePoolSize = 8 × (1 + 4) = 40
 */

int cpuCores  = Runtime.getRuntime().availableProcessors();  // 8
double wcRatio = 4.0;  // 等待时间 / 计算时间
int coreSize  = (int) (cpuCores * (1 + wcRatio));            // 40
int maxSize   = coreSize * 2;                                 // 80，留弹性空间

ThreadPoolExecutor pool = new ThreadPoolExecutor(
    coreSize,
    maxSize,
    60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(2000),
    new ThreadFactoryBuilder().setNameFormat("order-q-%d").build(),
    new ThreadPoolExecutor.CallerRunsPolicy()
);

// 定时采集，上报到 Prometheus / Grafana
ScheduledExecutorService monitor = ...;
monitor.scheduleAtFixedRate(() -> {
    log.info("[pool-monitor] active={}, poolSize={}, queue={}, completed={}",
        pool.getActiveCount(),         // 正在执行任务的线程数
        pool.getPoolSize(),             // 当前池中线程总数
        pool.getQueue().size(),          // 队列中待执行的任务数
        pool.getCompletedTaskCount()    // 已完成的任务总数
    );
}, 0, 10, TimeUnit.SECONDS);

// 陷阱 1：newFixedThreadPool —— 无界队列 → OOM
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(
        nThreads, nThreads,
        0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<>()   // ← 无界！任务堆积 → OOM
    );
}

// 陷阱 2：newSingleThreadExecutor —— 同样的无界队列
public static ExecutorService newSingleThreadExecutor() {
    return new ThreadPoolExecutor(
        1, 1,
        0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<>()   // ← 无界！
    );
}

// 陷阱 3：newCachedThreadPool —— 无限线程 → 线程爆炸
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(
        0, Integer.MAX_VALUE,             // ← 最大线程数 21 亿！
        60L, TimeUnit.SECONDS,
        new SynchronousQueue<>()         // 没有缓冲，每个任务必须分配线程
    );
}

// shutdown() —— 温和关闭
// 1. 不再接受新任务
// 2. 已提交的任务（队列中 + 执行中）继续执行完毕
// 3. 所有任务完成后，线程自然退出
pool.shutdown();
if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
    pool.shutdownNow();  // 30s 还没关完 → 强制中断
}

// shutdownNow() —— 暴力关闭
// 1. 不再接受新任务
// 2. 尝试中断所有正在执行的线程（Thread.interrupt()）
// 3. 返回队列中尚未执行的任务列表
List<Runnable> pending = pool.shutdownNow();

public class ThreadPoolTemplate {

    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();

    public static ThreadPoolExecutor create(String name, boolean ioBound) {
        int core = ioBound
            ? (int) (CPU_COUNT * (1 + 4.0))   // IO 密集，假设 W/C = 4
            : CPU_COUNT + 1;                    // CPU 密集
        int max  = core * 2;
        int queueCap = ioBound ? 2000 : 500;

        return new ThreadPoolExecutor(
            core, max,
            60L, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(queueCap),
            new ThreadFactoryBuilder()
                .setNameFormat(name + "-%d")
                .setUncaughtExceptionHandler((t, e) ->
                    log.error("Thread {} threw exception", t.getName(), e))
                .build(),
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }
}