37. thenCompose：一个异步任务依赖另一个异步任务
发布于 • 阅读量 0
37. thenCompose：一个异步任务依赖另一个异步任务
上一节讲了 thenCombine()。
它适合这种场景：
任务一和任务二互不依赖；
两个任务可以同时执行；
等两个任务都完成以后合并结果。
这一节看另一个很常见的场景：
第二个异步任务必须等待第一个异步任务完成；
并且第二个任务需要使用第一个任务的结果。
比如 PDF 水印项目里：
第一步：给 PDF 添加水印，返回水印文件路径；
第二步：把水印文件上传到远程存储，返回下载地址。
上传任务不能提前执行。
因为在水印文件生成之前，我连要上传哪个文件都不知道。
这种前后依赖的异步任务，就适合使用：
thenCompose(...)
thenCompose 解决什么问题
先看两个异步方法。
第一个方法处理 PDF：
private static CompletableFuture<String> addWatermarkAsync(File file) {
return CompletableFuture.supplyAsync(() -> {
return "output/test-watermark.pdf";
});
}
第二个方法上传 PDF：
private static CompletableFuture<String> uploadAsync(String targetPath) {
return CompletableFuture.supplyAsync(() -> {
return "https://file.example.com/test-watermark.pdf";
});
}
上传方法需要水印文件路径。
所以执行顺序必须是：
addWatermarkAsync(file)
↓
拿到 targetPath
↓
uploadAsync(targetPath)
两个步骤都是异步任务。
这时候可以写：
CompletableFuture<String> future = addWatermarkAsync(file)
.thenCompose(targetPath -> uploadAsync(targetPath));
最后：
String downloadUrl = future.join();
拿到的就是上传后的下载地址。
一个最简单的 thenCompose 示例
新建类：
com.succos.completablefuture.ThenComposeDemo
代码如下：
package com.succos.completablefuture;
import java.util.concurrent.CompletableFuture;
public class ThenComposeDemo {
public static void main(String[] args) {
CompletableFuture<String> resultFuture =
queryUserIdAsync("张三")
.thenCompose(userId -> {
return queryUserNameAsync(userId);
});
String result = resultFuture.join();
System.out.println("最终结果：" + result);
}
private static CompletableFuture<Integer> queryUserIdAsync(
String keyword
) {
return CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName()
+ " 开始查询用户 ID");
sleep(2000);
System.out.println(Thread.currentThread().getName()
+ " 用户 ID 查询完成");
return 1001;
});
}
private static CompletableFuture<String> queryUserNameAsync(
Integer userId
) {
return CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName()
+ " 根据用户 ID 查询姓名："
+ userId);
sleep(2000);
return "张三";
});
}
private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("任务被中断", e);
}
}
}
这里的执行顺序很明确：
先查询用户 ID；
拿到用户 ID 以后；
再根据 ID 查询用户姓名。
第二个任务不能和第一个任务同时执行，因为它需要第一个任务返回的 userId。
thenCompose 的 Lambda 返回 CompletableFuture
这一点是理解 thenCompose() 的关键。
这里写的是：
.thenCompose(userId -> {
return queryUserNameAsync(userId);
});
queryUserNameAsync() 返回的是：
CompletableFuture<String>
所以 thenCompose() 里面返回的不是普通字符串，而是另一个异步任务。
最终得到的类型仍然是：
CompletableFuture<String>
也就是说，thenCompose() 把两个前后依赖的异步任务连接成了一条流程。
PDF 水印和文件上传示例
回到 PDF 项目。
流程是：
添加水印；
拿到水印文件路径；
上传到远程存储；
拿到远程下载地址。
新建类：
com.succos.completablefuture.ThenComposePdfDemo
代码如下：
package com.succos.completablefuture;
import java.io.File;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class ThenComposePdfDemo {
public static void main(String[] args) {
ThreadPoolExecutor pdfExecutor = new ThreadPoolExecutor(
3,
3,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory("pdf-worker"),
new ThreadPoolExecutor.CallerRunsPolicy()
);
ThreadPoolExecutor uploadExecutor = new ThreadPoolExecutor(
3,
3,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory("upload-worker"),
new ThreadPoolExecutor.CallerRunsPolicy()
);
File file = new File("input/test.pdf");
CompletableFuture<String> resultFuture =
addWatermarkAsync(file, pdfExecutor)
.thenCompose(targetPath -> {
return uploadAsync(
targetPath,
uploadExecutor
);
});
String downloadUrl = resultFuture.join();
System.out.println("--------------------------------");
System.out.println("最终下载地址：" + downloadUrl);
pdfExecutor.shutdown();
uploadExecutor.shutdown();
}
private static CompletableFuture<String> addWatermarkAsync(
File file,
ThreadPoolExecutor pdfExecutor
) {
return CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName()
+ " 开始添加水印："
+ file.getName());
sleep(3000);
String targetPath = "output/"
+ file.getName()
.replace(".pdf", "-watermark.pdf");
System.out.println(Thread.currentThread().getName()
+ " 水印处理完成："
+ targetPath);
return targetPath;
}, pdfExecutor);
}
private static CompletableFuture<String> uploadAsync(
String targetPath,
ThreadPoolExecutor uploadExecutor
) {
return CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName()
+ " 开始上传："
+ targetPath);
sleep(2000);
String downloadUrl =
"https://file.example.com/test-watermark.pdf";
System.out.println(Thread.currentThread().getName()
+ " 上传完成："
+ downloadUrl);
return downloadUrl;
}, uploadExecutor);
}
private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("任务被中断", e);
}
}
static class NamedThreadFactory implements ThreadFactory {
private final String prefix;
private final AtomicInteger threadNumber =
new AtomicInteger(1);
public NamedThreadFactory(String prefix) {
this.prefix = prefix;
}
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName(
prefix
+ "-"
+ threadNumber.getAndIncrement()
);
return thread;
}
}
}
运行时可能看到：
pdf-worker-1 开始添加水印：test.pdf
等待 3 秒...
pdf-worker-1 水印处理完成：output/test-watermark.pdf
upload-worker-1 开始上传：output/test-watermark.pdf
等待 2 秒...
upload-worker-1 上传完成：https://file.example.com/test-watermark.pdf
最终下载地址：https://file.example.com/test-watermark.pdf
这里能看出：
上传任务没有提前开始；
必须等水印任务返回文件路径以后；
才会进入 uploadExecutor 执行上传任务。
为什么不用 thenApply
看到这里可能会有一个问题。
前面已经学过 thenApply()，它也能拿到上一步的结果。
那为什么不这样写：
CompletableFuture<CompletableFuture<String>> future =
addWatermarkAsync(file, pdfExecutor)
.thenApply(targetPath -> {
return uploadAsync(
targetPath,
uploadExecutor
);
});
这段代码可以编译。
但它的返回类型变成了：
CompletableFuture<CompletableFuture<String>>
也就是：
一个 CompletableFuture 里面；
又包了一个 CompletableFuture。
想拿到真正下载地址，就要写：
String downloadUrl = future
.join()
.join();
需要连续 join() 两次。
这就不太自然。
thenApply 会产生嵌套 Future
先看结构：
addWatermarkAsync(...)
.thenApply(targetPath -> uploadAsync(targetPath));
上一步返回：
String targetPath
Lambda 里返回：
CompletableFuture<String>
而 thenApply() 的作用是：
把 Lambda 的返回值当成普通结果包装起来。
所以最终类型就是：
CompletableFuture<CompletableFuture<String>>
可以把它想成：
外层 Future 完成后；
得到的结果不是下载地址；
而是另一个 Future。
所以还要再等一次。
thenCompose 会自动展开嵌套
如果换成：
addWatermarkAsync(...)
.thenCompose(targetPath -> uploadAsync(targetPath));
最终类型就是：
CompletableFuture<String>
thenCompose() 会把内层的 CompletableFuture<String> 展开。
可以简单理解成：
thenApply：
CompletableFuture<CompletableFuture<String>>
thenCompose：
CompletableFuture<String>
这就是 thenCompose() 存在的意义。
它专门处理“上一步完成后，再返回一个新的异步任务”的场景。
thenApply 和 thenCompose 怎么区分
我现在主要看 Lambda 返回什么。
如果 Lambda 返回普通结果：
.thenApply(path -> {
return buildDownloadUrl(path);
})
其中：
buildDownloadUrl(path)
返回普通 String。
就用 thenApply()。
如果 Lambda 返回另一个异步任务：
.thenCompose(path -> {
return uploadAsync(path);
})
其中：
uploadAsync(path)
返回 CompletableFuture<String>。
就用 thenCompose()。
可以直接记成：
返回普通值：thenApply；
返回 CompletableFuture：thenCompose。
一个直观对比
普通转换：
CompletableFuture<String> urlFuture =
pathFuture.thenApply(path -> {
return "http://localhost/" + path;
});
这里是：
String
↓
String
用 thenApply()。
异步转换：
CompletableFuture<String> urlFuture =
pathFuture.thenCompose(path -> {
return uploadAsync(path);
});
这里是：
String
↓
CompletableFuture<String>
用 thenCompose()。
thenCompose 和 thenCombine 的区别
这两个方法名字比较像，但场景完全不同。
thenCombine() 是两个独立任务：
任务一 ─┐
├─ 都完成后合并结果
任务二 ─┘
例如：
处理 PDF；
查询用户信息；
两个任务可以同时执行。
代码：
pdfFuture.thenCombine(
userFuture,
(path, userName) -> {
return userName + "：" + path;
}
);
而 thenCompose() 是前后依赖：
任务一完成
↓
拿到任务一结果
↓
才能创建或执行任务二
例如：
处理 PDF；
拿到文件路径；
才能上传这个文件。
代码：
pdfFuture.thenCompose(
path -> uploadAsync(path)
);
我会这样区分：
两个任务能不能同时开始？
能：可能适合 thenCombine；
不能，第二个依赖第一个结果：用 thenCompose。
thenComposeAsync 的区别
thenCompose() 也有带 Async 的版本：
thenComposeAsync(...)
还可以指定执行器：
thenComposeAsync(
result -> nextAsyncTask(result),
executor
)
不过这里要分清两个执行过程。
例如：
pdfFuture.thenComposeAsync(
targetPath -> uploadAsync(
targetPath,
uploadExecutor
),
composeExecutor
);
composeExecutor 控制的是：
创建和返回 uploadAsync 这个后续任务的 Lambda 在哪里执行。
而 uploadAsync() 内部真正的上传任务，还是由它自己指定的 uploadExecutor 执行。
如果 Lambda 里只是调用一下 uploadAsync()，逻辑非常轻，通常没有必要额外使用 thenComposeAsync()。
直接写：
thenCompose(targetPath -> uploadAsync(
targetPath,
uploadExecutor
))
就够了。
不要为了异步而重复包装
有时会看到这种写法：
CompletableFuture
.supplyAsync(() -> addWatermark(file), pdfExecutor)
.thenCompose(targetPath -> {
return CompletableFuture.supplyAsync(() -> {
return upload(targetPath);
}, uploadExecutor);
});
这本身没问题。
因为第二步确实是另一个异步任务。
但如果 uploadAsync() 已经封装好了：
private CompletableFuture<String> uploadAsync(
String path
) {
return CompletableFuture.supplyAsync(() -> {
return upload(path);
}, uploadExecutor);
}
那外面直接写：
.thenCompose(this::uploadAsync)
就可以了。
没有必要层层再包一遍。
方法引用写法
假设方法是：
private static CompletableFuture<String> uploadAsync(
String targetPath
) {
return CompletableFuture.supplyAsync(() -> {
return "https://file.example.com/test.pdf";
});
}
原来的写法：
.thenCompose(targetPath -> {
return uploadAsync(targetPath);
})
可以简化成：
.thenCompose(
ThenComposePdfDemo::uploadAsync
)
如果是实例方法，也可以写：
.thenCompose(this::uploadAsync)
不过学习阶段我觉得 Lambda 更直观。
能明确看到：
上一步的 targetPath；
被传给了 uploadAsync。
多级异步依赖
thenCompose() 可以连续使用。
例如完整 PDF 流程：
添加水印；
上传文件；
保存数据库；
发送通知。
假设每一步都返回 CompletableFuture，可以写成：
CompletableFuture<NotifyResult> future =
addWatermarkAsync(file)
.thenCompose(targetPath -> {
return uploadAsync(targetPath);
})
.thenCompose(downloadUrl -> {
return saveTaskAsync(downloadUrl);
})
.thenCompose(taskId -> {
return notifyUserAsync(taskId);
});
整个流程是串行依赖的。
但是每一步本身都由各自的线程池异步执行。
最后：
NotifyResult result = future.join();
就能拿到整条异步链的最终结果。
异步不等于并行
这里需要再强调一次。
这几个任务虽然都是异步任务：
添加水印；
上传文件；
保存数据库；
发送通知。
但它们不能并行执行。
因为后一步依赖前一步的结果。
执行顺序仍然是：
添加水印完成；
上传才能开始；
上传完成；
保存数据库才能开始；
数据库保存完成；
通知才能开始。
所以：
异步描述的是任务不一定由当前线程执行；
并行描述的是多个任务是否能同时执行。
thenCompose() 连接的是异步任务，但它们之间仍然是串行依赖。
thenCompose 中的异常传播
如果第一个任务失败：
CompletableFuture<String> future =
addWatermarkAsync(file)
.thenCompose(path -> uploadAsync(path));
那么 uploadAsync() 不会执行。
因为前一步没有正常返回 path。
最终调用：
future.join();
会抛出异常。
如果第一步成功，但上传失败，最终结果同样会异常完成。
也就是说，异常会顺着整条链向后传播。
在链尾统一处理异常
可以在最后统一写：
CompletableFuture<PdfTaskResult> future =
addWatermarkAsync(file, pdfExecutor)
.thenCompose(targetPath -> {
return uploadAsync(
targetPath,
uploadExecutor
);
})
.thenApply(downloadUrl -> {
return PdfTaskResult.success(
file.getName(),
downloadUrl
);
})
.exceptionally(ex -> {
return PdfTaskResult.fail(
file.getName(),
getErrorMessage(ex)
);
});
这里的流程是：
水印成功后上传；
上传成功后组装成功结果；
任何一步异常，都在 exceptionally 中转成失败结果。
这种写法比较适合完整业务链。
完整的异常处理示例
CompletableFuture<PdfTaskResult> future =
addWatermarkAsync(file, pdfExecutor)
.thenCompose(targetPath -> {
System.out.println("水印文件生成完成，准备上传："
+ targetPath);
return uploadAsync(
targetPath,
uploadExecutor
);
})
.thenApply(downloadUrl -> {
System.out.println("文件上传完成："
+ downloadUrl);
return PdfTaskResult.success(
file.getName(),
downloadUrl
);
})
.whenComplete((result, ex) -> {
if (ex != null) {
System.out.println("整条流程执行异常："
+ getErrorMessage(ex));
} else {
System.out.println("整条流程执行完成");
}
})
.exceptionally(ex -> PdfTaskResult.fail(
file.getName(),
getErrorMessage(ex)
));
这里：
thenCompose 连接两个异步任务；
thenApply 把下载地址转换成结果对象；
whenComplete 记录整个流程的执行情况；
exceptionally 负责最终兜底。
thenCompose 的典型场景
我觉得它比较适合这些业务：
先查询用户，再根据用户查询订单；
先创建订单，再异步发起支付；
先生成文件，再上传文件；
先获取 token，再调用需要 token 的接口；
先保存任务，再根据任务 ID 启动后续处理；
先添加 PDF 水印，再上传到远程存储。
这些场景都有共同特点：
第二个异步任务，需要第一个异步任务的返回值。
这一节小结
这一节我主要记住几点：
1. thenCompose 用来连接两个前后依赖的异步任务；
2. 第二个任务需要使用第一个任务的结果；
3. thenCompose 的 Lambda 返回的是另一个 CompletableFuture；
4. thenApply 返回异步任务时，会产生嵌套的 CompletableFuture；
5. thenCompose 会自动展开嵌套，得到一个普通的 CompletableFuture<T>；
6. 返回普通值用 thenApply，返回 CompletableFuture 用 thenCompose；
7. 两个任务互不依赖并且最后合并，用 thenCombine；
8. thenCompose 连接的是异步任务，但任务之间仍然是串行依赖。
用一句话总结：
上一步完成后，还要根据它的结果启动另一个异步任务，就用 thenCompose。
下一节继续看 anyOf。
前面的 allOf 是等待所有任务完成，而 anyOf 是多个任务里只要有一个先完成，就可以先拿它的结果继续执行。
