CompletableFuture ⭐⭐

// 传统 Future 的典型用法
ExecutorService executor = Executors.newFixedThreadPool(2);
 
// 提交一个耗时任务，返回 Future 对象
Future<String> future = executor.submit(() -> {
    Thread.sleep(2000); // 模拟耗时操作
    return "查询结果";
});
 
// ❌ 痛点1：get() 是阻塞的，调用线程只能干等
String result = future.get(); // 当前线程在此"卡住"，直到结果返回
 
// ❌ 痛点2：无法手动完成——如果想提前给 Future 赋值，做不到
// ❌ 痛点3：无法链式编排——"拿到结果后做 A，再做 B，再做 C" 写起来极为笨拙
// ❌ 痛点4：无法组合多个 Future——"等两个任务都完成后汇总" 需要手写循环轮询
// ❌ 痛点5：没有内建的异常处理机制——异常只有在 get() 时才会以 ExecutionException 抛出

// CompletableFuture 的类签名
public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
    // ...
}

// 一个典型的 CompletableFuture 流水线示例
CompletableFuture
    .supplyAsync(() -> queryDatabase())       // Step1: 异步查询数据库
    .thenApply(data -> parseData(data))       // Step2: 拿到结果后，同步解析数据
    .thenApplyAsync(parsed -> enrich(parsed)) // Step3: 异步地补充数据
    .thenAccept(result -> sendResponse(result))// Step4: 消费最终结果，发送响应
    .exceptionally(ex -> {                    // 异常兜底：流水线中任一环节出错都会走这里
        log.error("流水线执行失败", ex);
        return null;
    });

// 创建一个"空"的 CompletableFuture，此时没有结果
CompletableFuture<String> cf = new CompletableFuture<>();
 
// 在另一个线程中手动完成它（模拟某个外部事件触发）
new Thread(() -> {
    try {
        Thread.sleep(1000);           // 模拟耗时操作
        cf.complete("手动赋值的结果"); // ✅ 手动完成，所有等待者立即收到结果
    } catch (Exception e) {
        cf.completeExceptionally(e);  // ✅ 手动以异常完成
    }
}).start();
 
// 主线程可以通过回调非阻塞地消费结果
cf.thenAccept(result -> System.out.println("收到结果: " + result));

// 演示线程模型差异
ExecutorService myPool = Executors.newFixedThreadPool(4); // 自定义线程池
 
CompletableFuture
    .supplyAsync(() -> {
        // 在 ForkJoinPool.commonPool() 的某个线程中执行
        System.out.println("供给: " + Thread.currentThread().getName());
        return 42;
    })
    .thenApply(v -> {
        // 可能在 commonPool 线程中执行，也可能在主线程中执行（取决于前驱是否完成）
        System.out.println("同步转换: " + Thread.currentThread().getName());
        return v * 2;
    })
    .thenApplyAsync(v -> {
        // 一定在 myPool 的线程中执行（因为我们指定了自定义线程池）
        System.out.println("异步转换: " + Thread.currentThread().getName());
        return v + 1;
    }, myPool) // 👈 指定自定义线程池
    .thenAccept(v -> System.out.println("最终结果: " + v));

// CompletableFuture 内部核心字段（简化示意）
volatile Object result;     // 存储结果值或异常（AltResult 包装）
volatile Completion stack;  // 依赖此 CF 的回调链（Treiber Stack，无锁栈）

            ┌─────────────────────────────────────────────────────────┐
            │              CompletableFuture 内部结构                   │
            │                                                         │
            │   result: volatile Object ──► 存储最终值或异常            │
            │                                                         │
            │   stack:  volatile Completion (Treiber Stack)           │
            │            ┌──────────┐                                 │
            │            │ callback3│──► next ──► null                │
            │            ├──────────┤                                 │
            │            │ callback2│──► next ──► callback3           │
            │            ├──────────┤                                 │
            │   top ───► │ callback1│──► next ──► callback2           │
            │            └──────────┘                                 │
            │                                                         │
            │   complete(value):                                      │
            │     1. CAS 写入 result                                  │
            │     2. postComplete() 弹出栈中所有 Completion 并执行      │
            │     3. 每个 Completion 完成后驱动下游 CF                  │
            └─────────────────────────────────────────────────────────┘

// 使用默认的 ForkJoinPool.commonPool() 作为线程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
 
// 使用自定义的 Executor（推荐在生产环境中使用）
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

import java.util.concurrent.CompletableFuture;
 
public class SupplyAsyncDemo {
    public static void main(String[] args) throws Exception {
 
        // 1. 使用 supplyAsync 创建一个异步任务
        //    Lambda 体内的代码将在 ForkJoinPool.commonPool() 的某个线程中执行
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 打印当前执行线程的名称，验证它确实在异步线程中运行
            System.out.println("异步任务执行线程: " + Thread.currentThread().getName());
            // 模拟一个耗时操作（如调用远程 API、查询数据库）
            try {
                Thread.sleep(1000); // 模拟耗时 1 秒
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 恢复中断标志
            }
            // 返回计算结果，该结果将填充到 CompletableFuture 中
            return "Hello from async thread!";
        });
 
        // 2. 主线程可以继续做其他事情，不会被阻塞
        System.out.println("主线程继续执行: " + Thread.currentThread().getName());
 
        // 3. 当需要结果时，调用 get() 或 join() 获取
        //    get() 会抛出受检异常 (ExecutionException, InterruptedException)
        //    join() 会抛出非受检异常 (CompletionException)，在 Lambda 链式调用中更方便
        String result = future.join(); // 阻塞等待结果
        System.out.println("异步任务结果: " + result);
    }
}

import java.util.concurrent.*;
 
public class SupplyAsyncCustomPoolDemo {
    public static void main(String[] args) {
 
        // 1. 创建自定义线程池
        //    核心线程数 4，最大线程数 8，空闲线程存活 60 秒
        //    使用有界队列（容量 100）防止 OOM
        //    自定义线程工厂，给线程起一个有意义的名字
        ExecutorService executor = new ThreadPoolExecutor(
                4,                                   // corePoolSize: 核心线程数
                8,                                   // maximumPoolSize: 最大线程数
                60L,                                 // keepAliveTime: 空闲线程存活时间
                TimeUnit.SECONDS,                    // 时间单位
                new LinkedBlockingQueue<>(100),       // 工作队列，有界防止 OOM
                r -> {                                // ThreadFactory: 自定义线程命名
                    Thread t = new Thread(r);
                    t.setName("my-async-pool-" + t.getId()); // 设置线程名
                    t.setDaemon(true);                // 设置为守护线程，JVM 退出时自动销毁
                    return t;
                },
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：由提交线程自己执行
        );
 
        // 2. 将自定义线程池传入 supplyAsync 的第二个参数
        CompletableFuture<Double> priceFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询价格线程: " + Thread.currentThread().getName());
            // 模拟查询商品价格
            return 99.99;
        }, executor); // <-- 指定自定义 Executor
 
        // 3. 获取结果
        Double price = priceFuture.join();
        System.out.println("商品价格: " + price);
 
        // 4. 使用完毕后关闭线程池（重要！防止资源泄漏）
        executor.shutdown();
    }
}

// 使用默认的 ForkJoinPool.commonPool()
public static CompletableFuture<Void> runAsync(Runnable runnable)
 
// 使用自定义 Executor
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)

import java.util.concurrent.CompletableFuture;
 
public class RunAsyncDemo {
    public static void main(String[] args) {
 
        // 1. 使用 runAsync 创建一个无返回值的异步任务
        //    适用于 fire-and-forget（触发即忘）场景
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            // 打印线程名，验证异步执行
            System.out.println("日志记录线程: " + Thread.currentThread().getName());
            // 模拟写入审计日志到数据库
            System.out.println("正在写入审计日志...");
            try {
                Thread.sleep(500); // 模拟 I/O 耗时
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            System.out.println("审计日志写入完成");
            // 注意：这里没有 return 语句，因为 Runnable 不返回值
        });
 
        // 2. 主线程继续处理其他业务逻辑
        System.out.println("主线程: 继续处理核心业务...");
 
        // 3. 如果需要等待异步任务完成（但不需要结果）
        future.join(); // 阻塞直到任务完成，返回值为 null
        System.out.println("所有任务完成");
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class AsyncNotificationDemo {
    public static void main(String[] args) {
 
        // 创建专用的通知线程池
        ExecutorService notifyPool = Executors.newFixedThreadPool(2);
 
        // 模拟：用户下单成功后，异步发送各种通知
        System.out.println("订单创建成功，开始异步通知...");
 
        // 异步发送短信通知（不需要返回值）
        CompletableFuture<Void> smsFuture = CompletableFuture.runAsync(() -> {
            System.out.println("[" + Thread.currentThread().getName() + "] 发送短信通知...");
            sleep(300); // 模拟短信网关调用
            System.out.println("[" + Thread.currentThread().getName() + "] 短信发送成功");
        }, notifyPool); // 使用专用线程池
 
        // 异步发送邮件通知（不需要返回值）
        CompletableFuture<Void> emailFuture = CompletableFuture.runAsync(() -> {
            System.out.println("[" + Thread.currentThread().getName() + "] 发送邮件通知...");
            sleep(500); // 模拟邮件服务调用
            System.out.println("[" + Thread.currentThread().getName() + "] 邮件发送成功");
        }, notifyPool); // 使用同一个专用线程池
 
        // 等待所有通知完成（allOf 后面章节会详细讲）
        CompletableFuture.allOf(smsFuture, emailFuture).join();
        System.out.println("所有通知发送完毕");
 
        // 关闭线程池
        notifyPool.shutdown();
    }
 
    // 工具方法：简化 sleep 操作
    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 恢复中断标志
        }
    }
}

public static <U> CompletableFuture<U> completedFuture(U value)

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
 
public class CompletedFutureDemo {
 
    // 模拟一个本地缓存
    private static final ConcurrentHashMap<String, String> cache = new ConcurrentHashMap<>();
 
    static {
        cache.put("user:1001", "Alice"); // 预置缓存数据
    }
 
    /**
     * 查询用户名称。
     * 方法签名统一返回 CompletableFuture<String>，
     * 但内部根据缓存命中与否选择不同的实现路径。
     */
    public static CompletableFuture<String> getUserName(String userId) {
        // 1. 先查缓存
        String cached = cache.get(userId);
 
        if (cached != null) {
            // 缓存命中：直接用 completedFuture 包装已有结果
            // 不需要异步，不需要线程切换，零开销
            return CompletableFuture.completedFuture(cached);
        }
 
        // 2. 缓存未命中：异步查询数据库
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("从数据库查询: " + userId);
            // 模拟数据库查询
            String name = "User-" + userId;
            // 查到后写入缓存
            cache.put(userId, name);
            return name;
        });
    }
 
    public static void main(String[] args) {
        // 调用者无需关心结果来自缓存还是数据库，API 完全一致
        String name1 = getUserName("user:1001").join(); // 走缓存，同步返回
        String name2 = getUserName("user:9999").join(); // 走数据库，异步查询
 
        System.out.println("name1 = " + name1); // Alice
        System.out.println("name2 = " + name2); // User-user:9999
    }
}

import java.util.concurrent.CompletableFuture;
 
public class CompletedFutureChainDemo {
    public static void main(String[] args) {
 
        // 用已知值作为起点，启动一条处理链
        String result = CompletableFuture
                .completedFuture(100)               // 起点：已知值 100
                .thenApply(n -> n * 2)               // 同步转换：100 -> 200
                .thenApply(n -> "结果是: " + n)       // 同步转换：200 -> "结果是: 200"
                .join();                              // 获取最终结果
 
        System.out.println(result); // 输出: 结果是: 200
    }
}

// 测试代码中 Mock 异步服务
public class UserServiceTest {
 
    // Mock 的异步方法：直接返回已完成的 Future，不涉及线程池
    CompletableFuture<String> mockGetUserName(String userId) {
        return CompletableFuture.completedFuture("MockUser"); // 直接返回预设值
    }
 
    // @Test
    public void testGetUserName() {
        // 测试时不需要等待异步线程，join() 立即返回
        String name = mockGetUserName("any-id").join();
        assert "MockUser".equals(name); // 断言通过
    }
}

// Java 9+
// 创建一个已经以异常完成的 CompletableFuture
CompletableFuture<String> failed = CompletableFuture.failedFuture(
        new RuntimeException("模拟失败")
);
 
// 尝试获取结果会立即抛出异常
// failed.join(); // 抛出 CompletionException

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
 
public class ManualCompleteDemo {
    public static void main(String[] args) {
 
        // 1. 创建一个空壳 CompletableFuture，此时它处于"未完成"状态
        CompletableFuture<String> promise = new CompletableFuture<>();
 
        // 2. 模拟一个回调风格的异步 API（比如 Netty、Vert.x 等框架常见的模式）
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.schedule(() -> {
            // 假设回调在 1 秒后触发，手动完成这个 Future
            promise.complete("来自回调的结果"); // <-- 手动填入结果
        }, 1, TimeUnit.SECONDS);
 
        // 3. 现在 promise 可以像普通 CompletableFuture 一样使用
        System.out.println("等待回调...");
        String result = promise.join(); // 阻塞直到 complete() 被调用
        System.out.println("收到: " + result);
 
        scheduler.shutdown();
    }
}

CompletableFuture<String> cf1 = CompletableFuture.completedFuture("Hello");
CompletableFuture<Void> cf2 = CompletableFuture.runAsync(() -> {
    try { Thread.sleep(100); } catch (InterruptedException e) {}
});
 
System.out.println("cf1 isDone: " + cf1.isDone());
System.out.println("cf2 isDone: " + cf2.isDone());
System.out.println("cf1 result: " + cf1.join());

// CompletableFuture<T> 的实例方法
// 接收 Function<? super T, ? extends U>，返回 CompletableFuture<U>
public <U> CompletableFuture<U> thenApply(Function<? super T, ? extends U> fn)

import java.util.concurrent.CompletableFuture;
 
public class ThenApplyDemo {
    public static void main(String[] args) throws Exception {
 
        // 第一阶段：异步计算一个整数
        CompletableFuture<Integer> priceFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询价格线程: " + Thread.currentThread().getName()); // ForkJoinPool.commonPool-worker-x
            return 100; // 模拟查询到商品原价为 100
        });
 
        // 第二阶段：对价格进行转换（打八折）
        // thenApply 接收 Function<Integer, Double>，将 Integer 转换为 Double
        CompletableFuture<Double> discountFuture = priceFuture.thenApply(price -> {
            System.out.println("折扣计算线程: " + Thread.currentThread().getName()); // 可能是 main，也可能是 worker
            return price * 0.8; // 原价 × 0.8 = 折后价
        });
 
        // 第三阶段：继续转换 —— 格式化为字符串
        // 链式调用，将 Double 映射为 String
        CompletableFuture<String> resultFuture = discountFuture.thenApply(discounted -> {
            return String.format("折后价: ¥%.2f", discounted); // 格式化输出
        });
 
        // 获取最终结果（阻塞等待）
        System.out.println(resultFuture.get()); // 输出: 折后价: ¥80.00
    }
}

import java.util.concurrent.CompletableFuture;
 
public class ThenApplyChainDemo {
    public static void main(String[] args) throws Exception {
 
        // 一条链完成：查询 → 打折 → 格式化
        String result = CompletableFuture
                .supplyAsync(() -> 100)                          // Stage 1: 异步查询原价
                .thenApply(price -> price * 0.8)                 // Stage 2: 同步转换 —— 计算折扣
                .thenApply(discounted -> String.format("¥%.2f", discounted)) // Stage 3: 同步转换 —— 格式化
                .get();                                          // 终端操作: 阻塞获取最终 String 结果
 
        System.out.println(result); // 输出: ¥80.00
    }
}

import java.util.concurrent.CompletableFuture;
 
public class ThenApplyThreadDemo {
    public static void main(String[] args) throws Exception {
 
        // === 情况一：上游任务耗时较长，thenApply 注册时上游尚未完成 ===
        CompletableFuture<String> case1 = CompletableFuture.supplyAsync(() -> {
            sleep(500); // 模拟耗时操作（500ms）
            System.out.println("[Case1 上游] " + Thread.currentThread().getName());
            return "data";
        }).thenApply(s -> {
            // 由于注册 thenApply 时上游还没完成，
            // 所以此函数会由 "完成上游的那个线程" 来执行
            System.out.println("[Case1 thenApply] " + Thread.currentThread().getName());
            return s.toUpperCase();
        });
 
        case1.get(); // 等待完成
 
        System.out.println("----------");
 
        // === 情况二：上游任务瞬间完成，thenApply 注册时上游已完成 ===
        CompletableFuture<String> alreadyDone = CompletableFuture.supplyAsync(() -> {
            System.out.println("[Case2 上游] " + Thread.currentThread().getName());
            return "data"; // 几乎无耗时，瞬间完成
        });
 
        sleep(200); // 主线程休眠 200ms，确保上游已经完成
 
        CompletableFuture<String> case2 = alreadyDone.thenApply(s -> {
            // 此时上游已完成，thenApply 在 **当前调用线程（main）** 上执行
            System.out.println("[Case2 thenApply] " + Thread.currentThread().getName());
            return s.toUpperCase();
        });
 
        case2.get();
    }
 
    // 辅助方法：安全休眠
    private static void sleep(long millis) {
        try { Thread.sleep(millis); } catch (InterruptedException ignored) {}
    }
}

[Case1 上游] ForkJoinPool.commonPool-worker-1
[Case1 thenApply] ForkJoinPool.commonPool-worker-1    ← 同一线程执行
----------
[Case2 上游] ForkJoinPool.commonPool-worker-1
[Case2 thenApply] main                                ← 主线程执行！

// 版本 1：使用默认的 ForkJoinPool.commonPool()
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T, ? extends U> fn)
 
// 版本 2：使用自定义的 Executor（推荐在生产环境中使用）
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor)

import java.util.concurrent.*;
 
public class ThenApplyAsyncDemo {
    public static void main(String[] args) throws Exception {
 
        // 创建自定义线程池（线程名前缀为 "my-pool"）
        ExecutorService myPool = Executors.newFixedThreadPool(2, r -> {
            Thread t = new Thread(r);                  // 创建新线程
            t.setName("my-pool-" + t.getId());         // 自定义线程名
            return t;
        });
 
        CompletableFuture<String> base = CompletableFuture.supplyAsync(() -> {
            System.out.println("[上游] " + Thread.currentThread().getName());
            return "hello";
        });
 
        // 方式一：thenApply —— 同步，线程不确定
        base.thenApply(s -> {
            System.out.println("[thenApply] " + Thread.currentThread().getName());
            return s.toUpperCase();
        }).get();
 
        // 方式二：thenApplyAsync —— 异步，使用默认 ForkJoinPool
        base.thenApplyAsync(s -> {
            System.out.println("[thenApplyAsync 默认池] " + Thread.currentThread().getName());
            return s.toUpperCase();
        }).get();
 
        // 方式三：thenApplyAsync + 自定义线程池（生产推荐）
        base.thenApplyAsync(s -> {
            System.out.println("[thenApplyAsync 自定义池] " + Thread.currentThread().getName());
            return s.toUpperCase();
        }, myPool).get(); // 传入 myPool 作为 Executor
 
        myPool.shutdown(); // 优雅关闭自定义线程池
    }
}

[上游] ForkJoinPool.commonPool-worker-1
[thenApply] main
[thenApplyAsync 默认池] ForkJoinPool.commonPool-worker-1
[thenApplyAsync 自定义池] my-pool-23

// 生产级线程池配置示例
ExecutorService bizPool = new ThreadPoolExecutor(
        4,                                      // 核心线程数
        8,                                      // 最大线程数
        60L, TimeUnit.SECONDS,                  // 空闲线程存活时间
        new LinkedBlockingQueue<>(100),          // 有界队列，防止 OOM
        new ThreadFactoryBuilder()               // Guava 的线程工厂（需引入 guava 依赖）
                .setNameFormat("biz-async-%d")   // 线程命名模板
                .build(),
        new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：由调用者线程执行
);

// 根据用户 ID 异步查询用户名
CompletableFuture<String> getUserName(Long userId) { ... }
 
// 根据用户名异步查询信用评分
CompletableFuture<Integer> getCreditScore(String userName) { ... }

// ❌ 错误示范：thenApply 导致嵌套
CompletableFuture<CompletableFuture<Integer>> nested =
        getUserName(1001L)
                .thenApply(name -> getCreditScore(name)); // Function 返回的是 CF<Integer>
// 结果类型变成了 CF<CF<Integer>>，需要两次 get() 才能取到值——非常丑陋！

// thenCompose 的签名
public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)

import java.util.concurrent.CompletableFuture;
 
public class ThenComposeDemo {
 
    // 模拟：根据用户 ID 异步查询用户名
    static CompletableFuture<String> getUserName(Long userId) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("查询用户名, userId=" + userId
                    + ", thread=" + Thread.currentThread().getName());
            return "Alice"; // 模拟返回用户名
        });
    }
 
    // 模拟：根据用户名异步查询信用评分
    static CompletableFuture<Integer> getCreditScore(String userName) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("查询信用分, userName=" + userName
                    + ", thread=" + Thread.currentThread().getName());
            return 750; // 模拟返回信用分
        });
    }
 
    public static void main(String[] args) throws Exception {
 
        // ===== 方式一：thenApply —— 产生嵌套（不推荐） =====
        CompletableFuture<CompletableFuture<Integer>> nestedFuture =
                getUserName(1001L)
                        .thenApply(name -> getCreditScore(name)); // 返回 CF<CF<Integer>>
 
        // 需要两次 get() 才能拿到最终值
        Integer score1 = nestedFuture.get().get(); // 丑陋的双重解包
        System.out.println("嵌套方式得分: " + score1);
 
        System.out.println("----------");
 
        // ===== 方式二：thenCompose —— 自动扁平化（推荐） =====
        CompletableFuture<Integer> flatFuture =
                getUserName(1001L)
                        .thenCompose(name -> getCreditScore(name)); // 返回 CF<Integer>，自动解包
 
        // 只需要一次 get()
        Integer score2 = flatFuture.get(); // 清爽！
        System.out.println("扁平化方式得分: " + score2);
    }
}

查询用户名, userId=1001, thread=ForkJoinPool.commonPool-worker-1
查询信用分, userName=Alice, thread=ForkJoinPool.commonPool-worker-1
嵌套方式得分: 750
----------
查询用户名, userId=1001, thread=ForkJoinPool.commonPool-worker-1
查询信用分, userName=Alice, thread=ForkJoinPool.commonPool-worker-2
扁平化方式得分: 750

import java.util.concurrent.CompletableFuture;
 
public class MicroserviceChainDemo {
 
    // 模拟：调用订单服务，根据用户 ID 获取最近订单号
    static CompletableFuture<String> getLatestOrderId(Long userId) {
        return CompletableFuture.supplyAsync(() -> "ORD-20250224-" + userId);
    }
 
    // 模拟：调用物流服务，根据订单号获取物流状态
    static CompletableFuture<String> getShippingStatus(String orderId) {
        return CompletableFuture.supplyAsync(() -> orderId + " -> 已发货，预计明天送达");
    }
 
    // 模拟：调用通知服务，发送推送消息
    static CompletableFuture<Boolean> sendNotification(String message) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("发送通知: " + message);
            return true; // 发送成功
        });
    }
 
    public static void main(String[] args) throws Exception {
 
        // 完整的异步链：用户ID → 订单号 → 物流状态 → 发送通知
        CompletableFuture<Boolean> pipeline =
                getLatestOrderId(1001L)                          // Stage 1: CF<String>   获取订单号
                        .thenCompose(orderId ->                  // 扁平化串联
                                getShippingStatus(orderId))      // Stage 2: CF<String>   获取物流状态
                        .thenCompose(status ->                   // 再次扁平化串联
                                sendNotification(status));       // Stage 3: CF<Boolean>  发送通知
 
        Boolean success = pipeline.get();                        // 阻塞获取最终结果
        System.out.println("通知发送结果: " + success);         // 输出: 通知发送结果: true
    }
}

// 同步版本：Function 在完成线程或调用线程上执行
public <U> CompletableFuture<U> thenCompose(Function<T, ? extends CompletionStage<U>> fn)
 
// 异步版本：Function 提交到默认 ForkJoinPool 执行
public <U> CompletableFuture<U> thenComposeAsync(Function<T, ? extends CompletionStage<U>> fn)
 
// 异步版本 + 自定义线程池
public <U> CompletableFuture<U> thenComposeAsync(Function<T, ? extends CompletionStage<U>> fn, Executor executor)

CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> 42);
 
// 使用 thenApply，Function 返回了一个 CompletableFuture<String>
var result = cf1.thenApply(num ->
        CompletableFuture.supplyAsync(() -> "Value: " + num));

// 接受一个 Consumer<T> 函数式接口
// Consumer 的特征：接收一个参数，没有返回值 (void accept(T t))
public CompletableFuture<Void> thenAccept(Consumer<? super T> action)

import java.util.concurrent.CompletableFuture;
 
public class ThenAcceptDemo {
    public static void main(String[] args) throws Exception {
 
        // 1. supplyAsync: 模拟异步查询用户信息，返回用户名
        CompletableFuture<String> userFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询线程: " + Thread.currentThread().getName()); // 打印执行线程
            try {
                Thread.sleep(500); // 模拟耗时的数据库查询
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 恢复中断状态
            }
            return "Alice"; // 返回查询结果
        });
 
        // 2. thenAccept: 拿到用户名后，执行消费动作（打印欢迎信息）
        //    注意：这里的 Lambda 参数 name 就是上游返回的 "Alice"
        CompletableFuture<Void> resultFuture = userFuture.thenAccept(name -> {
            // name 即为上游 supplyAsync 的返回值 "Alice"
            System.out.println("消费线程: " + Thread.currentThread().getName()); // 打印消费线程
            System.out.println("欢迎回来, " + name + "!");                      // 使用结果执行副作用
        });
 
        // 3. 阻塞等待整条链完成（生产代码中通常不这样做）
        resultFuture.get(); // get() 返回 null, 因为 thenAccept 返回 Void
        System.out.println("返回值: " + resultFuture.get()); // 输出: 返回值: null
    }
}

// thenApply: T → R, 有输入有输出, 链条继续携带数据
CompletableFuture<Integer> lengthFuture = userFuture.thenApply(name -> {
    return name.length(); // "Alice" → 5, 数据继续向下流动
});
 
// thenAccept: T → void, 有输入无输出, 链条数据到此终止
CompletableFuture<Void> voidFuture = userFuture.thenAccept(name -> {
    System.out.println(name); // 仅消费, 不产出新值
});

// 默认版本：可能在当前线程或上游线程执行
userFuture.thenAccept(name -> { /* ... */ });
 
// Async 版本：强制在 ForkJoinPool.commonPool 中执行
userFuture.thenAcceptAsync(name -> { /* ... */ });
 
// Async + 自定义线程池：强制在指定的 Executor 中执行
userFuture.thenAcceptAsync(name -> { /* ... */ }, myExecutor);

// 接受一个 Runnable: 没有参数，没有返回值
public CompletableFuture<Void> thenRun(Runnable action)

import java.util.concurrent.CompletableFuture;
 
public class ThenRunDemo {
    public static void main(String[] args) throws Exception {
 
        // 1. 模拟一个异步的文件上传任务
        CompletableFuture<String> uploadFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("[上传线程] " + Thread.currentThread().getName()); // 打印线程名
            try {
                Thread.sleep(800); // 模拟文件上传耗时
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 恢复中断
            }
            return "file_2024_report.pdf"; // 返回上传后的文件名
        });
 
        // 2. thenAccept: 消费上传结果, 打印文件名
        CompletableFuture<Void> logFuture = uploadFuture.thenAccept(fileName -> {
            System.out.println("[消费] 文件已上传: " + fileName); // 需要用到上游的 fileName
        });
 
        // 3. thenRun: 在 thenAccept 之后执行, 但完全不关心之前的值
        //    注意: thenRun 的 Runnable 没有任何参数
        CompletableFuture<Void> cleanupFuture = logFuture.thenRun(() -> {
            // 这里拿不到任何上游的数据（fileName 在这里不可见）
            System.out.println("[清理] 临时缓存已清除");           // 做一些与结果无关的善后工作
            System.out.println("[清理线程] " + Thread.currentThread().getName()); // 打印线程
        });
 
        // 4. 等待整条链完成
        cleanupFuture.get();
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class ConsumeChainDemo {
    public static void main(String[] args) throws Exception {
 
        // 创建专用线程池, 避免污染公共池
        ExecutorService bizPool = Executors.newFixedThreadPool(4);
 
        CompletableFuture
            // Step 1: 异步下单, 返回订单号
            .supplyAsync(() -> {
                System.out.println("1. 创建订单 - " + Thread.currentThread().getName());
                return "ORD-20240101-0042"; // 模拟返回订单号
            }, bizPool)
 
            // Step 2: 拿到订单号, 转换为库存扣减结果 (thenApply: T → R)
            .thenApplyAsync(orderId -> {
                System.out.println("2. 扣减库存 - 订单: " + orderId);
                return orderId + " | 库存已扣减"; // 返回处理后的信息
            }, bizPool)
 
            // Step 3: 拿到扣减结果, 发送通知 (thenAccept: T → void)
            //         这是一个典型的"消费"场景: 用数据做事, 但不产出新数据
            .thenAcceptAsync(result -> {
                System.out.println("3. 发送通知 - " + result); // 消费 result, 发短信/推送
            }, bizPool)
 
            // Step 4: 通知发完了, 记一条日志 (thenRun: () → void)
            //         不关心之前的任何结果, 只要前面都完成了就行
            .thenRunAsync(() -> {
                System.out.println("4. 记录审计日志 - 流程全部完成");  // 纯善后动作
            }, bizPool)
 
            // 阻塞等待整条链执行完毕
            .get();
 
        bizPool.shutdown(); // 关闭线程池
    }
}

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "Hello");
 
CompletableFuture<Void> a = cf.thenAccept(s -> System.out.println("[A] " + s));
 
CompletableFuture<Void> b = a.thenRun(() -> System.out.println("[B] done"));
 
b.get();

// stage: 另一个要等待的 CompletableFuture
// fn:    接收两个结果、返回新值的函数
public <U, V> CompletableFuture<V> thenCombine(
    CompletionStage<? extends U> other,
    BiFunction<? super T, ? super U, ? extends V> fn
)

public class ThenCombineDemo {
    public static void main(String[] args) throws Exception {
 
        // 异步任务1：模拟从价格服务获取原价（耗时 1 秒）
        CompletableFuture<Double> priceFuture = CompletableFuture.supplyAsync(() -> {
            sleep(1000); // 模拟网络延迟
            System.out.println("价格服务返回: 原价 299.0");
            return 299.0; // 原价
        });
 
        // 异步任务2：模拟从折扣服务获取折扣率（耗时 1.2 秒）
        CompletableFuture<Double> discountFuture = CompletableFuture.supplyAsync(() -> {
            sleep(1200); // 模拟网络延迟
            System.out.println("折扣服务返回: 折扣 0.85");
            return 0.85; // 八五折
        });
 
        // thenCombine: 等待两个任务都完成后，用 BiFunction 合并结果
        CompletableFuture<String> resultFuture = priceFuture.thenCombine(
            discountFuture,                          // other: 另一个要等待的 CF
            (price, discount) -> {                   // BiFunction: 接收两个结果
                double finalPrice = price * discount;// 计算实付价
                return String.format("原价: %.1f, 折扣: %.0f%%, 实付: %.1f",
                        price, discount * 100, finalPrice);
            }
        );
 
        // 阻塞获取最终结果（实际项目中应避免阻塞）
        System.out.println(resultFuture.get());
        // 输出: 原价: 299.0, 折扣: 85%, 实付: 254.2
    }
 
    // 辅助方法：线程睡眠
    private static void sleep(long millis) {
        try { Thread.sleep(millis); } catch (InterruptedException e) { /* ignore */ }
    }
}

// cf1, cf2, cf3 三个并行任务
CompletableFuture<String> merged = cf1
    .thenCombine(cf2, (r1, r2) -> r1 + "," + r2)  // 先合并前两个的结果
    .thenCombine(cf3, (r12, r3) -> r12 + "," + r3); // 再合并第三个

// 等两个都完成后，用 BiConsumer 消费两个结果，没有返回值
public <U> CompletableFuture<Void> thenAcceptBoth(
    CompletionStage<? extends U> other,
    BiConsumer<? super T, ? super U> action
)

public class ThenAcceptBothDemo {
    public static void main(String[] args) throws Exception {
 
        // 异步查询用户名
        CompletableFuture<String> userFuture = CompletableFuture.supplyAsync(() -> {
            sleep(800);                              // 模拟查询延迟
            return "张三";                            // 返回用户名
        });
 
        // 异步查询订单号
        CompletableFuture<String> orderFuture = CompletableFuture.supplyAsync(() -> {
            sleep(600);                              // 模拟查询延迟
            return "ORD-20250224-0001";              // 返回订单号
        });
 
        // thenAcceptBoth: 两个都完成后，执行消费逻辑（发送通知）
        CompletableFuture<Void> notifyFuture = userFuture.thenAcceptBoth(
            orderFuture,                             // other: 另一个 CF
            (user, order) -> {                       // BiConsumer: 消费两个结果
                // 模拟发送短信通知，不需要返回值
                System.out.println("发送通知 -> 用户: " + user + ", 订单: " + order + " 已发货");
            }
        );
 
        // 等待完成
        notifyFuture.get();
        // 输出: 发送通知 -> 用户: 张三, 订单: ORD-20250224-0001 已发货
    }
 
    private static void sleep(long millis) {
        try { Thread.sleep(millis); } catch (InterruptedException e) { /* ignore */ }
    }
}

// 两个都完成后，执行一个 Runnable，不接收参数也没有返回值
public CompletableFuture<Void> runAfterBoth(
    CompletionStage<?> other,
    Runnable action
)

public class RunAfterBothDemo {
    public static void main(String[] args) throws Exception {
 
        // 异步任务1：初始化数据库连接池
        CompletableFuture<Void> dbInit = CompletableFuture.runAsync(() -> {
            sleep(1000);                             // 模拟初始化耗时
            System.out.println("[DB] 数据库连接池初始化完成");
        });
 
        // 异步任务2：初始化缓存预热
        CompletableFuture<Void> cacheInit = CompletableFuture.runAsync(() -> {
            sleep(1500);                             // 模拟预热耗时
            System.out.println("[Cache] 缓存预热完成");
        });
 
        // runAfterBoth: 两个都完成后，打印就绪日志（不需要任何结果）
        CompletableFuture<Void> ready = dbInit.runAfterBoth(
            cacheInit,                               // other: 另一个 CF
            () -> {                                  // Runnable: 无入参无返回
                System.out.println("=============================");
                System.out.println("所有基础设施初始化完成，系统就绪！");
                System.out.println("=============================");
            }
        );
 
        // 等待就绪
        ready.get();
    }
 
    private static void sleep(long millis) {
        try { Thread.sleep(millis); } catch (InterruptedException e) { /* ignore */ }
    }
}

// 谁先完成，就把谁的结果交给 fn 处理并返回
public <U> CompletableFuture<U> applyToEither(
    CompletionStage<? extends T> other,   // 注意：other 的类型必须和 this 兼容
    Function<? super T, U> fn
)

public class ApplyToEitherDemo {
    public static void main(String[] args) throws Exception {
 
        // 镜像源1：模拟较慢的服务器（2 秒）
        CompletableFuture<String> mirror1 = CompletableFuture.supplyAsync(() -> {
            sleep(2000);                             // 模拟较慢的网络
            System.out.println("[Mirror-1] 返回数据");
            return "data-from-mirror-1";             // 返回数据
        });
 
        // 镜像源2：模拟较快的服务器（800 毫秒）
        CompletableFuture<String> mirror2 = CompletableFuture.supplyAsync(() -> {
            sleep(800);                              // 模拟较快的网络
            System.out.println("[Mirror-2] 返回数据");
            return "data-from-mirror-2";             // 返回数据
        });
 
        // applyToEither: 谁先完成，就对谁的结果进行转换
        CompletableFuture<String> fastest = mirror1.applyToEither(
            mirror2,                                 // other: 另一个竞争者
            (data) -> {                              // Function: 对先到的结果做转换
                return "【已处理】" + data.toUpperCase(); // 转为大写并添加前缀
            }
        );
 
        // 获取结果 —— 总耗时约 800ms，而非 2000ms
        System.out.println(fastest.get());
        // 输出: 【已处理】DATA-FROM-MIRROR-2
    }
 
    private static void sleep(long millis) {
        try { Thread.sleep(millis); } catch (InterruptedException e) { /* ignore */ }
    }
}

// 谁先完成，就消费谁的结果，没有返回值
public CompletableFuture<Void> acceptEither(
    CompletionStage<? extends T> other,
    Consumer<? super T> action
)

public class AcceptEitherDemo {
    public static void main(String[] args) throws Exception {
 
        // 推送渠道1：邮件通知（慢）
        CompletableFuture<String> emailChannel = CompletableFuture.supplyAsync(() -> {
            sleep(3000);                             // 邮件服务较慢
            return "邮件渠道: 验证码 8842";           // 返回通知内容
        });
 
        // 推送渠道2：短信通知（快）
        CompletableFuture<String> smsChannel = CompletableFuture.supplyAsync(() -> {
            sleep(500);                              // 短信服务较快
            return "短信渠道: 验证码 8842";           // 返回通知内容
        });
 
        // acceptEither: 哪个渠道先送达，就显示哪个的结果
        CompletableFuture<Void> done = emailChannel.acceptEither(
            smsChannel,                              // other: 另一个竞争的渠道
            (msg) -> {                               // Consumer: 消费先到的结果
                System.out.println("用户收到通知 -> " + msg);
            }
        );
 
        done.get();
        // 输出: 用户收到通知 -> 短信渠道: 验证码 8842
    }
 
    private static void sleep(long millis) {
        try { Thread.sleep(millis); } catch (InterruptedException e) { /* ignore */ }
    }
}

// 任一完成后，执行 Runnable，不接收结果也没有返回值
public CompletableFuture<Void> runAfterEither(
    CompletionStage<?> other,
    Runnable action
)

public class RunAfterEitherDemo {
    public static void main(String[] args) throws Exception {
 
        // 健康检查1：检测主数据库
        CompletableFuture<Void> checkPrimary = CompletableFuture.runAsync(() -> {
            sleep(2000);                             // 主库响应较慢
            System.out.println("[Primary DB] 健康检查通过");
        });
 
        // 健康检查2：检测从数据库
        CompletableFuture<Void> checkReplica = CompletableFuture.runAsync(() -> {
            sleep(300);                              // 从库响应很快
            System.out.println("[Replica DB] 健康检查通过");
        });
 
        // runAfterEither: 任一检查通过，就标记数据库服务可用
        CompletableFuture<Void> serviceReady = checkPrimary.runAfterEither(
            checkReplica,                            // other: 另一个健康检查
            () -> {                                  // Runnable: 无入参无返回
                System.out.println("✅ 数据库服务可用，系统启动！");
            }
        );
 
        serviceReady.get();
        // 输出:
        // [Replica DB] 健康检查通过
        // ✅ 数据库服务可用，系统启动！
    }
 
    private static void sleep(long millis) {
        try { Thread.sleep(millis); } catch (InterruptedException e) { /* ignore */ }
    }
}

CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
    try { Thread.sleep(100); } catch (Exception e) {}
    return "A";
});
 
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
    try { Thread.sleep(2000); } catch (Exception e) {}
    return "B";
});
 
CompletableFuture<String> result = cf1.applyToEither(cf2, s -> s + "-WINS");
System.out.println(result.join());

// 接收可变参数，返回 CompletableFuture<Void>
// 当所有传入的 CF 都正常完成时，返回的 CF 才会完成
// 如果其中任何一个 CF 异常完成，返回的 CF 也会异常完成
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
 
public class AllOfBasicDemo {
    public static void main(String[] args) {
        // 模拟三个独立的异步任务
        CompletableFuture<String> productCF = CompletableFuture.supplyAsync(() -> {
            sleep(800); // 模拟商品服务耗时 800ms
            return "iPhone 15 Pro";
        });
 
        CompletableFuture<Integer> stockCF = CompletableFuture.supplyAsync(() -> {
            sleep(500); // 模拟库存服务耗时 500ms
            return 128;
        });
 
        CompletableFuture<Double> priceCF = CompletableFuture.supplyAsync(() -> {
            sleep(600); // 模拟价格服务耗时 600ms
            return 7999.0;
        });
 
        // allOf 创建一个"屏障"，等待三个任务全部完成
        // 返回类型是 CompletableFuture<Void>，不直接持有结果
        CompletableFuture<Void> allDone = CompletableFuture.allOf(productCF, stockCF, priceCF);
 
        // 在屏障完成后，通过 thenRun 执行后续逻辑
        allDone.thenRun(() -> {
            // 此时三个 CF 都已完成，join() 不会阻塞，立即返回结果
            String product = productCF.join();   // "iPhone 15 Pro"
            int stock = stockCF.join();           // 128
            double price = priceCF.join();        // 7999.0
 
            // 聚合结果
            System.out.println("商品: " + product);
            System.out.println("库存: " + stock);
            System.out.println("价格: ¥" + price);
        }).join(); // 主线程等待整个流程结束
 
        // 总耗时 ≈ max(800, 500, 600) = 800ms，而非 800+500+600=1900ms
    }
 
    // 辅助方法：安全地休眠指定毫秒
    private static void sleep(long millis) {
        try {
            TimeUnit.MILLISECONDS.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 恢复中断标志
        }
    }
}