CompletableFuture 的方法很多,先按任务之间的关系分类:依赖前一个结果、两个独立结果合并、等待一组任务完成。

thenApply 和 thenCompose

thenApply() 做普通转换,函数返回什么,新阶段的结果就是什么。如果函数本身返回 CompletableFuture,就会得到嵌套的 Future。

thenCompose() 用于连接另一个异步阶段,把嵌套关系展开:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ComposeDemo {
    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        try {
            CompletableFuture<String> result = CompletableFuture
                    .supplyAsync(() -> 7, pool)
                    .thenCompose(id -> CompletableFuture.supplyAsync(
                            () -> "user-" + id, pool));
            System.out.println(result.join());
        } finally {
            pool.shutdown();
        }
    }
}

输出 user-7。第二个任务依赖第一个任务得到的 id,它们不能无条件同时开始。

两个独立结果合并

两个任务没有依赖关系时,可以先分别启动,再用 thenCombine 合并:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CombineDemo {
    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        try {
            CompletableFuture<Integer> price = CompletableFuture.supplyAsync(() -> 100, pool);
            CompletableFuture<Integer> freight = CompletableFuture.supplyAsync(() -> 8, pool);
            System.out.println(price.thenCombine(freight, Integer::sum).join());
        } finally {
            pool.shutdown();
        }
    }
}

结果为 108。thenCombine 负责两个结果都正常完成后的合并,它本身不是启动两个线程的命令。

只消费结果、不产生新值时,可以用 thenAccept();不关心前一步结果,只需要在成功后执行动作,可以用 thenRun()。

allOf 与 anyOf

allOf() 返回 CompletableFuture<Void>,表示这组任务的完成,不会替你把结果收集成列表。正常完成后,再分别读取每个 Future 的结果即可。如果有任务异常,这个组合结果也会异常完成,而且不应把它当作自动取消其他任务的机制。

anyOf() 采用率先完成任务的结果,可能是正常值,也可能是异常。它不是“第一个成功结果”。返回类型是 Object,使用时还要留意参与任务的结果类型。

applyToEither() 用于两个同类型阶段中一个正常完成后的转换;涉及异常竞争时,不要假定它一定会忽略先到的异常,耐心等待另一个成功结果。要求这样的业务规则,应当显式收集和判断结果。

不要在池里互相等

如果一个只有一个工作线程的池,正在执行的任务又向同一个池提交子任务并 join,子任务可能永远没有机会执行。

用 thenCompose 表达依赖,让前一个阶段完成后交出线程,比在任务内部反复 get/join 更容易避免这种等待关系。最后在调用边界取得结果即可,不必每连接一步都阻塞一次。

最后修改:2018 年 04 月 29 日
如果觉得我的文章对你有用,请随意赞赏