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

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

private static final int COUNT_BITS = 9;                  // 2^29 = 5 亿，线程数上限
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// 高 3 位的状态值，按数值大小排好序，isRunning(c) 这类判断才成立
private static final int RUNNING    = -1 << COUNT_BITS;  // 正常运行，接受新任务
private static final int SHUTDOWN   =  0 << COUNT_BITS;  // 已 shutdown，不接新任务，处理存量
private static final int STOP       =  1 << COUNT_BITS;  // 已 stop，中断运行中任务，丢弃队列
private static final int TIDYING    =  2 << COUNT_BITS;  // 所有线程已退出，workerCount == 0
private static final int TERMINATED =  3 << COUNT_BITS;  // terminated() 钩子已执行

private static int workerCountOf(int c)  { return c & CAPACITY; }  // 取低 29 位 = 线程数
private static boolean isRunning(int c)  { return c < SHUTDOWN; }   // 高 3 位 < 000 只能是 111

// ① execute：异常从工作线程"飞"出来
pool.execute(() -> {
    int x = 1 / 0;   // ArithmeticException
});
// 提交线程完全无感。异常处理路径：
// 工作线程死亡 → UncaughtExceptionHandler 打印堆栈（默认打到 stderr）
// → 线程池再创建一个新工作线程替补。
// 如果没人处理 UncaughtExceptionHandler，异常就"蒸发"了——不告警、不记录，
// 任务失败这件事只有翻日志才知道。

// ② submit：异常被"吞"进 Future
Future<String> f = pool.submit(() -> {
    int x = 1 / 0;
    return "ok";
});
// f.isDone() == true，看起来任务"正常完成"了！
// 异常被包装成 ExecutionException 存在 Future 里，
// 只有调用 f.get() 才会抛出来。
// 如果业务代码从不 get()（比如 fire-and-forget 场景），
// 这个异常就永远不可见。

// 兜底 1：任务体内永远 try-catch（最可靠，两种提交都管用）
pool.execute(() -> {
    try {
        doBusiness();
    } catch (Exception e) {
        log.error("订单任务执行失败, orderId={}", orderId, e);
        metrics.inc("task.failed");  // 打点告警
    }
});

// 兜底 2：ThreadFactory 挂 UncaughtExceptionHandler（抓 execute 的漏网之鱼）
new ThreadFactoryBuilder()
    .setNameFormat("order-pool-%d")
    .setUncaughtExceptionHandler((t, e) ->
        log.error("线程 {} 未捕获异常", t.getName(), e))
    .build()

// 兜底 3：submit 的结果要么 get/join，要么 exceptionally 处理
Future<String> f = pool.submit(task);
f.get(5, TimeUnit.SECONDS);  // 带超时，顺便防住任务卡死

public class NamedThreadFactory implements ThreadFactory {
    private final AtomicInteger seq = new AtomicInteger(1);
    private final String prefix;

    public NamedThreadFactory(String prefix) {
        this.prefix = prefix;
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r, prefix + "-" + seq.getAndIncrement());
        t.setDaemon(false);   // 业务池建议非守护：JVM 退出前把任务跑完
        return t;
    }
}

// 默认：核心线程永生
pool.allowCoreThreadTimeOut(true);
// 之后核心线程空闲超过 keepAliveTime 也会被回收。
// 适用：流量波动剧烈的服务——白天 40 个线程忙死，凌晨 1 个就够，
//      回收核心线程能省下几十 MB 栈内存（每线程默认 1MB 栈）。
// 代价：流量突然回升时要重新创建线程（毫秒级，通常可接受）。

// 单个预热：创建一个空闲核心线程
pool.prestartCoreThread();

// 全部预热：把 corePoolSize 个核心线程一次性建好
int started = pool.prestartAllCoreThreads();
// 返回本次真正新建的线程数。放在 @PostConstruct / 应用启动钩子里，
// 避免第一个用户为你的线程池付「建线程」的成本。

// 陷阱 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<>()         // 没有缓冲，每个任务必须分配线程
    );
}

/*
 * 场景：订单查询服务，每次请求要调 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={}, largest={}, queue={}, completed={}",
        pool.getActiveCount(),         // 正在执行任务的线程数
        pool.getPoolSize(),             // 当前池中线程总数
        pool.getLargestPoolSize(),      // 历史峰值线程数（评估 max 设得合不合理）
        pool.getQueue().size(),          // 队列中待执行的任务数
        pool.getCompletedTaskCount()    // 已完成的任务总数
    );
}, 0, 10, TimeUnit.SECONDS);

public class PersistingRejectedHandler implements RejectedExecutionHandler {

    private final LongAdder rejectedCount = new LongAdder();
    private final TaskFallbackStore store;   // 落库 / 发 MQ 都行

    public PersistingRejectedHandler(TaskFallbackStore store) {
        this.store = store;
    }

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        rejectedCount.increment();
        log.warn("任务被拒绝: task={}, active={}, queueSize={}",
                r, e.getActiveCount(), e.getQueue().size());
        store.save(r);   // 持久化，后台线程定时捞出重放
    }

    // 暴露给监控系统（Micrometer / 自采）
    public long rejectedSum() {
        return rejectedCount.sum();
    }
}

// 阶段一：stop accepting（RUNNING → SHUTDOWN）
// 不再接受新任务；队列里 + 执行中的任务继续跑完
pool.shutdown();

// 阶段二：给存量任务一个处理窗口
if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
    // 30s 还没跑完 → 中断运行中线程 + 取出未执行任务（→ STOP）
    List<Runnable> dropped = pool.shutdownNow();
    if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
        log.error("线程池强制关闭失败，仍有线程存活");
    }
    // shutdownNow() 返回的队列任务要自己兜底（落盘补偿）
    dropped.forEach(fallbackStore::save);
}
// 全部线程退出后 → TIDYING → TERMINATED，terminated() 钩子执行

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;

        ThreadPoolExecutor pool = 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()
        );
        pool.prestartAllCoreThreads();   // 预热，避免首请求付建线程成本
        return pool;
    }

    public static void shutdownGracefully(ThreadPoolExecutor pool) {
        pool.shutdown();
        if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
            pool.shutdownNow();
        }
    }
}