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 更容易避免这种等待关系。最后在调用边界取得结果即可,不必每连接一步都阻塞一次。