// 提交任务的人不需要关心线程如何创建、如何管理
Executor executor = new ThreadPoolExecutor(...);
executor.execute(() -> System.out.println("任务被提交，具体谁执行不关心"));

public ThreadPoolExecutor(
    int corePoolSize,              // ① 核心线程数
    int maximumPoolSize,           // ② 最大线程数
    long keepAliveTime,            // ③ 空闲存活时间
    TimeUnit unit,                 // ④ 时间单位
    BlockingQueue<Runnable> workQueue,     // ⑤ 阻塞队列
    ThreadFactory threadFactory,           // ⑥ 线程工厂
    RejectedExecutionHandler handler       // ⑦ 拒绝策略
)

ExecutorService executor = Executors.newFixedThreadPool(5);
// 实际构造：
// new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
//     new LinkedBlockingQueue<Runnable>())

ExecutorService executor = Executors.newCachedThreadPool();
// 实际构造：
// new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
//     new SynchronousQueue<Runnable>())

ExecutorService executor = Executors.newSingleThreadExecutor();
// 实际构造：
// new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
//     new LinkedBlockingQueue<Runnable>())

ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
// 实际构造：
// new ScheduledThreadPoolExecutor(5)
// 内部：super(corePoolSize, Integer.MAX_VALUE, ...)

ThreadPoolExecutor executor = new ThreadPoolExecutor(...);

// 第一步：不再接收新任务
executor.shutdown();
try {
    // 第二步：等待现有任务完成（最多等 60 秒）
    if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
        // 第三步：超时后强制关闭
        executor.shutdownNow();
        // 第四步：再等一会儿，看是否真的终止
        if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
            System.err.println("线程池未能完全终止");
        }
    }
} catch (InterruptedException e) {
    // 第五步：当前线程被中断时，强制关闭
    executor.shutdownNow();
    Thread.currentThread().interrupt();
}

// execute — 无返回值，异常直抛
executor.execute(() -> {
    throw new RuntimeException("这个异常会被直接抛出");
});

// submit — 有返回值，异常被 Future 捕获
Future<String> future = executor.submit(() -> {
    throw new RuntimeException("这个异常被 Future 捕获");
});
// future.get(); // 此时才抛出 ExecutionException（包装了原始异常）

// === 提交任务 ===
void execute(Runnable command);                    // 提交无返回值任务
<T> Future<T> submit(Callable<T> task);            // 提交有返回值任务
<T> Future<T> submit(Runnable task, T result);     // 提交任务，指定返回值
<T> List<Future<T>> invokeAll(Collection<Callable<T>> tasks); // 批量提交，全部完成返回
<T> T invokeAny(Collection<Callable<T>> tasks);    // 批量提交，任一完成即返回

// === 关闭线程池 ===
void shutdown();                                   // 优雅关闭
List<Runnable> shutdownNow();                      // 强制关闭（返回未执行任务）
boolean isShutdown();                              // 是否已关闭
boolean isTerminated();                            // 是否已终止
boolean awaitTermination(long timeout, TimeUnit unit); // 等待终止

// === 配置与监控 ===
void prestartAllCoreThreads();                     // 预启动所有核心线程
void allowCoreThreadTimeOut(boolean value);        // 允许核心线程超时回收
int getPoolSize();                                 // 当前线程数
int getActiveCount();                              // 活跃线程数
long getTaskCount();                               // 已执行 + 队列中任务数
long getCompletedTaskCount();                      // 已完成任务数
int getQueue().size();                             // 队列中等待的任务数

// 提交任务
ExecutorService executor = Executors.newFixedThreadPool(2);
Future<String> future = executor.submit(() -> {
    Thread.sleep(2000);
    return "结果";
});

// 核心方法
future.get();                              // 阻塞等待结果（一直等）
future.get(3, TimeUnit.SECONDS);           // 超时等待，超时抛 TimeoutException
future.isDone();                           // 任务是否完成
future.isCancelled();                      // 任务是否被取消
future.cancel(true);                       // 尝试取消任务
                                          // true = 即使正在运行也中断
                                          // false = 未启动才取消

// 方式一：直接给 Thread
FutureTask<String> futureTask = new FutureTask<>(() -> {
    return "通过 FutureTask 执行";
});
new Thread(futureTask).start();
String result = futureTask.get(); // 获取结果

// 方式二：提交给 Executor
FutureTask<String> task = new FutureTask<>(() -> "结果");
executor.execute(task); // execute 也能用！因为实现了 Runnable
String r = task.get();

// 方式三：一次性保证
// FutureTask 的 run() 只能执行一次，多次调用无效
// 适合做"缓存"——计算一次，结果复用

ExecutorService executor = Executors.newFixedThreadPool(2);

// 方式一：submit 任务的异常
Future<String> future = executor.submit(() -> {
    throw new RuntimeException("业务异常");
});
try {
    future.get(); // 抛出 ExecutionException
} catch (ExecutionException e) {
    Throwable cause = e.getCause(); // 获取原始异常
    System.out.println("异常原因: " + cause.getMessage());
}

// 方式二：execute 任务的异常
executor.execute(() -> {
    throw new RuntimeException("异常会直接抛出到线程");
});
// 可以通过设置 UncaughtExceptionHandler 捕获
ThreadFactory factory = r -> {
    Thread t = new Thread(r);
    t.setUncaughtExceptionHandler((thread, ex) -> {
        System.out.println("线程 " + thread.getName() + " 异常: " + ex.getMessage());
    });
    return t;
};

// 指定线程池（重要！默认用 ForkJoinPool.commonPool() 线程数 = CPU 核心数 - 1）
ExecutorService executor = new ThreadPoolExecutor(
    10, 20, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000)
);

// 无返回值
CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
    System.out.println("异步执行，无返回值");
}, executor);

// 有返回值
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    return "异步结果";
}, executor);

// 手动完成
CompletableFuture<String> future3 = new CompletableFuture<>();
future3.complete("手动设置结果");           // 立即完成
future3.completeExceptionally(new Exception("异常")); // 异常完成

// 获取结果
String result = future2.get();     // 阻塞等待（抛受检异常）
String result2 = future2.join();   // 阻塞等待（抛非受检异常，推荐）
String result3 = future2.getNow("默认值"); // 未完成返回默认值

// thenApply — 对结果做转换（有返回值）
CompletableFuture.supplyAsync(() -> "hello", executor)
    .thenApply(s -> s.toUpperCase())   // HELLO
    .thenApply(s -> s + " WORLD")      // HELLO WORLD
    .thenAccept(System.out::println);  // 消费结果

// thenAccept — 消费结果，无返回值
CompletableFuture.supplyAsync(() -> "结果", executor)
    .thenAccept(r -> System.out.println("消费: " + r));

// thenRun — 不关心结果，任务完了就执行
CompletableFuture.supplyAsync(() -> "结果", executor)
    .thenRun(() -> System.out.println("任务完成通知"));

// exceptionally — 异常兜底
CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) throw new RuntimeException("出错了");
    return "成功";
}, executor)
    .exceptionally(ex -> {
        System.out.println("异常: " + ex.getMessage());
        return "兜底结果";
    })
    .thenAccept(System.out::println);

// handle — 同时处理正常和异常
CompletableFuture.supplyAsync(() -> "正常结果", executor)
    .handle((result, ex) -> {
        if (ex != null) {
            System.out.println("异常: " + ex.getMessage());
            return "兜底结果";
        }
        return result;
    });

// thenCompose — 任务依赖：A → B（展开嵌套）
CompletableFuture.supplyAsync(() -> "A", executor)
    .thenCompose(a -> CompletableFuture.supplyAsync(() -> a + "B"));

// thenCombine — 合并两个并行任务的独立结果
CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> {
    sleep(1000); return "任务1结果";
}, executor);
CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> {
    sleep(2000); return "任务2结果";
}, executor);

task1.thenCombine(task2, (r1, r2) -> r1 + " + " + r2)
     .thenAccept(System.out::println);

// allOf — 等待所有任务完成（需手动收集结果）
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "A");
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "B");
CompletableFuture<String> f3 = CompletableFuture.supplyAsync(() -> "C");

CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3);
all.thenRun(() -> {
    String result = f1.join() + f2.join() + f3.join();
    System.out.println("全部完成: " + result);
});

// 收集结果的通用方式
public <T> CompletableFuture<List<T>> allOfList(List<CompletableFuture<T>> futures) {
    return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
        .thenApply(v -> futures.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList()));
}

// anyOf — 任意一个完成即可
CompletableFuture.anyOf(f1, f2, f3)
    .thenAccept(result -> System.out.println("最快的任务: " + result));

// 超时抛 TimeoutException
CompletableFuture.supplyAsync(() -> {
    sleep(5000); return "结果";
}, executor).orTimeout(2, TimeUnit.SECONDS);

// 超时返回默认值
CompletableFuture.supplyAsync(() -> {
    sleep(5000); return "结果";
}, executor).completeOnTimeout("默认值", 2, TimeUnit.SECONDS);

// 场景：查询商品详情，需要同时获取基本信息 + 库存 + 价格
public CompletableFuture<ProductVO> queryProduct(Long productId) {
    CompletableFuture<ProductInfo> infoFuture =
        CompletableFuture.supplyAsync(() -> productService.getInfo(productId), executor);

    CompletableFuture<Integer> stockFuture =
        CompletableFuture.supplyAsync(() -> stockService.getStock(productId), executor);

    CompletableFuture<BigDecimal> priceFuture =
        CompletableFuture.supplyAsync(() -> priceService.getPrice(productId), executor);

    return infoFuture
        .thenCombine(stockFuture, (info, stock) -> {
            ProductVO vo = new ProductVO();
            vo.setInfo(info);
            vo.setStock(stock);
            return vo;
        })
        .thenCombine(priceFuture, (vo, price) -> {
            vo.setPrice(price);
            return vo;
        })
        .exceptionally(ex -> {
            log.error("查询商品失败，productId: {}", productId, ex);
            return new ProductVO(); // 降级返回空对象
        });
}

public class ThreadPoolConfig {

    // 业务线程池（IO 密集型）
    public ThreadPoolExecutor ioExecutor() {
        return new ThreadPoolExecutor(
            10,                          // corePoolSize
            20,                          // maximumPoolSize
            60L, TimeUnit.SECONDS,       // 空闲 60s 回收
            new LinkedBlockingQueue<>(1000), // 有界队列
            new ThreadFactory() {
                private final AtomicInteger counter = new AtomicInteger(1);
                @Override
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r);
                    t.setName("io-pool-" + counter.getAndIncrement());
                    t.setDaemon(false);
                    return t;
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy() // 提交者自执行
        );
    }

    // 计算线程池（CPU 密集型）
    public ThreadPoolExecutor cpuExecutor() {
        int cores = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(
            cores + 1,
            cores + 1,
            0L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),
            r -> {
                Thread t = new Thread(r);
                t.setName("cpu-pool-" + t.getId());
                return t;
            },
            new ThreadPoolExecutor.AbortPolicy()
        );
    }
}

// 通过 ThreadPoolExecutor 提供的方法监控
public class ThreadPoolMonitor {

    public void monitor(ThreadPoolExecutor executor) {
        System.out.println(String.format(
            "线程池监控: [核心:%d, 当前:%d, 活跃:%d, 最大:%d, 队列:%d, 已完成:%d]",
            executor.getCorePoolSize(),
            executor.getPoolSize(),
            executor.getActiveCount(),
            executor.getMaximumPoolSize(),
            executor.getQueue().size(),
            executor.getCompletedTaskCount()
        ));
    }

    // 通过继承 ThreadPoolExecutor 自定义监控
    public static class MonitoredThreadPool extends ThreadPoolExecutor {

        public MonitoredThreadPool(int core, int max, long keepAlive, TimeUnit unit,
                                    BlockingQueue<Runnable> queue, ThreadFactory factory,
                                    RejectedExecutionHandler handler) {
            super(core, max, keepAlive, unit, queue, factory, handler);
        }

        @Override
        protected void beforeExecute(Thread t, Runnable r) {
            super.beforeExecute(t, r);
            // 记录任务开始时间（ThreadLocal）
        }

        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            super.afterExecute(r, t);
            // 计算任务耗时，记录监控指标
            if (t != null) {
                System.err.println("任务执行异常: " + t.getMessage());
            }
        }

        @Override
        protected void terminated() {
            super.terminated();
            System.out.println("线程池已终止");
        }
    }
}