completablefuture-tutorial.md 25 KB

JDK CompletableFuture API 使用教程

本文所有代码均为自包含、可直接运行的 main() 方法,对应 src/main/java/space/anyi/Cf*.java 文件。 建议边读边跑:mvn exec:java -Dexec.mainClass=space.anyi.Cf01_CreateStage

还记得我们在 JUC 阶段认识的 Future 吗?它是 JDK 5 时代异步任务的代表:

ExecutorService pool = Executors.newFixedThreadPool(4);
Future<String> future = pool.submit(() -> "结果");
String result = future.get();   // 阻塞!拿不到就先卡在这里

用起来总觉得不够爽,痛点相当明显:

  1. get() 是阻塞的:想拿结果只能干等,要么就自己轮询 isDone(),无法"结果好了再通知我"。
  2. 串不起任务:做完"查用户",想接着异步"查订单",Future 做不到,只能再 submit 一次、把上一个结果手动传进去。
  3. 劝不动多个任务:并发调三个接口想一起汇总,Future 只能一个个 get,三个请求等于被"拉直"成串行。
  4. 异常难处理:只能 try-catch 包住 get(),异常传递全靠一层层往外抛。

于是 JDK 8 给 Java 带来了一个对标前端 Promise 思想的神器——CompletableFuture。它把"异步 + 回调 + 组合"揉在一个类里,让你像写流程图一样描述异步逻辑。这一系列,我们就来把它吃透。


一、创建异步任务:runAsync / supplyAsync / completedFuture

先从一个问题开始:我要在另一个线程里干点活儿,怎么把任务"交出去"?

CompletableFuture 提供了三个入口(详见 Cf01_CreateStage):

public static void main(String[] args) throws Exception {
    // runAsync:没有返回值的异步任务,相当于"让个线程去跑一段代码"
    CompletableFuture<Void> voidFuture = CompletableFuture.runAsync(() -> {
        System.out.println("runAsync 正在执行: " + Thread.currentThread().getName());
    });
    voidFuture.join(); // 阻塞等待任务完成

    // supplyAsync:有返回值的异步任务,这是最常用的入口
    CompletableFuture<String> stringFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("supplyAsync 正在执行: " + Thread.currentThread().getName());
        return "Hello CompletableFuture";
    });
    System.out.println("supplyAsync 拿到的结果: " + stringFuture.get());

    // completedFuture:直接得到一个已经完成的任务,无需起线程
    CompletableFuture<Integer> done = CompletableFuture.completedFuture(42);
    System.out.println("completedFuture 的结果: " + done.getNow(-1));

    // getNow:任务已完成就返回结果,未完成不阻塞,直接返回默认值
    CompletableFuture<String> slowTask = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "完成了";
    });
    System.out.println("任务还没完成 getNow 返回: " + slowTask.getNow("默认值"));
    System.out.println("阻塞等待任务结束得到: " + slowTask.join());
}

拿到结果有四种姿势,别一次全用,按需选择一个即可:

方法 行为 什么时候用
get() 阻塞等待,结果以受检异常方式抛出 方法签名本就允许抛异常时
join() 阻塞等待,把异常包装成 CompletionException 扔出 日常首选,不强制 try-catch
getNow(默认值) 已完成返回结果,未完成立即返回默认值,绝不阻塞 想"有就有、没有算了"
get(超时, 单位) 等待限定时间,超时抛 TimeoutException 给外部调用设置超时上限

join()get() 都是阻塞方法,阻塞的是"调用它们的线程"。 学习阶段无妨,但在生产环境的回调里不要随便调用,否则会反向堵塞线程池(详见第九章)。

使用场景:

  • runAsync:发短信、写异步日志、触发一个不用回执的后台任务。
  • supplyAsync:把耗时的 RPC/IO 丢到后台,主线程先去干别的,结果好了再取。
  • completedFuture:在拼接处理链时,给"已知结果"一个统一的起点。

二、串行处理:thenApply / thenAccept / thenRun

我们已经能做到"异步开跑",但现在关心另一个问题:上一个任务的结果,能不能直接喂给下一个任务?

还记得 JavaSE 里学多线程时的"排序加 sleep"吗?那种靠 Thread + 标志位手撕回调的方式,在这里可以用一条链优雅搞定(详见 Cf02_ThenSeries):

public static void main(String[] args) throws Exception {
    CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("1. 创建订单");
        return "订单号 2024001";
    }).thenApply(orderId -> {
        System.out.println("2. 扣减库存,关联订单: " + orderId);
        return orderId + " 已扣库存";
    }).thenApply(info -> {
        System.out.println("3. 记录日志: " + info);
        return info + " 日志已落库";
    }).thenAccept(result -> {
        System.out.println("4. 打印流程结果: " + result);   // 只消费结果,无返回值
    }).thenRun(() -> {
        System.out.println("5. 发送短信通知(不关心数据)");  // 连结果都不关心
    });
    future.join();
}

三个串行方法,对应三种需求:

方法 能拿到什么 要返回什么 对应场景
thenApply 上一个结果 返回新结果(继续传给下一环) 加工、转换,链条式的数据流
thenAccept 上一个结果 无需返回 消费掉,比如打日志、落库、发通知
thenRun 什么都不要 无需返回 只关心"到了这个点",不关心数据

使用场景: 一切"前一步结果喂后一步"的流水线,比如下单流程:创建订单 → 扣库存 → 生成对账单 → 发送通知;或者接口收到请求后:解析参数 → 查询数据 → 组装响应。


三、组合两个任务:thenCombine / thenAcceptBoth / runAfterBoth

串行解决了"先后依赖",但现实里更多是两个互不依赖的任务同时干,最后凑一块儿。比如页面上要同时展示"用户信息"和"订单信息",分两次异步请求发出,谁先回来无所谓,关键是两个都回来后拼在一起展示(详见 Cf03_ThenCombine):

public static void main(String[] args) throws Exception {
    CompletableFuture<String> userFuture = CompletableFuture.supplyAsync(() -> {
        sleep(100);
        return "用户:李雷";
    });
    CompletableFuture<String> orderFuture = CompletableFuture.supplyAsync(() -> {
        sleep(200);
        return "订单:O-10008";
    });

    // thenCombine: 等两个都完成,把两个结果合并,返回一个新结果
    CompletableFuture<String> combined = userFuture.thenCombine(orderFuture,
            (user, order) -> user + " | " + order);
    System.out.println("thenCombine 合并结果: " + combined.join());

    // thenAcceptBoth: 等两个都完成,消费两个结果,但无返回值
    userFuture.thenAcceptBoth(orderFuture, (user, order) ->
            System.out.println("thenAcceptBoth 同时拿到: " + user + " & " + order)).join();

    // runAfterBoth: 等两个都完成,不关心结果,只等一个时间点
    userFuture.runAfterBoth(orderFuture, () ->
            System.out.println("runAfterBoth: 两路结果都就绪")).join();
}

注意看 thenCombine 的回调参数 (user, order) —— 它和 thenApply 的关键区别就在于:回调能不能拿到两个任务的结果。简单记:thenApply(BiFunction) 永远只接收一个 Task 的结果,所以它是 Function 不是 BiFunction;而 thenCombine 接收的是两个

使用场景:

  • thenCombine:并发拉取两个接口,合并字段后返回给前端。
  • thenAcceptBoth:两个异步结果都完成后触发一个副作用(入库、发通知)。
  • runAfterBoth:做"两个条件都满足才能继续"的闸门,比如数据校验 + 权限校验都通过才放行。

四、二选一:applyToEither / acceptEither / runAfterEither

与"两个都要"相对的,是"两个谁先来都行"。典型场景:读数据先走本地缓存(快),缓存没有就等远程(慢),谁先返回用谁,避免干等最慢那一路(详见 Cf04_ThenEither):

public static void main(String[] args) throws Exception {
    CompletableFuture<String> cache = CompletableFuture.supplyAsync(() -> {
        sleep(100);
        return "缓存数据";
    });
    CompletableFuture<String> remote = CompletableFuture.supplyAsync(() -> {
        sleep(400);
        return "远程数据";
    });

    // applyToEither: 谁先完成用谁的结果,并继续用指定函数加工
    CompletableFuture<String> result = cache.applyToEither(remote, data -> "优先使用 -> " + data);
    System.out.println("applyToEither 结果: " + result.join());

    // acceptEither: 谁先完成消费谁的结果,无返回值
    CompletableFuture<String> a = CompletableFuture.supplyAsync(() -> { sleep(50);  return "A"; });
    CompletableFuture<String> b = CompletableFuture.supplyAsync(() -> { sleep(400); return "B"; });
    a.acceptEither(b, s -> System.out.println("acceptEither 选中了先完成的: " + s)).join();

    // runAfterEither: 谁先完成都行,等时间点即可,不关心数据
    CompletableFuture<Void> r1 = CompletableFuture.runAsync(() -> sleep(60));
    CompletableFuture<Void> r2 = CompletableFuture.runAsync(() -> sleep(500));
    r1.runAfterEither(r2, () -> System.out.println("runAfterEither: 有一路先完成了")).join();
    r2.join();   // 等待较慢的一路也结束,避免 main 退出时打断它
}

上一个示例输出 缓存数据 为主,因为线程调度下 cache(sleep 100ms)一定先完成。

使用场景: 缓存与远程的双通道读取、主备服务的容灾切换、多数据源"先到先得"。


五、展平组合:thenCompose

看到这里,你的脑中可能冒出一个问题:如果 thenApply 的回调里,返回的又是一个 CompletableFuture,会怎样?

结果就是嵌套:CompletableFuture<CompletableFuture<String>>。要取最里层真的结果,得 join().join(),层级一深就开始"回调地狱"了。此时应该换用 thenCompose(详见 Cf05_ThenCompose):

public static void main(String[] args) throws Exception {
    CompletableFuture<String> userFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("1. 查询用户");
        return "用户:李雷";
    });

    // 使用 thenApply,返回嵌套结构,需要两层 join 才能取到结果
    CompletableFuture<CompletableFuture<String>> nested =
            userFuture.thenApply(Cf05_ThenCompose::getOrder);
    System.out.println("thenApply 拼出嵌套: " + nested.getClass().getSimpleName());
    System.out.println("  两层 join 取结果: " + nested.join().join());

    // thenCompose 会自动展平内层 Future,链式调用只有一个层级
    CompletableFuture<String> flat = userFuture.thenCompose(Cf05_ThenCompose::getOrder);
    System.out.println("thenCompose 展平结果: " + flat.join());
}

private static CompletableFuture<String> getOrder(String user) {
    return CompletableFuture.supplyAsync(() -> {
        System.out.println("2. 异步查询 \"" + user + "\" 的订单");
        return "订单 O-10008";
    });
}

thenComposethenApply 长得像,但本质不同:

  • thenApply 回调返回普通值或 CompletableFuture原样透传,你给的什么它就是什么。
  • thenCompose 要求回调返回 CompletionStage,并帮你把这一层展平

这和 StreammapflatMap 的关系一模一样,可以类比记忆。

使用场景: 前后依赖、且每一环都是异步的调用链,比如"根据用户ID异步查用户 → 再根据用户异步查其订单 → 再查订单中的商品",贯穿三层以上时 thenCompose 让代码保持扁平。


六、批量聚合:allOf / anyOf

单个任务好办,如果是一任务呢?比如某页面需要同时调"基础信息、库存、优惠"三个接口,全部返回后一起渲染。这时别傻傻写三次 get()——那会退化成串行,应该用 allOf / anyOf(详见 Cf06_AllOfAnyOf):

public static void main(String[] args) throws Exception {
    CompletableFuture<String> baseInfo = CompletableFuture.supplyAsync(() -> {
        sleep(100);
        return "基础信息";
    });
    CompletableFuture<String> stock = CompletableFuture.supplyAsync(() -> {
        sleep(200);
        return "库存信息";
    });
    CompletableFuture<String> coupon = CompletableFuture.supplyAsync(() -> {
        sleep(300);
        return "优惠信息";
    });

    // allOf: 等全部完成(返回 Void),再各自 join 汇总,让三路请求真正并行
    CompletableFuture<Void> all = CompletableFuture.allOf(baseInfo, stock, coupon);
    all.join();
    List<String> data = Arrays.stream(new CompletableFuture[]{baseInfo, stock, coupon})
            .map(CompletableFuture::join)
            .map(Object::toString)
            .collect(Collectors.toList());
    System.out.println("allOf 全部就绪,汇总结果: " + String.join(" | ", data));

    // anyOf: 任一路先完成即返回(结果是 Object),用于"谁先回来先给谁出结果"
    CompletableFuture<Object> any = CompletableFuture.anyOf(baseInfo, stock, coupon);
    System.out.println("anyOf 最先返回的一路: " + any.join());
}

一个很容易掉的坑:allOf 返回的是 CompletableFuture<Void>,本身不带任何汇总结果。 想拿每个任务的结果,得在 allOf().join() 之后再对每个 Future 分别 join()。 但注意,此时它们都已经完成了,再 join 只是"取数据",不会再等。

anyOf 返回类型是 CompletableFuture<Object>,结果要做类型判断/强转回自己的类型。

使用场景:

  • allOf:BFF 层并发聚合多个下游接口、批量文件下载后统一处理、多副本校验。
  • anyOf:多个候选源"谁先出结果就用谁"、健康检查里"只要一路通就视为可用"。

七、异常处理:exceptionally / whenComplete / handle

异步任务挂了怎么办?CompletableFuture 把异常也当作"一种结果"来处理,等着我们用回调接住(详见 Cf07_ExceptionHandling):

public static void main(String[] args) throws Exception {
    // exceptionally: 任务失败时给出降级结果(类似异常被"恢复")
    CompletableFuture<Integer> fallback = failTask().exceptionally(e -> {
        System.out.println("exceptionally 捕获: " + e.getClass().getSimpleName());
        return -1;   // 降级返回
    });
    System.out.println("exceptionally 降级结果: " + fallback.join());

    // whenComplete: 只"观察"结果或异常,做了日志记录就够,
    // 注意它不消化异常 —— 原任务失败,链上的 join/get 依然会抛 CompletionException
    CompletableFuture<Integer> observed = failTask();
    try {
        observed.whenComplete((res, ex) -> {
            if (ex != null) {
                System.out.println("whenComplete 观察到异常: " + ex.getClass().getSimpleName());
            } else {
                System.out.println("whenComplete 观察到结果: " + res);
            }
        }).join();
    } catch (Exception e) {
        System.out.println("whenComplete 不吞异常, join() 抛: " + e.getClass().getSimpleName());
    }

    // handle: 无论成功失败都给一个函数处理,并可返回新结果(可恢复)
    CompletableFuture<String> handled = failTask().handle((res, ex) ->
            ex == null ? "正常结果: " + res : "出错被恢复: " + ex.getMessage());
    System.out.println("handle 结果: " + handled.join());
}

三个方法放一起对比,区别一目了然:

方法 成功时触发? 失败时触发? 能拿到异常? 能改变结果? 会不会吞掉异常
exceptionally ✗(成功就跳过) ✓ 返回降级值
whenComplete ✗ 只观察 不会
handle ✓ 可恢复

关键记忆点:回调里出现异常时,所有完成入口抛的都是 CompletionException.getCause() 才是原始异常

使用场景:

  • exceptionally:调用外部服务失败后返回兜底默认值,或记录并返回空数据。
  • whenComplete:收尾日志、打监控埋点,看看"这任务到底成了还是挂了"。
  • handle:最灵活,"成功就取结果、失败就换方案",适合做降级/默认值策略。

八、手动完成:complete / completeExceptionally / obtrudeValue / cancel

前面聊的都是"任务自己往结束跑",但异步世界里还有一种情况:结果的给出,不由任务决定,而由外部决定。比如某个阶段要等用户点击、等外部回调、等一个守护线程汇报(详见 Cf08_ManualComplete):

public static void main(String[] args) throws Exception {
    // complete: 任务的完成时机由我们手动决定(比如等待外部事件/用户操作)
    CompletableFuture<String> gate = new CompletableFuture<>();
    new Thread(() -> {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        gate.complete("外部线程触发了完成");
    }).start();
    System.out.println("阻塞等待手动完成: " + gate.join());

    // completeExceptionally: 手动让任务以异常结束,常用于"超时兜底"
    CompletableFuture<Integer> timeout = new CompletableFuture<>();
    timeout.completeExceptionally(new RuntimeException("调用超时,兜底结束"));
    try {
        timeout.join();
    } catch (Exception e) {
        System.out.println("completeExceptionally 抛给调用方: " + e.getCause());
    }

    // obtrudeValue: 无论之前是否已完成,强制覆盖为新值(慎用)
    CompletableFuture<String> forced = CompletableFuture.completedFuture("旧结果");
    forced.obtrudeValue("强制更新的结果");
    System.out.println("obtrudeValue 覆盖后: " + forced.join());

    // cancel: 取消任务, isCancelled 与 isCompletedExceptionally 都会为 true
    CompletableFuture<String> cancelling = new CompletableFuture<>();
    System.out.println("cancel() 返回值: " + cancelling.cancel(true));
    System.out.println("取消后 isCancelled: " + cancelling.isCancelled()
            + ", isCompletedExceptionally: " + cancelling.isCompletedExceptionally());
}

先说结论:complete 期待任务还没有完成;obtrudeValue 则不管状态强制覆盖,一般不要在生产乱用,否则正常逻辑都会被它"篡改"。

使用场景:

  • new CompletableFuture<>() 当"闸门/栅栏",等某个外部事件(消息推送、第三方回调、人工审核)到来后再 complete,让等待方放行。
  • 配合超时:等待超时后调 completeExceptionally 快速失败,别让调用方无限挂起。
  • cancel 用于显式放弃一个已无意义的任务,例如用户离开了页面后取消剩余请求。

九、线程池与 Async 变体

以上所有 xxxAsync 方法,我们没有特地指定线程池时,它们最终都跑在同一个地方——ForkJoinPool.commonPool()。这会带来两个大坑,必须提前讲清楚(详见 Cf09_CustomExecutor):

public static void main(String[] args) throws Exception {
    // 不指定线程池时, 默认全部任务都跑在 ForkJoinPool.commonPool 上
    CompletableFuture<String> defaultOne = CompletableFuture.supplyAsync(() -> {
        System.out.println("默认线程池: " + Thread.currentThread().getName());
        return "OK";
    });
    defaultOne.join();

    // 重活建议显式指定自定义线程池, 避免抢占公共池
    ExecutorService pool = Executors.newFixedThreadPool(4);
    CompletableFuture<String> custom = CompletableFuture.supplyAsync(() -> {
        System.out.println("自定义池 supplyAsync: " + Thread.currentThread().getName());
        return "结果";
    }, pool).thenApplyAsync(r -> {
        System.out.println("自定义池 thenApplyAsync: " + Thread.currentThread().getName());
        return r + " 加工完成";
    }, pool);
    System.out.println("最终结果: " + custom.join());
    pool.shutdown();

    // 关键区别: 不带 Async 后缀的方法, 新任务在哪执行由"触发者"所在的线程决定,
    // 一般沿用前一个任务的执行线程; 带 Async 且不传池子时, 提交到 commonPool
    CompletableFuture<String> noAsync = CompletableFuture.supplyAsync(() -> {
        System.out.println("supplyAsync        : " + Thread.currentThread().getName());
        return "r";
    }).thenApply(r -> {
        System.out.println("thenApply(无Async) : " + Thread.currentThread().getName());
        return r;
    }).thenApplyAsync(r -> {
        System.out.println("thenApplyAsync(无池): " + Thread.currentThread().getName());
        return r;
    });
    noAsync.join();
}

一次典型运行的输出(线程名因环境而异,但规律一致):

默认线程池: ForkJoinPool.commonPool-worker-1
自定义池 supplyAsync: pool-1-thread-1
自定义池 thenApplyAsync: pool-1-thread-2
最终结果: 结果 加工完成
supplyAsync        : ForkJoinPool.commonPool-worker-1
thenApply(无Async) : space.anyi.Cf09_CustomExecutor.main()
thenApplyAsync(无池): ForkJoinPool.commonPool-worker-1

这里藏着 CompletableFuture 最容易踩的两个坑:

坑一:commonPool 被阻塞任务饿死(Thread Starvation) commonPool 的线程数默认是 CPU核心数 - 1。当你在回调里再嵌套 supplyAsync 时,外层任务会占用 worker 等待,而内层任务又在排队等 worker 执行,互等,最终整个池子卡死。 生产环境里 RPC/HTTP/DB 等着始终受限,绝不要做不可控的阻塞操作。解决:给耗时的 I/O 调用显式传入一个独立的 ThreadPoolExecutor

坑二:Async 后缀 ≠ 线程池更大,只是"换个池子提交" thenApply(无 Async)在哪个线程执行,取决于谁触发了它——往往沿用前一个任务的线程,甚至主线程(见上面输出第 4 行)。 所以别指望"加上 Async 就更快",它更多是在让你指定提交的线程池

使用场景与建议:

  • 简单、短暂、CPU 密集的任务可以用默认池,省事。
  • 遇到 I/O 阻塞、长耗时任务,务必 new ThreadPoolExecutor(...)(或固定大小池)并传给带 Async 的方法。
  • 回调里切忌调用 join()/get() 等阻塞点完成自己,这是服务生产事故的高发区。

十、总结与速查

至此,CompletableFuture 的核心骨架我们都过了一遍。把它当作一张"异步流程图"来记:

需求 方法 回调形态 返回
无返回值异步任务 runAsync Runnable CompletableFuture<Void>
有返回值异步任务 supplyAsync Supplier CompletableFuture<T>
拿来即用 completedFuture CompletableFuture<T>
手拿上一个结果继续加工 thenApply Function CompletableFuture<U>
消费上一个结果 thenAccept Consumer CompletableFuture<Void>
等上一环跑完即可 thenRun Runnable CompletableFuture<Void>
两个都完成,合并结果 thenCombine BiFunction CompletableFuture<U>
两个都完成,消费结果 thenAcceptBoth BiConsumer CompletableFuture<Void>
两个都完成,不等数据 runAfterBoth Runnable CompletableFuture<Void>
谁先完成,加工谁的结果 applyToEither Function CompletableFuture<U>
谁先完成,消费谁的结果 acceptEither Consumer CompletableFuture<Void>
谁先完成,只等时间点 runAfterEither Runnable CompletableFuture<Void>
展平嵌套的异步调用 thenCompose Function→CompletionStage CompletableFuture<U>
全部完成(各自再 join 汇总) allOf CompletableFuture<Void>
任一完成(结果需转类型) anyOf CompletableFuture<Object>
失败时降级 exceptionally Function CompletableFuture<T>
成功失败都观察(不吞异常) whenComplete BiConsumer 不可改结果
成功失败都可处理(可恢复) handle BiFunction CompletableFuture<U>
手动设定结果 complete boolean
手动以异常结束 completeExceptionally boolean
强制覆盖结果(慎用) obtrudeValue void
取消任务 cancel boolean

最后,把最容易犯的错再点一遍:

  1. 默认线程池是 ForkJoinPool.commonPool,阻塞任务会把它饿死,I/O 要交给独立线程池。
  2. allOf 返回的是 Void,汇总需自己逐个 join()
  3. whenComplete 不消化异常,异常会继续沿着链传下去。
  4. 异常都会包装成 CompletionException,记得用 getCause() 取原始异常。
  5. 回调里不要 join()/get() 阻塞自己的执行线程。
  6. 非 Async 后缀的方法在"触发线程"上执行,别脑补它在公共池上跑。

版权声明:本文由 computrueFeture-demo 项目产出,适用于学习交流。