import java.util.concurrent.atomic.AtomicInteger;

/**
 * 反例：无限制创建线程，最终 OOM。
 * 运行前请先保存好工作，可能会拖慢系统。
 * 常见报错：
 *   java.lang.OutOfMemoryError: unable to create new native thread
 *   java.lang.OutOfMemoryError: Java heap space
 */
public class ThreadBombDemo {
    public static void main(String[] args) {
        AtomicInteger counter = new AtomicInteger();
        while (true) {
            new Thread(() -> {
                try {
                    // 每个线程活 60 秒，模拟"线程还没销毁，新的又来了"
                    Thread.sleep(60_000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "bomb-" + counter.incrementAndGet()).start();

            if (counter.get() % 500 == 0) {
                System.out.println("已创建线程数：" + counter.get());
            }
        }
    }
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

// ThreadPoolExecutor.execute 的三步走（简化版，保留关键分支）
public void execute(Runnable command) {
    if (command == null) throw new NullPointerException();
    int c = ctl.get();

    // 第一步：工作线程数 < corePoolSize，直接新建核心线程，把 command 作为 firstTask
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();   // addWorker 失败（并发或状态变化），重新读取
    }

    // 第二步：线程池还在 RUNNING，尝试入队
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 双重检查：入队期间可能发生了 shutdown，或所有工作线程刚好都退出了
        if (!isRunning(recheck) && remove(command))
            reject(command);                        // 已关闭 → 移除并拒绝
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);                 // 没有存活线程 → 补一个空任务线程来消费队列
    }
    // 第三步：入队失败（队列有界且已满）→ 尝试创建非核心线程
    else if (!addWorker(command, false))
        reject(command);                            // 达到 maximumPoolSize → 拒绝
}

// ctl 的位运算：高 3 位 runState，低 29 位 workerCount
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;          // 29
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;     // 00011111...111（低29位全1）

private static final int RUNNING     = -1 << COUNT_BITS;         // 111 开头
private static final int SHUTDOWN    =  0 << COUNT_BITS;         // 000
private static final int STOP        =  1 << COUNT_BITS;         // 001
private static final int TIDYING     =  2 << COUNT_BITS;         // 010
private static final int TERMINATED  =  3 << COUNT_BITS;         // 011

// 取高 3 位：用 CAPACITY 的反码做掩码
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// 取低 29 位
private static int workerCountOf(int c)  { return c & CAPACITY; }
// 合成：状态在高位、数量在低位，直接按位或
private static int ctlOf(int rs, int wc) { return rs | wc; }

// 状态比较：c 是否处于 s 之后（数值更大 = 更接近终结）
private static boolean runStateAtLeast(int c, int s) { return c >= s; }
private static boolean isRunning(int c) { return c < SHUTDOWN; }   // 只有 RUNNING 是负数

// 验证 ctl 编解码：动手跑一遍比看十遍注释管用
public class CtlBitsDemo {
    private static final int COUNT_BITS = Integer.SIZE - 3;      // 29
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;

    static int runStateOf(int c)    { return c & ~CAPACITY; }
    static int workerCountOf(int c) { return c & CAPACITY; }
    static int ctlOf(int rs, int wc){ return rs | wc; }

    public static void main(String[] args) {
        int c = ctlOf(RUNNING, 3);
        System.out.println("ctl = " + c + "，二进制 = " + Integer.toBinaryString(c));
        System.out.println("线程数 = " + workerCountOf(c));                 // 3
        System.out.println("是否 RUNNING = " + (runStateOf(c) == RUNNING)); // true
        System.out.println("容量上限 = " + CAPACITY);                       // 536870911

        // 线程数 +1：直接对 ctl 做整数加法，低位进位不会污染高位
        int c2 = c + 1;
        System.out.println("+1 后线程数 = " + workerCountOf(c2) + "，状态不变 = " + (runStateOf(c2) == RUNNING));

        // 状态迁移：改成 SHUTDOWN，线程数保持不变
        int c3 = ctlOf(SHUTDOWN, workerCountOf(c));
        System.out.println("SHUTDOWN 后线程数 = " + workerCountOf(c3) + "，isRunning = " + (c3 < SHUTDOWN));
    }
}

// addWorker 简化版：CAS 增加计数 + 创建 Worker 并启动
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 状态校验：SHUTDOWN 之后不再接收新任务（但 SHUTDOWN 时允许 firstTask==null 去消费队列）
        if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            // 超过容量上限或超过本次的目标边界（core ? corePoolSize : maximumPoolSize）则失败
            if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // CAS 自增 workerCount，成功则跳出外层循环
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();               // CAS 失败，说明有并发，重新读
            if (runStateOf(c) != rs)     // 状态变了，回到外层重新做状态校验
                continue retry;
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = new Worker(firstTask);    // 封装 Thread + firstTask + AQS 锁
    Thread t = w.thread;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();                     // workers 是 HashSet，非线程安全，需要 mainLock 保护
    try {
        if (runStateOf(ctl.get()) < SHUTDOWN || (runStateOf(ctl.get()) == SHUTDOWN && firstTask == null)) {
            workers.add(w);
            workerAdded = true;
        }
    } finally {
        mainLock.unlock();
    }
    if (workerAdded) {
        t.start();                       // 真正启动线程，进入 Worker.run() → runWorker()
        workerStarted = true;
    }
    if (!workerStarted)
        addWorkerFailed(w);              // 回滚：从 workers 移除并 CAS 减少计数
    return workerStarted;
}

// Worker 简化版：包装线程 + 首任务 + 已完成计数；继承 AQS 实现"不可重入独占锁"
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    final Thread thread;          // 真正干活的线程（由 threadFactory 创建，target 为 this）
    Runnable firstTask;           // 创建时顺带执行的第一个任务，可为 null
    volatile long completedTasks; // 该线程完成的任务数，用于统计

    Worker(Runnable firstTask) {
        setState(-1);                        // 初始 state = -1，禁止在 runWorker 之前被中断
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    public void run() { runWorker(this); }   // 线程启动后进入主循环

    // --- AQS：实现不可重入的独占锁，state 0=未锁定 1=已锁定 ---
    protected boolean isHeldExclusively() { return getState() != 0; }

    protected boolean tryAcquire(int unused) {
        // CAS 0→1：不可重入，已在锁定状态再次 acquire 必然失败
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    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); }
    public boolean isLocked() { return isHeldExclusively(); }

    // 中断"正在执行任务"的 worker 之前，先把 state 从 -1 归零，允许中断生效
    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try { t.interrupt(); } catch (SecurityException ignore) {}
        }
    }
}

// runWorker 简化版：循环从队列取任务执行，前后各留一个扩展钩子
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock();                 // state 从 -1 归 0，允许被中断（此后才进入"可被中断"状态）
    boolean completedAbruptly = true;
    try {
        // 先执行 firstTask，之后不断 getTask()；getTask 返回 null 则线程退出、被回收
        while (task != null || (task = getTask()) != null) {
            w.lock();           // 加锁标记"忙"，shutdown 时 tryLock 失败 → 不会被中断
            try {
                // 如果线程池已到 STOP，确保当前线程是中断状态，让任务能感知
                if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP)))
                        && !wt.isInterrupted())
                    wt.interrupt();
                beforeExecute(wt, task);   // 扩展点：可在子类中埋监控
                try {
                    task.run();            // 注意：直接调用 run()，不 start 新线程
                    afterExecute(task, null);
                } catch (Throwable ex) {
                    afterExecute(task, ex);
                    throw ex;              // 抛出 → 线程死亡 → processWorkerExit 补位
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 正常退出或异常退出都会走到这里：从 workers 移除、统计任务数、可能需要补一个线程
        processWorkerExit(w, completedAbruptly);
    }
}

// getTask 简化版：决定线程是阻塞等待、超时等待，还是直接返回 null 被回收
private Runnable getTask() {
    boolean timedOut = false;
    for (;;) {
        int c = ctl.get();
        // 线程池已停止时，把 workerCount 减到 0 并返回 null，让线程退出
        if (runStateAtLeast(c, SHUTDOWN) && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);
        // 关键：是否允许核心线程超时。allowCoreThreadTimeOut=true 时，核心线程也会超时退出
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))   // CAS 减少计数后返回 null → 线程消亡
                return null;
            continue;
        }
        try {
            // 超时等待 poll(keepAlive) vs 无限阻塞 take()
            Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)
                               : workQueue.take();
            if (r != null) return r;
            timedOut = true;                          // 本轮超时，下一轮循环会判定回收
        } catch (InterruptedException retry) {
            timedOut = false;                         // 被中断（多半是 shutdown）→ 重新判断状态
        }
    }
}

import java.util.List;
import java.util.concurrent.*;

/** shutdown 与 shutdownNow 的行为差异 */
public class ShutdownDemo {
    public static void main(String[] args) throws Exception {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                1, 1, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10),
                r -> new Thread(r, "worker-1"));

        // 提交一个 3 秒的长任务 + 3 个队列任务
        pool.execute(() -> sleep(3000));
        for (int i = 0; i < 3; i++) {
            final int n = i;
            pool.execute(() -> System.out.println("执行队列任务 " + n));
        }
        Thread.sleep(100);

        System.out.println("--- shutdownNow：立即停止 ---");
        List<Runnable> dropped = pool.shutdownNow();   // 返回被丢弃的队列任务
        System.out.println("被丢弃未执行的任务数 = " + dropped.size());   // 3
        System.out.println("awaitTermination = " + pool.awaitTermination(5, TimeUnit.SECONDS));

        // 对比：若这里换成 shutdown()，则 3 个队列任务都会被正常执行完
    }

    static void sleep(long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }
}

// 同一个 core/max 配置，仅仅换一个队列，行为天差地别
public class QueueImpactDemo {
    public static void main(String[] args) {
        // A：无界队列 —— maximumPoolSize 永远是摆设，第 3 步永远走不到
        ThreadPoolExecutor unbounded = new ThreadPoolExecutor(
                2, 16, 60, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>());          // 容量 = Integer.MAX_VALUE

        // B：有界队列 —— 队列满 100 之后才会创建第 3~16 个线程，再满则拒绝
        ThreadPoolExecutor bounded = new ThreadPoolExecutor(
                2, 16, 60, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100));       // 显式容量
    }
}

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 生产级拒绝策略：记日志（含线程池运行状态）+ 落库待补偿 + 触发告警
 */
public class AlertingRejectHandler implements RejectedExecutionHandler {
    private final String poolName;
    private final AtomicLong rejectedCount = new AtomicLong();
    // 实际项目中注入 DAO / MQ / 告警客户端
    // private final TaskRepository repository;

    public AlertingRejectHandler(String poolName) { this.poolName = poolName; }

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        long n = rejectedCount.incrementAndGet();
        // 1）日志：带上关键指标，便于事后还原现场
        System.err.printf("[REJECT] pool=%s seq=%d active=%d poolSize=%d queue=%d/%d completed=%d%n",
                poolName, n, e.getActiveCount(), e.getPoolSize(),
                e.getQueue().size(), e.getQueue().remainingCapacity() + e.getQueue().size(),
                e.getCompletedTaskCount());

        // 2）落库 / 投递到 MQ，等待补偿任务重放（保证任务不丢）
        // repository.saveFailedTask(TaskStatus.REJECTED, r.toString());

        // 3）降级：尝试由调用线程执行（等效 CallerRunsPolicy），但要防止拖垮上游
        // 这里做个判断：只在线程池仍在运行时才让调用方兜底
        if (!e.isShutdown()) {
            r.run();
        }

        // 4）告警：按 n % 100 == 1 之类的频率做抑制，避免告警风暴
        if (n % 100 == 1) {
            System.err.println("[ALERT] 线程池 " + poolName + " 已拒绝 " + n + " 个任务，请扩容或限流！");
        }
    }
}

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/** 生产环境线程池创建标准模板 */
public class PoolFactory {
    public static ThreadPoolExecutor create(String bizName, int core, int max,
                                            int queueCapacity, long keepAliveSec) {
        // 1）自定义 ThreadFactory：命名 + 非守护 + 异常处理器 + 明确栈大小
        ThreadFactory factory = new ThreadFactory() {
            private final AtomicInteger idx = new AtomicInteger(1);
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r, bizName + "-pool-" + idx.getAndIncrement());
                t.setDaemon(false);                 // 业务池必须非守护，否则 JVM 退出会丢任务
                t.setUncaughtExceptionHandler((th, e) ->
                        System.err.println("线程 " + th.getName() + " 发生未捕获异常：" + e.getMessage()));
                return t;
            }
        };

        // 2）有界队列 + 3）自定义拒绝策略 + 4）明确的 keepAlive
        return new ThreadPoolExecutor(
                core, max, keepAliveSec, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(queueCapacity),
                factory,
                new AlertingRejectHandler(bizName));
    }

    // 拆分线程池是"舱壁隔离"的关键：慢接口不能拖垮快接口
    public static void main(String[] args) {
        ThreadPoolExecutor orderPool = create("order", 8, 16, 500, 60);
        ThreadPoolExecutor notifyPool = create("notify", 2, 4, 200, 30);
        orderPool.execute(() -> System.out.println("订单任务执行于 " + Thread.currentThread().getName()));
        notifyPool.execute(() -> System.out.println("通知任务执行于 " + Thread.currentThread().getName()));
        orderPool.shutdown();
        notifyPool.shutdown();
    }
}

import java.util.concurrent.*;

/** 线程池动态调参 + 指标暴露 */
public class DynamicPoolTuning {
    public static void main(String[] args) {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                4, 8, 60, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                r -> new Thread(r, "dyn-pool"));

        // 运行时调参：setCorePoolSize / setMaximumPoolSize 均线程安全，立即生效
        pool.setCorePoolSize(16);
        pool.setMaximumPoolSize(32);
        pool.setKeepAliveTime(120, TimeUnit.SECONDS);
        // 开启后核心线程也会超时回收，低峰期可缩容到 0（注意 core 不能为 0 且要保证有线程消费队列）
        pool.allowCoreThreadTimeOut(true);

        // 注意：setCorePoolSize(新值 < 旧值) 时，多余的核心线程会在下次 getTask 超时后被回收；
        // setCorePoolSize(新值 > 旧值) 时，若队列非空，会立即创建新线程去消费积压任务。

        // 监控指标：建议每 10~60 秒采集一次并上报 Micrometer / Prometheus
        printMetrics(pool);
        pool.shutdown();
    }

    static void printMetrics(ThreadPoolExecutor pool) {
        System.out.printf("poolSize=%d(core=%d,max=%d) active=%d queue=%d completed=%d taskTotal=%d%n",
                pool.getPoolSize(), pool.getCorePoolSize(), pool.getMaximumPoolSize(),
                pool.getActiveCount(), pool.getQueue().size(),
                pool.getCompletedTaskCount(), pool.getTaskCount());
    }
}

import java.util.concurrent.*;

/** Future 的局限：get 阻塞、无法组合、异常被包装 */
public class FutureLimitDemo {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);

        Future<String> f1 = pool.submit(() -> queryUser());
        Future<String> f2 = pool.submit(() -> queryOrder());

        // 痛点 1：get() 阻塞当前线程 —— 想"两个都完成再做"只能顺序阻塞，无法声明式组合
        String u = f1.get();
        String o = f2.get();          // 即便 f2 先完成，也要等 f1

        // 痛点 2：无法把 f1 的结果直接传给下一个异步任务，只能自己写回调或再 submit
        // 痛点 3：异常被包装成 ExecutionException，需要层层 unwrap
        // 痛点 4：没有超时 + 默认值的组合能力（get(timeout) 超时抛异常，还得自己写降级）

        System.out.println(u + " | " + o);
        pool.shutdown();
    }

    static String queryUser()  { sleep(100); return "user:张三"; }
    static String queryOrder() { sleep(150); return "order:A001"; }
    static void sleep(long ms) { try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }
}

import java.util.concurrent.*;

/** 串行编排：thenApply / thenCompose / thenAccept / 异步回调线程归属 */
public class SerialChainDemo {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(4, r -> new Thread(r, "biz-" + r.hashCode()));

        // 1）thenApply：同步转换，由"上一个任务完成的线程"继续执行
        CompletableFuture<String> f = CompletableFuture
                .supplyAsync(() -> "订单A", pool)
                .thenApply(s -> s + "-已支付")            // 同一个 biz 线程继续跑
                .thenApply(String::toUpperCase);

        // 2）thenCompose：把 "返回 CF 的函数" 扁平化，避免 CF<CF<T>>
        CompletableFuture<String> flat = CompletableFuture
                .supplyAsync(() -> 1001L, pool)                  // CF<Long>
                .thenCompose(id -> queryDetailAsync(id, pool));   // Long -> CF<String>，展开为 CF<String>

        // 若误用 thenApply，类型会变成 CompletableFuture<CompletableFuture<String>>
        CompletableFuture<CompletableFuture<String>> nested = CompletableFuture
                .supplyAsync(() -> 1001L, pool)
                .thenApply(id -> queryDetailAsync(id, pool));

        // 3）thenAccept / thenRun：只消费不产生新值
        CompletableFuture<Void> done = flat.thenAccept(v -> System.out.println("结果：" + v))
                                           .thenRun(() -> System.out.println("链路结束"));

        System.out.println("thenApply 结果 = " + f.get());
        System.out.println("thenCompose 结果 = " + flat.get());
        System.out.println("嵌套类型结果 = " + nested.get().get());   // 要 get 两次，很别扭
        done.join();
        pool.shutdown();
    }

    static CompletableFuture<String> queryDetailAsync(long id, Executor ex) {
        return CompletableFuture.supplyAsync(() -> "detail-" + id, ex);
    }
}

import java.util.concurrent.*;

/** 三种异常处理方式的差异 */
public class ExceptionHandleDemo {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);

        // exceptionally：只处理异常，正常时透传。相当于 catch → 返回默认值
        CompletableFuture<String> f1 = CompletableFuture.<String>supplyAsync(() -> {
            throw new RuntimeException("下游超时");
        }, pool).exceptionally(ex -> "默认用户");

        // handle：无论成功失败都进入，可以做统一转换，返回值会替换原结果
        CompletableFuture<String> f2 = CompletableFuture.<String>supplyAsync(() -> {
            throw new RuntimeException("库存服务不可用");
        }, pool).handle((res, ex) -> res != null ? res : "降级结果(" + ex.getMessage() + ")");

        // whenComplete：类似 finally，不改变结果；如果这里不处理，异常会继续往下游抛
        CompletableFuture<String> f3 = CompletableFuture.supplyAsync(() -> "ok", pool)
                .whenComplete((res, ex) -> System.out.println("清理资源，res=" + res + ", ex=" + ex));

        System.out.println(f1.get());          // 默认用户
        System.out.println(f2.get());          // 降级结果(...)
        System.out.println(f3.get());          // ok

        // 注意：未处理的异常在调用 get() 时会被包装成 ExecutionException
        CompletableFuture<String> f4 = CompletableFuture.<String>supplyAsync(() -> {
            throw new IllegalStateException("boom");
        }, pool);
        try {
            f4.get();
        } catch (ExecutionException e) {
            System.out.println("捕获到真实原因：" + e.getCause());  // IllegalStateException: boom
        }
        pool.shutdown();
    }
}

import java.util.concurrent.*;
import java.util.function.Supplier;

/** 实战：三接口并行聚合 + 超时降级 + 线程池隔离 */
public class AggregateDemo {
    // 关键：为三个下游分别准备线程池，实现舱壁隔离，避免一个慢服务拖垮整条链路
    static final ExecutorService ITEM_POOL  = newFixedPool("item", 8);
    static final ExecutorService STOCK_POOL = newFixedPool("stock", 8);
    static final ExecutorService PROMO_POOL = newFixedPool("promo", 8);

    public static void main(String[] args) {
        long start = System.currentTimeMillis();

        CompletableFuture<String> itemF  = callAsync(() -> rpc("商品服务", 120), "商品", ITEM_POOL);
        CompletableFuture<String> stockF = callAsync(() -> rpc("库存服务", 80),  "库存", STOCK_POOL);
        CompletableFuture<String> promoF = callAsync(() -> rpc("营销服务", 200), "营销", PROMO_POOL);

        // allOf 本身返回 CF<Void>，需要自己从各个 Future 里取值
        CompletableFuture<Detail> detail = CompletableFuture.allOf(itemF, stockF, promoF)
                .thenApply(v -> new Detail(itemF.join(), stockF.join(), promoF.join()));

        // 整体再加一层超时兜底
        Detail result = detail
                .completeOnTimeout(Detail.partial(itemF.getNow("商品-未知"),
                                                  stockF.getNow("库存-未知"),
                                                  promoF.getNow("营销-无")),
                                   300, TimeUnit.MILLISECONDS)   // 超过 300ms 用已有结果兜底
                .exceptionally(ex -> Detail.fallback("聚合失败：" + ex.getMessage()))
                .join();

        System.out.println(result);
        System.out.println("总耗时 = " + (System.currentTimeMillis() - start) + "ms");
        shutdownAll();
    }

    /** 单个调用：带超时降级，任一服务超时不影响整体 */
    static <T> CompletableFuture<T> callAsync(Supplier<T> supplier, String name, ExecutorService pool) {
        return CompletableFuture.supplyAsync(supplier, pool)
                .orTimeout(150, TimeUnit.MILLISECONDS)                 // JDK 9+
                .exceptionally(ex -> (T) (name + "-降级"));            // 超时或异常都降级
    }

    static ExecutorService newFixedPool(String name, int n) {
        return Executors.newFixedThreadPool(n, r -> new Thread(r, name + "-" + r.hashCode()));
    }

    static String rpc(String svc, int costMs) {
        try { Thread.sleep(costMs); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return svc + "数据";
    }

    static void shutdownAll() {
        ITEM_POOL.shutdown(); STOCK_POOL.shutdown(); PROMO_POOL.shutdown();
    }

    record Detail(String item, String stock, String promo) {
        static Detail partial(String i, String s, String p) { return new Detail(i, s, p); }
        static Detail fallback(String msg) { return new Detail(msg, msg, msg); }
    }
}

import java.util.concurrent.*;

/** RecursiveTask：大数组并行求和 */
public class ForkJoinSumDemo {
    static final int THRESHOLD = 10_000;   // 阈值：小于它就直接算，避免过度拆分

    public static void main(String[] args) {
        long[] array = new long[100_000_000];
        for (int i = 0; i < array.length; i++) array[i] = i + 1;

        // 方式一：使用 commonPool（ForkJoinPool.commonPool()）
        long r1 = new ForkJoinPool().invoke(new SumTask(array, 0, array.length));
        // 顺便对比一下串行
        long serial = 0;
        for (long v : array) serial += v;
        System.out.println("ForkJoin = " + r1 + "，串行 = " + serial);
    }

    static class SumTask extends RecursiveTask<Long> {
        private final long[] arr;
        private final int start, end;

        SumTask(long[] arr, int start, int end) { this.arr = arr; this.start = start; this.end = end; }

        @Override
        protected Long compute() {
            int len = end - start;
            if (len <= THRESHOLD) {              // 足够小，直接计算（这是防止过度拆分的关键）
                long sum = 0;
                for (int i = start; i < end; i++) sum += arr[i];
                return sum;
            }
            int mid = start + (len >>> 1);
            SumTask left  = new SumTask(arr, start, mid);
            SumTask right = new SumTask(arr, mid, end);
            left.fork();                          // 左半异步执行（push 到自己的 WorkQueue）
            long rightResult = right.compute();   // 右半同步执行（复用当前线程，减少一次调度）
            long leftResult = left.join();        // 等待左半结果
            return leftResult + rightResult;
        }
    }
}

import java.util.Arrays;
import java.util.concurrent.*;

/** RecursiveAction：并行归并排序（无返回值） */
public class ForkJoinSortDemo {
    public static void main(String[] args) {
        int[] data = new int[1_000_000];
        for (int i = 0; i < data.length; i++) data[i] = (int) (Math.random() * 1_000_000);

        ForkJoinPool pool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
        pool.invoke(new SortTask(data, new int[data.length]));
        pool.shutdown();

        System.out.println("是否有序 = " + isSorted(data));
        System.out.println("前 10 个 = " + Arrays.toString(Arrays.copyOf(data, 10)));
    }

    static class SortTask extends RecursiveAction {
        private final int[] src;
        private final int[] tmp;
        private final int lo, hi;
        private static final int THRESHOLD = 5_000;

        SortTask(int[] src, int[] tmp, int lo, int hi) { this.src = src; this.tmp = tmp; this.lo = lo; this.hi = hi; }

        @Override
        protected void compute() {
            if (hi - lo <= THRESHOLD) {
                Arrays.sort(src, lo, hi);       // 小规模直接交给双轴快排，比继续拆分更快
                return;
            }
            int mid = (lo + hi) >>> 1;
            SortTask left  = new SortTask(src, tmp, lo, mid);
            SortTask right = new SortTask(src, tmp, mid, hi);
            invokeAll(left, right);             // 等价 fork+fork+join+join，语义更清晰
            merge(src, tmp, lo, mid, hi);
        }

        private void merge(int[] a, int[] tmp, int lo, int mid, int hi) {
            System.arraycopy(a, lo, tmp, lo, hi - lo);
            int i = lo, j = mid, k = lo;
            while (i < mid && j < hi) {
                a[k++] = (tmp[i] <= tmp[j]) ? tmp[i++] : tmp[j++];
            }
            while (i < mid) a[k++] = tmp[i++];
            while (j < hi)  a[k++] = tmp[j++];
        }
    }

    static boolean isSorted(int[] a) {
        for (int i = 1; i < a.length; i++) if (a[i - 1] > a[i]) return false;
        return true;
    }
}

import java.util.*;
import java.util.stream.*;

/** parallelStream 的正确与错误用法 */
public class ParallelStreamDemo {
    public static void main(String[] args) {
        List<Integer> nums = new ArrayList<>();
        for (int i = 1; i <= 1_000_000; i++) nums.add(i);

        // 正确：无状态、纯 CPU 计算、数据源易拆分、结果用 reduce 聚合 → 有加速
        long sum = nums.parallelStream().mapToLong(Integer::longValue).sum();
        System.out.println("sum = " + sum);

        // 错误 1：共享可变状态。ArrayList 非线程安全 → 结果丢失甚至抛异常
        List<Integer> unsafe = new ArrayList<>();
        nums.parallelStream().forEach(unsafe::add);          // 数据量小时"看起来没问题"，上线就翻车
        System.out.println("错误用法结果数 = " + unsafe.size() + "，期望 = " + nums.size());

        // 正确写法：用 collect 归约，让框架处理线程安全
        List<Integer> safe = nums.parallelStream().filter(x -> x % 2 == 0).toList();
        System.out.println("安全结果数 = " + safe.size());

        // 错误 2：用 forEach + 外部锁，把并行退化成串行，还多付了调度成本
        // 错误 3：IO 密集任务（数据库查询、HTTP 调用）放并行流 → 占满 commonPool，影响全 JVM
    }
}

import java.time.Duration;
import java.util.concurrent.*;
import java.util.stream.IntStream;

/** JDK 21 虚拟线程：三种创建方式 + 百万级并发演示（需 JDK 21+） */
public class VirtualThreadDemo {
    public static void main(String[] args) throws Exception {
        // 方式一：Thread.ofVirtual() 构建器
        Thread t1 = Thread.ofVirtual().name("vt-1").start(() -> System.out.println("hello " + Thread.currentThread()));
        t1.join();

        // 方式二：快捷方法 startVirtualThread
        Thread.startVirtualThread(() -> System.out.println("快捷创建：" + Thread.currentThread().isVirtual()));

        // 方式三：Executor —— 每个任务一个虚拟线程（注意：虚拟线程不要池化！）
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            long start = System.currentTimeMillis();
            // 10 万个"阻塞 100 毫秒"的任务：换成平台线程池需要巨大线程数，虚拟线程轻松完成
            IntStream.range(0, 100_000).forEach(i ->
                    executor.submit(() -> {
                        Thread.sleep(Duration.ofMillis(100));
                        return i;
                    }));
            System.out.println("提交 10 万任务耗时 = " + (System.currentTimeMillis() - start) + "ms");
        }
    }
}

import java.util.concurrent.locks.ReentrantLock;

/** 钉扎对比：synchronized 会 pin 住载体线程，ReentrantLock 不会 */
public class PinningDemo {
    private static final Object MONITOR = new Object();
    private static final ReentrantLock LOCK = new ReentrantLock();

    public static void main(String[] args) throws Exception {
        // 反例：synchronized 内阻塞 → 虚拟线程无法卸载 → carrier 被占用
        Thread v1 = Thread.ofVirtual().start(() -> {
            synchronized (MONITOR) {
                try { Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }
        });

        // 正例：换成 ReentrantLock，阻塞时可安全卸载
        Thread v2 = Thread.ofVirtual().start(() -> {
            LOCK.lock();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                LOCK.unlock();
            }
        });

        v1.join(); v2.join();
        System.out.println("诊断方式：启动参数加 -Djdk.tracePinnedThreads=short 可看到 synchronized 的钉扎堆栈");
    }
}