new ThreadPoolExecutor(
    4, // 核心线程数
    8, // 最大线程数
    60L, // 线程空闲时间
    TimeUnit.SECONDS, // 线程空闲时间单位
    new LinkedBlockingQueue<>(1024), // 线程队列
    new ThreadPoolExecutor.DiscardOldestPolicy() // 拒绝策略
);

import java.util.concurrent.*;

public class CallableFutureExample {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建一个线程池
        ExecutorService executor = Executors.newSingleThreadExecutor();

        // 2. 创建一个Callable任务，该任务会计算1到100的和
        Callable<Integer> myTask = () -> {
            System.out.println("子线程 " + Thread.currentThread().getName() + " 开始计算...");
            TimeUnit.SECONDS.sleep(2); // 模拟耗时计算
            int sum = 0;
            for (int i = 1; i <= 100; i++) {
                sum += i;
            }
            return sum;
        };

        // 3. 提交Callable任务，获取Future对象（“提货单”）
        System.out.println("主线程：提交任务给线程池。");
        Future<Integer> future = executor.submit(myTask);

        // 4. 主线程可以继续做其他事情...
        System.out.println("主线程：我先去忙别的了...");
        TimeUnit.SECONDS.sleep(1);
        System.out.println("主线程：忙完了，现在来看看子线程的结果。");

        // 5. 通过Future.get()获取结果（如果任务没执行完，这里会阻塞）
        // ExecutionException 会包装Callable中抛出的原始异常
        Integer result = future.get(); 
        System.out.println("主线程：拿到任务执行结果 -> " + result);

        // 6. 关闭线程池
        executor.shutdown();
    }
}

// 传统 Future 的方式：主线程被阻塞
Future<String> future = executor.submit(() -> fetchDataFromRemote());
String data = future.get();  // 主线程卡在这里，啥也干不了
processData(data);

// CompletableFuture 的方式：注册回调，主线程不阻塞
CompletableFuture.supplyAsync(() -> fetchDataFromRemote(), executor)
    .thenAccept(data -> processData(data));  // 注册回调函数
// 主线程继续执行，不会被卡住

// 1. supplyAsync：用于有返回值的任务
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
    return "返回一个字符串";
});

// 2. runAsync：用于无返回值的任务（仅执行动作）
CompletableFuture<Void> cf2 = CompletableFuture.runAsync(() -> {
    System.out.println("执行一个动作，不返回结果");
});

// 问题场景演示
// 假设你的业务代码这样写
CompletableFuture.supplyAsync(() -> callSlowAPI());  // 使用默认线程池

// 同时，你的代码中还有并行流
list.parallelStream().map(...).collect(...);  // 也使用 commonPool

// 问题：两者共用同一个线程池，会互相干扰！
// 如果 callSlowAPI 耗时很长，会导致并行流也变慢

// 推荐做法：为业务创建专用线程池
ExecutorService bizExecutor = Executors.newFixedThreadPool(10, new ThreadFactory() {
    private AtomicInteger counter = new AtomicInteger(0);
    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r);
        t.setName("biz-async-" + counter.incrementAndGet());
        return t;
    }
});

// 明确指定线程池
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 模拟调用远程服务
    try {
        TimeUnit.SECONDS.sleep(2);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return "远程数据";
}, bizExecutor);  // 使用自定义线程池

ExecutorService executor = Executors.newFixedThreadPool(3);

CompletableFuture.supplyAsync(() -> {
    // 步骤1：模拟从远程服务获取商品ID
    System.out.println("[步骤1] 线程 " + Thread.currentThread().getName() + " 获取商品ID");
    return "12345";
}, executor)
.thenApply(idStr -> {
    // 步骤2：将字符串ID转换为整数
    // 这里使用 thenApply 是因为需要返回转换后的结果
    System.out.println("[步骤2] 线程 " + Thread.currentThread().getName() + " 转换ID为整数");
    return Integer.parseInt(idStr);
})
.thenApply(id -> {
    // 步骤3：将ID乘以价格系数10
    // 继续使用 thenApply 进行计算
    System.out.println("[步骤3] 线程 " + Thread.currentThread().getName() + " 计算价格");
    return id * 10;
})
.thenAccept(price -> {
    // 步骤4：打印最终价格
    // 使用 thenAccept 因为这一步只是输出，不需要返回值
    System.out.println("[步骤4] 线程 " + Thread.currentThread().getName() + " 最终价格: " + price);
})
.thenRun(() -> {
    // 步骤5：记录处理完成日志
    // 使用 thenRun 因为这一步不需要上一步的结果
    System.out.println("[步骤5] 线程 " + Thread.currentThread().getName() + " 处理流程结束");
});

executor.shutdown();

CompletableFuture.supplyAsync(() -> "数据", executor)
    .thenApply(s -> s.toUpperCase())      // 在完成 supplyAsync 的那个线程中执行
    .thenApplyAsync(s -> s + "!", executor); // 重新提交到 executor 线程池执行

ExecutorService executor = Executors.newFixedThreadPool(5);

CompletableFuture<String> userInfoFuture = CompletableFuture.supplyAsync(() -> {
    // 模拟查询用户信息
    sleep(1000);
    return "用户：张三";
}, executor);

CompletableFuture<String> balanceFuture = CompletableFuture.supplyAsync(() -> {
    // 模拟查询账户余额
    sleep(1500);
    return "余额：1000元";
}, executor);

// 使用 thenCombine 合并两个结果
CompletableFuture<String> combinedFuture = userInfoFuture.thenCombine(
    balanceFuture,
    (userInfo, balance) -> userInfo + ", " + balance  // 合并函数
);

// 这里为了演示我们调用 get()，实际应该继续用 thenAccept
System.out.println(combinedFuture.get());  // 输出: 用户：张三, 余额：1000元
// 总耗时约 1.5 秒（取决于最慢的那个任务）

executor.shutdown();

CompletableFuture<String> server1 = CompletableFuture.supplyAsync(() -> {
    sleep(2000);
    return "服务器1的数据";
}, executor);

CompletableFuture<String> server2 = CompletableFuture.supplyAsync(() -> {
    sleep(1000);
    return "服务器2的数据";
}, executor);

// 使用 applyToEither，哪个先完成就用哪个的结果
CompletableFuture<String> fastest = server1.applyToEither(
    server2,
    result -> "获取到：" + result
);

System.out.println(fastest.get());  // 输出: 获取到：服务器2的数据
// 总耗时约 1 秒

CompletableFuture<Void> all = CompletableFuture.allOf(future1, future2, future3);
all.join();  // 等待所有任务完成
// join() 方法类似 get()，但它抛出的是非受检异常（RuntimeException），不需要显式捕获

import java.util.concurrent.*;

public class ProductDetailService {
    // 为服务调用创建专用线程池
    private static final ExecutorService serviceExecutor = Executors.newFixedThreadPool(10);
    
    // 模拟调用商品服务
    static String fetchProductInfo(long productId) {
        sleep(1000);
        return "商品ID:" + productId + ", 名称:iPhone 15";
    }
    
    // 模拟调用价格服务
    static String fetchPriceInfo(long productId) {
        sleep(2000);
        return "价格:¥5999";
    }
    
    // 模拟调用库存服务
    static String fetchStockInfo(long productId) {
        sleep(1000);
        return "库存:100件";
    }
    
    static void sleep(long millis) {
        try { TimeUnit.MILLISECONDS.sleep(millis); } 
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }
}

public static void main(String[] args) {
    long productId = 12345L;
    System.out.println("开始获取商品详情...");
    long startTime = System.currentTimeMillis();
    
    // 并行发起三个异步调用
    CompletableFuture<String> productFuture = CompletableFuture.supplyAsync(
        () -> fetchProductInfo(productId), 
        serviceExecutor
    );
    
    CompletableFuture<String> priceFuture = CompletableFuture.supplyAsync(
        () -> fetchPriceInfo(productId), 
        serviceExecutor
    );
    
    CompletableFuture<String> stockFuture = CompletableFuture.supplyAsync(
        () -> fetchStockInfo(productId), 
        serviceExecutor
    );

// 使用 allOf 等待所有任务完成
CompletableFuture<Void> allTasks = CompletableFuture.allOf(
    productFuture, 
    priceFuture, 
    stockFuture
);

// join() 会阻塞当前线程，直到所有任务完成
// 这里阻塞是必要的，因为我们需要等所有数据都拿到才能组装结果
allTasks.join();

    // 此时所有 future 都已完成，调用 get() 不会阻塞
    try {
        String finalResult = String.format(
            "===== 商品详情页 =====\n%s\n%s\n%s",
            productFuture.get(),
            priceFuture.get(),
            stockFuture.get()
        );
        
        System.out.println(finalResult);
        System.out.println("总耗时: " + (System.currentTimeMillis() - startTime) + " ms");
        // 预期输出约 2000 ms，而不是 4000 ms
        
    } catch (Exception e) {
        System.err.println("获取商品详情失败: " + e.getMessage());
    } finally {
        serviceExecutor.shutdown();
    }
}

// 在一台 8GB 内存的服务器上
// 平台线程模式：创建 8000 个线程就接近内存极限
// 虚拟线程模式：可以轻松创建 1,000,000 个线程

// 理想中清晰易懂的代码
public void handleUserRequest(long userId) {
    // 步骤1：查询用户信息（等待数据库返回，线程阻塞）
    UserInfo user = userDao.findById(userId);
    
    // 步骤2：查询订单列表（等待数据库返回，线程阻塞）
    List<Order> orders = orderDao.findByUserId(userId);
    
    // 步骤3：调用推荐服务（等待HTTP响应，线程阻塞）
    Recommendations recs = recommendClient.getRecommendations(userId);
    
    // 步骤4：渲染页面
    return renderPage(user, orders, recs);
}

// 为了性能而被迫复杂化的代码
public CompletableFuture<PageData> handleUserRequestAsync(long userId) {
    return userDao.findByIdAsync(userId)
        .thenCompose(user -> {
            CompletableFuture<List<Order>> ordersFuture = orderDao.findByUserIdAsync(userId);
            CompletableFuture<Recommendations> recsFuture = recommendClient.getRecommendationsAsync(userId);
            
            return ordersFuture.thenCombine(recsFuture, (orders, recs) -> {
                return new PageData(user, orders, recs);
            });
        })
        .thenApply(data -> renderPage(data));
}

// 在虚拟线程 VT-1 中执行
Socket socket = new Socket("api.example.com", 80);
byte[] data = socket.getInputStream().read(); // 阻塞点

// 传统异步方式：为了性能而复杂
CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> callServiceA(), executor)
    .thenCompose(resultA -> callServiceB(resultA))
    .thenApply(resultB -> processData(resultB));

// 虚拟线程方式：简单的同步写法，自动获得异步性能
Thread.ofVirtual().start(() -> {
    String resultA = callServiceA();  // 看似阻塞，实际会自动让出载体线程
    String resultB = callServiceB(resultA);
    String finalResult = processData(resultB);
});

// 创建并启动一个虚拟线程
Thread vt = Thread.ofVirtual().start(() -> {
    System.out.println("Hello from virtual thread!");
    System.out.println("线程信息: " + Thread.currentThread());
    // 输出示例: VirtualThread[#21]/runnable@ForkJoinPool-1-worker-1
});

vt.join();  // 等待虚拟线程完成

// 使用构建器设置线程名称（支持自动递增）
Thread.Builder builder = Thread.ofVirtual().name("worker-", 0);

Runnable task = () -> {
    System.out.println("执行线程: " + Thread.currentThread().getName());
};

Thread t1 = builder.start(task);  // 创建 worker-0
Thread t2 = builder.start(task);  // 创建 worker-1
Thread t3 = builder.start(task);  // 创建 worker-2

t1.join(); t2.join(); t3.join();

// 旧方式：创建平台线程
Thread platformThread = new Thread(() -> {
    System.out.println("Platform thread");
});
platformThread.start();

// 新方式：创建虚拟线程
Thread virtualThread = Thread.ofVirtual().start(() -> {
    System.out.println("Virtual thread");
});

ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();

import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class VirtualThreadBatchProcessing {
    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();
        
        // 使用 try-with-resources 确保 executor 自动关闭
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            
            // 提交 10000 个模拟的 I/O 任务
            for (int i = 0; i < 10000; i++) {
                final int taskId = i;
                executor.submit(() -> {
                    try {
                        // 模拟调用远程服务（阻塞 100ms）
                        Thread.sleep(Duration.ofMillis(100));
                        
                        if (taskId % 1000 == 0) {
                            System.out.println("任务 #" + taskId + " 完成，线程: " + 
                                Thread.currentThread());
                        }
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                });
            }
            
        } // executor.close() 会自动调用，等待所有任务完成
        
        long elapsedTime = System.currentTimeMillis() - startTime;
        System.out.println("\n10000 个任务总耗时: " + elapsedTime + " ms");
        // 预期输出约 100-200ms（取决于CPU核心数）
        // 如果用传统线程池，可能需要数秒甚至数十秒
    }
}

// 如果用传统固定大小线程池处理上述任务
ExecutorService traditionalPool = Executors.newFixedThreadPool(100);

// 问题：
// 1. 10000个任务需要排队，只能由100个线程串行处理
// 2. 总耗时 = 10000 / 100 * 100ms = 10秒
// 3. 如果增加线程数到10000，会直接导致内存溢出

// 场景1：单次简单任务，使用 Thread.ofVirtual()
Thread.ofVirtual().start(() -> sendNotification(user));

// 场景2：批量任务处理，使用 ExecutorService
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (User user : users) {
        executor.submit(() -> sendNotification(user));
    }
}

// 场景3：需要获取任务结果
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Future<String>> futures = new ArrayList<>();
    for (int i = 0; i < 100; i++) {
        futures.add(executor.submit(() -> fetchData()));
    }
    
    for (Future<String> future : futures) {
        String result = future.get();
        processResult(result);
    }
}

// 场景：需要并发调用 1000 个微服务接口
// 每个接口响应时间 100ms

// 方案1：传统固定线程池（100个线程）
// 总耗时 = 1000 / 100 * 100ms = 1000ms

// 方案2：虚拟线程
// 总耗时 = 约 100ms（所有请求并发发出）

// CPU 密集型任务：使用平台线程池
int cpuCores = Runtime.getRuntime().availableProcessors();
ExecutorService cpuBoundExecutor = Executors.newFixedThreadPool(cpuCores);

cpuBoundExecutor.submit(() -> {
    // 执行复杂计算
    performHeavyComputation();
});

// I/O 密集型任务：使用虚拟线程
ExecutorService ioBoundExecutor = Executors.newVirtualThreadPerTaskExecutor();

ioBoundExecutor.submit(() -> {
    // 执行网络请求
    String data = httpClient.get("https://api.example.com");
    processData(data);
});

// 传统做法：因为线程昂贵，必须使用线程池
ExecutorService threadPool = Executors.newFixedThreadPool(200);

public void handleRequest(Request req) {
    threadPool.submit(() -> {
        // 任务1
        // 任务2
        // ...
    });
}

// 虚拟线程做法：每个请求创建一个新线程
public void handleRequest(Request req) {
    Thread.ofVirtual().start(() -> {
        // 任务1：查询用户信息（阻塞等待）
        User user = userService.getUser(req.getUserId());
        
        // 任务2：查询订单（阻塞等待）
        List<Order> orders = orderService.getOrders(user.getId());
        
        // 任务3：调用推荐服务（阻塞等待）
        Recommendations recs = recService.getRecommendations(user);
        
        // 任务4：渲染结果
        renderResponse(user, orders, recs);
    });
}

// 误解：虚拟线程能加速数据库查询
String result = database.query("SELECT..."); // 耗时 100ms

// 真相：无论用平台线程还是虚拟线程，这个查询仍然需要 100ms
// 虚拟线程解决的是"吞吐量"，而非"延迟"

// 误解：创建 100 万个虚拟线程进行计算就能提速
for (int i = 0; i < 1_000_000; i++) {
    Thread.ofVirtual().start(() -> computeHash(data));
}

// 真相：所有虚拟线程最终还是要在有限的 CPU 核心上执行
// 如果只有 8 核，同时只能真正并行 8 个计算任务

// 测试场景：10000 个并发 HTTP 请求，每个请求耗时 100ms

// 平台线程池（200个线程）：
// 总耗时约 5 秒
// 原因：10000 / 200 * 100ms = 5000ms

// 虚拟线程：
// 总耗时约 100ms
// 原因：所有请求几乎同时发出，只受网络带宽限制