CompletableFuture 异步编排：从回调地狱到流式组合

// 串行：800ms
Product product  = productService.getById(id);       // 200ms
Price   price    = priceService.getPrice(id);        // 200ms
List<Review> reviews = reviewService.list(id);   // 200ms
List<Recommend> recs = recommendService.list(id); // 200ms
// 总计 ≈ 800ms

// 并行 + 编排：≈ 300ms（max + 组装开销）
CompletableFuture<Product>   f1 = supplyAsync(() -> productService.getById(id));
CompletableFuture<Price>     f2 = supplyAsync(() -> priceService.getPrice(id));
CompletableFuture<List<Review>>   f3 = supplyAsync(() -> reviewService.list(id));
CompletableFuture<List<Recommend>> f4 = supplyAsync(() -> recommendService.list(id));

CompletableFuture.allOf(f1, f2, f3, f4).join();  // 等全部完成 ≈ 200ms
DetailVO vo = assemble(f1.get(), f2.get(), f3.get(), f4.get()); // 组装 ≈ 100ms
// 总计 ≈ 300ms

// ① supplyAsync —— 有返回值（Supplier → CompletableFuture<T>）
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
    // 耗时操作，最终返回结果
    return productService.getById(123).getName();
});

// ② runAsync —— 无返回值（Runnable → CompletableFuture<Void>）
CompletableFuture<Void> f2 = CompletableFuture.runAsync(() -> {
    // 只执行操作，不返回结果（如发通知、写日志）
    notifyService.send("order_created");
});

// ③ completedFuture —— 已经完成的 Future（常用于测试或默认值）
CompletableFuture<String> f3 = CompletableFuture.completedFuture("cached-value");
f3.isDone();  // true，立即完成

ExecutorService bizPool = new ThreadPoolExecutor(16, 32, 60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(1000),
    new ThreadFactoryBuilder().setNameFormat("biz-async-%d").build(),
    new ThreadPoolExecutor.CallerRunsPolicy());

CompletableFuture<Product> f = CompletableFuture.supplyAsync(
    () -> productService.getById(123), bizPool);  // ← 第二个参数指定线程池

// thenApply：同步转换（Function<T, U> → CompletableFuture<U>）
// 类比 Stream.map()：输入 T，输出 U
CompletableFuture<String> nameFuture =
    supplyAsync(() -> productService.getById(123))   // CompletableFuture<Product>
    .thenApply(product -> product.getName());           // CompletableFuture<String>

// thenCompose：扁平化转换（Function<T, CompletableFuture<U>> → CompletableFuture<U>）
// 类比 Stream.flatMap()：下一步本身也是异步的，需要"拍平"嵌套
CompletableFuture<Price> priceFuture =
    supplyAsync(() -> productService.getById(123))       // CompletableFuture<Product>
    .thenCompose(product ->
        priceService.getPriceAsync(product.getSkuId()));   // 返回 CompletableFuture<Price>
    // 如果用 thenApply，结果会是 CompletableFuture<CompletableFuture<Price>> ← 嵌套！

// thenApply（同步转换）vs thenApplyAsync（异步转换）
f.thenApply(result -> result.toUpperCase());         // 在上一步线程中同步执行
f.thenApplyAsync(result -> callRPC(result));           // 提交到线程池异步执行（转换本身耗时）

// thenAccept：消费结果，无返回值（类比 forEach）
supplyAsync(() -> orderService.create(dto))
    .thenAccept(orderId -> log.info("订单创建成功: {}", orderId));

// thenRun：不关心上一步结果，只执行后续动作
supplyAsync(() -> orderService.create(dto))
    .thenRun(() -> metricsService.increment("order.created"));

CompletableFuture<Product> productF = supplyAsync(() -> productService.getById(id));
CompletableFuture<Price>   priceF   = supplyAsync(() -> priceService.getPrice(id));

CompletableFuture<DetailVO> detailF = productF.thenCombine(priceF, (product, price) -> {
    DetailVO vo = new DetailVO();
    vo.setName(product.getName());  vo.setPrice(price.getAmount());
    return vo;
});

CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3, f4);
all.thenRun(() -> {
    // 所有 Future 已完成，安全 get()
    DetailVO vo = assemble(f1.join(), f2.join(), f3.join(), f4.join());
    response.complete(vo);
});
// 注意：allOf 返回 CompletableFuture<Void>，不携带结果，必须手动 join

// 多数据源竞速：谁先返回用谁（常用于缓存 + DB 双读）
CompletableFuture<String> fromCache = supplyAsync(() -> cacheService.get(key));
CompletableFuture<String> fromDB = supplyAsync(() -> dbService.query(key));
CompletableFuture<Object> fastest = CompletableFuture.anyOf(fromCache, fromDB);
// 注意：anyOf 返回 CompletableFuture<Object>，需要强转

// ① exceptionally —— 异常兜底，返回同类型默认值
CompletableFuture<List<Review>> reviewsF =
    supplyAsync(() -> reviewService.list(productId))
    .exceptionally(ex -> {
        log.warn("评价服务异常", ex);
        return Collections.emptyList();  // ← 必须返回同类型
    });
// ② handle —— 成功和异常都处理，可返回不同类型
CompletableFuture<String> resultF = supplyAsync(() -> riskyCall())
    .handle((result, ex) -> ex != null ? "fallback" : result);
// ③ whenComplete —— 副作用（日志/埋点），不改变结果
supplyAsync(() -> orderService.create(dto))
    .whenComplete((orderId, ex) -> {
        if (ex != null) log.error("创建订单失败", ex);
        else log.info("创建订单成功: {}", orderId);
    });  // 结果不变

// 异常会跳过中间步骤，直接传到最近的 exceptionally
supplyAsync(() -> { throw new RuntimeException("boom"); })
    .thenApply(s -> s.toUpperCase())   // ← 跳过
    .thenCompose(s -> callRPC(s))      // ← 跳过
    .exceptionally(ex -> { log.error("链路异常", ex); return "default"; });

// orTimeout —— 超时则标记为 TimeoutException
CompletableFuture<Product> f = supplyAsync(() -> productService.getById(id))
    .orTimeout(500, TimeUnit.MILLISECONDS);
// 500ms 内未完成 → future 以 TimeoutException 完成
// 后续 exceptionally/handle 可以捕获该异常

// completeOnTimeout —— 超时则用默认值完成（不会抛异常）
CompletableFuture<List<Review>> reviewsF =
    supplyAsync(() -> reviewService.list(id))
    .completeOnTimeout(Collections.emptyList(), 300, TimeUnit.MILLISECONDS);
// 300ms 内未完成 → 自动用空列表完成，下游正常执行

// JDK 8 没有 orTimeout / completeOnTimeout，需手动实现
private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

public static <T> CompletableFuture<T> withTimeout(
        CompletableFuture<T> future, long timeout, TimeUnit unit) {
    scheduler.schedule(() ->
        future.completeExceptionally(new TimeoutException("timed out")), timeout, unit);
    return future;  // complete 幂等，Future 已完成则此调用无效
}

CompletableFuture<Product> f = withTimeout(
    supplyAsync(() -> productService.getById(id)), 500, TimeUnit.MILLISECONDS);

public CompletableFuture<DetailVO> getProductDetail(long productId) {
    // Pipeline：查用户 → 查会员价（有依赖，用 thenCompose）
    CompletableFuture<PriceVO> priceF = supplyAsync(() -> userService.getCurrentUser(), pool)
        .thenCompose(user -> priceService.getMemberPriceAsync(productId, user.getLevel()))
        .orTimeout(300, TimeUnit.MILLISECONDS)
        .exceptionally(ex -> priceService.getDefaultPrice(productId));

    // Fan-out：商品、评价、推荐无依赖，并行发起
    CompletableFuture<Product> productF = supplyAsync(() -> productService.getById(productId), pool)
        .orTimeout(200, TimeUnit.MILLISECONDS);
    CompletableFuture<List<Review>> reviewsF = supplyAsync(() -> reviewService.list(productId), pool)
        .completeOnTimeout(Collections.emptyList(), 300, TimeUnit.MILLISECONDS);
    CompletableFuture<List<Recommend>> recsF = supplyAsync(() -> recommendService.list(productId), pool)
        .completeOnTimeout(Collections.emptyList(), 300, TimeUnit.MILLISECONDS);

    // Fan-in：聚合所有结果
    return CompletableFuture.allOf(productF, priceF, reviewsF, recsF)
        .thenApply(v -> {
            DetailVO vo = new DetailVO();
            vo.setProduct(productF.join());  vo.setPrice(priceF.join());
            vo.setReviews(reviewsF.join());  vo.setRecommendations(recsF.join());
            return vo;
        });
}

public static <T> CompletableFuture<T> withRetry(
        Supplier<CompletableFuture<T>> action, int maxRetries, long timeoutMs) {
    return action.get()
        .orTimeout(timeoutMs, TimeUnit.MILLISECONDS)
        .exceptionally(ex -> {
            if (maxRetries > 0) return withRetry(action, maxRetries - 1, timeoutMs).join();
            throw new CompletionException("重试耗尽", ex);
        });
}
// 使用：最多重试 2 次，每次超时 500ms
CompletableFuture<Product> f = withRetry(
    () -> supplyAsync(() -> productService.getById(id), pool), 2, 500);