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（四路同时发，等最慢的一路）
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 —— RT 直接砍到原来的 1/3

ExecutorService pool = Executors.newFixedThreadPool(4);

Future<Product> f1 = pool.submit(() -> productService.getById(id));
Future<Price>   f2 = pool.submit(() -> priceService.getPrice(id));

// ① get() 会阻塞调用线程 —— 无论 f2 多快，只要 f1 慢，这条线程就干等着
Product p = f1.get();            // 阻塞，最多等 800ms，期间线程啥也干不了

// ② 想"结果一出来就处理"？没有回调，只能轮询 isDone()
while (!f2.isDone()) {          // 轮询浪费 CPU，且延迟不可控
    Thread.sleep(10);
}

// ③ 想组合？f3 依赖 f1 和 f2 的结果 —— 只能先 get 出来，再手动 submit 一个新任务
Price pr = f2.get();             // 取结果的顺序被代码写死，无法"谁先好先处理谁"
Future<Coupon> f3 = pool.submit(() -> couponService.get(p, pr));

// 查用户 → 查会员价 → 查优惠券，每步都依赖上一步
ListenableFuture<User> userF = executor.submit(() -> userService.get(userId));
Futures.addCallback(userF, new FutureCallback<User>() {
    public void onSuccess(User user) {                    // 第 1 层
        ListenableFuture<Price> priceF = executor.submit(() -> priceService.get(user));
        Futures.addCallback(priceF, new FutureCallback<Price>() {
            public void onSuccess(Price price) {              // 第 2 层
                ListenableFuture<Coupon> couponF = executor.submit(() -> couponService.get(price));
                Futures.addCallback(couponF, new FutureCallback<Coupon>() {
                    public void onSuccess(Coupon coupon) {      // 第 3 层，才拿到全部数据
                        render(user, price, coupon);
                    }
                    public void onFailure(Throwable t) { // 每层都要写一遍错误处理
                        log.error("查询优惠券失败", t);
                    }
                }, executor);
            }
            public void onFailure(Throwable t) { log.error("查询价格失败", t); }
        }, executor);
    }
    public void onFailure(Throwable t) { log.error("查询用户失败", t); }
}, executor);

CompletableFuture<String> f = new CompletableFuture<>();  // 手动创建，此刻还没有任务

f.thenAccept(result -> log.info("收到结果：{}", result));  // ① 先"登记"依赖动作

new Thread(() -> {                                    // ② 另一个线程 1 秒后完成
    Thread.sleep(1000);
    f.complete("hello");                               // ③ 触发上面的回调（幂等，只有第一次生效）
}).start();

log.info("主线程继续做自己的事");                       // ④ 主线程不用等、不用轮询
// 输出顺序：先"主线程继续做自己的事"，1 秒后"收到结果：hello"

// ① 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，立即完成，后面接 thenXxx 会立刻执行

CompletableFuture<Price> loadPrice(long id) {
    Price cached = localCache.get(id);
    if (cached != null) {
        return CompletableFuture.completedFuture(cached);  // 命中缓存：立即完成
    }
    return CompletableFuture.supplyAsync(() -> priceService.getPrice(id), pool);
}

// 4 核机器 → commonPool 只有 3 个线程
CompletableFuture.supplyAsync(() -> rpcA.call());   // 默认丢进 commonPool，等 RPC 200ms
CompletableFuture.supplyAsync(() -> rpcB.call());   // 又占 1 个
CompletableFuture.supplyAsync(() -> rpcC.call());   // 又占 1 个 —— 3 个全在阻塞等 IO

// 此时全项目的 parallelStream() 和所有不传线程池的 CF 一起饿死排队！
list.parallelStream().map(...).collect(...);           // ← 也被拖垮

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>），类比 Stream.map()：输入 T，输出 U
CompletableFuture<String> nameFuture =
    supplyAsync(() -> productService.getById(123))   // CompletableFuture<Product>
    .thenApply(product -> product.getName());           // CompletableFuture<String>

// 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"));

// thenApply：同步转换（Function<T, U>）—— 回调里做的是同步操作
CompletableFuture<String> nameF =
    supplyAsync(() -> productService.getById(123))   // CompletableFuture<Product>
    .thenApply(product -> product.getName());           // getName() 是同步方法 → CompletableFuture<String>

// thenCompose：扁平化转换（Function<T, CompletableFuture<U>>）—— 回调里做的是异步操作
CompletableFuture<Price> priceF =
    supplyAsync(() -> productService.getById(123))        // CompletableFuture<Product>
    .thenCompose(product ->
        priceService.getPriceAsync(product.getSkuId()));    // 返回 CompletableFuture<Price>，自动拍平

// ✗ 错误：用 thenApply 串联异步步骤，得到套娃
CompletableFuture<CompletableFuture<Price>> nested =
    supplyAsync(() -> productService.getById(123))
    .thenApply(product -> priceService.getPriceAsync(product.getSkuId()));  // 嵌套！

Price price = nested.join().join();   // 只能再 join 一层 —— 又阻塞又难看

// ✓ 正确：thenCompose 自动拍平，链上始终只有一层
CompletableFuture<Price> flat =
    supplyAsync(() -> productService.getById(123))
    .thenCompose(product -> priceService.getPriceAsync(product.getSkuId()));

Price price = flat.join();   // 一层搞定

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 已完成，安全 join()
    DetailVO vo = assemble(f1.join(), f2.join(), f3.join(), f4.join());
    response.complete(vo);
});
// 注意：allOf 返回 CompletableFuture<Void>，不携带结果，必须手动 join 每个子 Future

// 多数据源竞速：谁先返回用谁（常用于缓存 + 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 —— 异常兜底，返回同类型默认值（类比 catch + return 默认值）
CompletableFuture<List<Review>> reviewsF =
    supplyAsync(() -> reviewService.list(productId))
    .exceptionally(ex -> {
        log.warn("评价服务异常", ex);
        return Collections.emptyList();  // ← 必须返回同类型
    });

// ② handle —— 成功和异常都处理，可返回不同类型（类比 try-catch-finally 里统一加工）
CompletableFuture<String> resultF = supplyAsync(() -> riskyCall())
    .handle((result, ex) -> ex != null ? "fallback" : result);

// ③ whenComplete —— 只读副作用（日志/埋点），不改变结果（类比 finally）
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"; });

// get()：抛两个受检异常，编译器逼你处理
try {
    Product p = f1.get();                        // throws InterruptedException, ExecutionException
} catch (InterruptedException e) {          // 等待期间线程被中断
    Thread.currentThread().interrupt();          // 最佳实践：恢复中断标志
} catch (ExecutionException e) {            // 任务执行失败
    Throwable cause = e.getCause();              // 原始异常在这里（要剥一层）
    log.error("任务执行失败", cause);
}

// join()：只抛 CompletionException（运行时异常），不写 try 也能编译
Product p = f2.join();                          // 失败了直接抛出来，自己接住
// cause 同样在 e.getCause() 里，异常类型是 CompletionException

// 缓存异步预热中：读到了就用，没读到先返回默认值，绝不阻塞请求线程
CompletableFuture<Price> preheating = loadPrice(id);          // 可能还没完成
Price price = preheating.getNow(Price.UNKNOWN);                // 未完成 → 立即返回 UNKNOWN

// orTimeout —— 超时则把 future 标记为"异常完成"（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, r -> {
        Thread t = new Thread(r, "cf-timeout-scheduler");
        t.setDaemon(true);
        return t;
    });

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);

// 每一步都要：传回调 + 传错误处理，越嵌越深
fetchUser(userId, user ->
    fetchMemberPrice(user, price ->
        calcPay(price, pay -> render(pay), onError),
    onError),
onError);

CompletableFuture.supplyAsync(() -> userService.get(userId), pool)
    .thenCompose(user -> priceService.getMemberPriceAsync(user, pool))
    .thenApply(price -> calcService.calcPay(price))
    .exceptionally(ex -> defaultPay())
    .thenAccept(pay -> render(pay));

// 还没列全的"双 Future 变体"：thenAcceptBoth（两个都完成再消费）、
// applyToEither / acceptEither（两个任一完成就处理）。面试提到"还有变体"即可加分。

// 团队内统一封装，禁止裸写 supplyAsync
public class Asyncs {
    public static <T> CompletableFuture<T> supply(Supplier<T> task,
            long timeoutMs, Supplier<T> fallback) {
        return CompletableFuture.supplyAsync(task, BizPools.ASYNC)
            .completeOnTimeout(fallback.get(), timeoutMs, TimeUnit.MILLISECONDS)
            .exceptionally(ex -> { log.warn("async task failed, fallback", ex); return fallback.get(); });
    }
}
// 调用方一行搞定：Asyncs.supply(() -> rpc.call(), 500, () -> defaultV)