34. 带 Async 和不带 Async 的区别
发布于 • 阅读量 0
34. 带 Async 和不带 Async 的区别
前面用过这些方法：
thenApply(...)
thenAccept(...)
thenRun(...)
实际上，它们都有对应的 Async 版本：
thenApplyAsync(...)
thenAcceptAsync(...)
thenRunAsync(...)
名字只多了一个 Async，但执行线程可能完全不同。
我现在理解这组方法时，主要抓住一句话：
不带 Async：通常由完成上一步任务的线程继续执行。
带 Async：把后续任务重新提交到线程池执行。
这里我用了“通常”，而不是“绝对”。
因为不带 Async 的回调到底在哪个线程执行，还和注册回调时上一步是否已经完成有关。
这一节就把这个问题说清楚。
先看不带 Async 的 thenApply
新建类：
com.succos.completablefuture.ThenApplyThreadDemo
代码如下：
package com.succos.completablefuture;
import java.util.concurrent.CompletableFuture;
public class ThenApplyThreadDemo {
public static void main(String[] args) {
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> {
System.out.println("supplyAsync 执行线程："
+ Thread.currentThread().getName());
sleep(2000);
return "output/test-watermark.pdf";
})
.thenApply(path -> {
System.out.println("thenApply 执行线程："
+ Thread.currentThread().getName());
return "http://localhost/download?file=" + path;
});
String result = future.join();
System.out.println("最终结果：" + result);
}
private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
运行后，可能看到：
supplyAsync 执行线程：ForkJoinPool.commonPool-worker-1
thenApply 执行线程：ForkJoinPool.commonPool-worker-1
也就是说，前一步任务执行完成以后，当前工作线程直接继续执行了 thenApply。
没有重新把 thenApply 提交给另一个线程池。
这个过程可以理解成：
同一个工人处理完上一步以后，顺手继续做下一步。
再看 thenApplyAsync
把上面的 thenApply 改成：
thenApplyAsync(...)
完整代码如下：
package com.succos.completablefuture;
import java.util.concurrent.CompletableFuture;
public class ThenApplyAsyncThreadDemo {
public static void main(String[] args) {
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> {
System.out.println("supplyAsync 执行线程："
+ Thread.currentThread().getName());
sleep(2000);
return "output/test-watermark.pdf";
})
.thenApplyAsync(path -> {
System.out.println("thenApplyAsync 执行线程："
+ Thread.currentThread().getName());
return "http://localhost/download?file=" + path;
});
String result = future.join();
System.out.println("最终结果：" + result);
}
private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
运行后可能看到：
supplyAsync 执行线程：ForkJoinPool.commonPool-worker-1
thenApplyAsync 执行线程：ForkJoinPool.commonPool-worker-2
也可能还是同一个线程名。
因为公共线程池最终怎么调度，并不是固定的。
但关键区别在于：
thenApply：上一步完成后，可以直接在当前线程继续执行。
thenApplyAsync：后续步骤会被当成新的异步任务，再交给执行器调度。
所以不能只根据一次日志里线程名是否相同，判断 Async 有没有生效。
它们的调度方式已经不同了。
不带 Async 不代表在 main 线程执行
这个地方我以前容易理解错。
thenApply() 不带 Async，并不代表它一定在 main 线程执行。
它通常会由完成上一步任务的线程继续执行。
比如：
CompletableFuture
.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName());
return "a.pdf";
})
.thenApply(name -> {
System.out.println(Thread.currentThread().getName());
return name.toUpperCase();
});
如果 supplyAsync 跑在：
ForkJoinPool.commonPool-worker-1
那 thenApply 很可能也由这个线程执行。
所以正确理解是：
不带 Async，不是指定 main 线程；
而是不主动切换到新的异步执行器。
上一步已经完成时，情况又不一样
再看一个例子：
CompletableFuture<String> future =
CompletableFuture.completedFuture("a.pdf");
CompletableFuture<String> resultFuture = future.thenApply(name -> {
System.out.println("执行线程：" + Thread.currentThread().getName());
return name.toUpperCase();
});
System.out.println(resultFuture.join());
completedFuture() 创建出来时，任务已经完成了。
这时候调用 thenApply()，后续逻辑可能直接由当前调用线程执行。
如果这段代码在 main 方法里，可能打印：
执行线程：main
所以不带 Async 的方法，可能由两类线程执行：
如果上一步还没完成：通常由完成上一步的线程执行；
如果上一步已经完成：可能由当前注册回调的线程直接执行。
这也是为什么不能简单记成：
thenApply 一定和上一步是同一个线程。
它不是绝对的。
带 Async 默认使用哪个线程池
如果写：
thenApplyAsync(path -> {
return buildDownloadUrl(path);
});
又没有传线程池，那么默认会使用公共线程池。
也就是：
ForkJoinPool.commonPool
这和 supplyAsync() 不传线程池时类似。
所以这种写法：
CompletableFuture
.supplyAsync(() -> processPdf(file), pdfExecutor)
.thenApplyAsync(path -> buildDownloadUrl(path));
第一步使用的是：
pdfExecutor
但第二步没有指定线程池，就可能跑到：
ForkJoinPool.commonPool
里。
这点很容易忽略。
带 Async 时可以明确指定线程池
更清楚的写法是：
CompletableFuture
.supplyAsync(() -> processPdf(file), pdfExecutor)
.thenApplyAsync(path -> buildDownloadUrl(path), urlExecutor);
这里明确表示：
PDF 处理使用 pdfExecutor；
下载地址生成使用 urlExecutor。
如果后面的任务也想继续使用 PDF 线程池，也可以写：
CompletableFuture
.supplyAsync(() -> processPdf(file), pdfExecutor)
.thenApplyAsync(path -> buildDownloadUrl(path), pdfExecutor);
不过是否应该共用同一个线程池，要看任务性质。
如果 buildDownloadUrl() 只是拼接字符串，就没必要专门异步切换。
直接用 thenApply() 更简单。
简单转换通常不用 Async
例如：
.thenApply(path -> {
return "http://localhost/download?file=" + path;
})
这里只是拼接一个字符串。
耗时非常短。
如果为了这一点小事，再写成：
.thenApplyAsync(path -> {
return "http://localhost/download?file=" + path;
}, anotherExecutor)
反而会增加一次任务提交和线程调度。
这不一定更快。
所以我现在的判断是：
后续逻辑很轻：优先不带 Async；
后续逻辑耗时明显，或者需要线程池隔离：考虑带 Async。
并不是看到异步流程，就每一步都加 Async。
重任务可以考虑 Async
假设 PDF 水印完成以后，还要做一个比较重的操作：
把 PDF 上传到远程对象存储；
调用外部接口生成下载凭证；
进行病毒扫描；
做二次文件压缩。
这时候可以考虑：
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> {
return addWatermark(file);
}, pdfExecutor)
.thenApplyAsync(path -> {
return uploadToStorage(path);
}, uploadExecutor);
这里：
addWatermark 使用 PDF 线程池；
uploadToStorage 使用上传线程池。
这样可以隔离不同类型的任务。
PDF 任务忙，不一定把上传任务也一起堵住。
为什么线程池隔离很重要
假设我把所有任务都放进一个线程池：
PDF 处理；
远程上传；
数据库保存；
消息通知。
如果 PDF 处理很慢，占满了所有工作线程，其他任务也只能等。
这会造成不同业务互相影响。
所以真实项目里，有时会按任务类型拆线程池：
pdf-worker-
upload-worker-
notify-worker-
然后通过带 Async 的方法指定不同线程池。
比如：
CompletableFuture
.supplyAsync(() -> addWatermark(file), pdfExecutor)
.thenApplyAsync(path -> upload(path), uploadExecutor)
.thenAcceptAsync(url -> notifyUser(url), notifyExecutor);
这个流程就很清楚：
PDF 处理在线程池 A；
上传在线程池 B；
通知在线程池 C。
当然，也不能为了隔离而创建几十个线程池。
线程池数量太多，同样会带来资源和管理成本。
还是要看业务规模。
thenAccept 和 thenAcceptAsync 也是一样
例如：
future.thenAccept(path -> {
saveDatabase(path);
});
不带 Async 时，通常由完成上一步的线程继续执行。
而：
future.thenAcceptAsync(path -> {
saveDatabase(path);
}, databaseExecutor);
表示把保存数据库这一步重新提交给指定线程池。
不过数据库操作本身已经由数据库连接池管理，不代表一定要再单独搞一个异步线程池。
是否拆开，要看当前调用链和业务需求。
不要为了使用 CompletableFuture，把每一行代码都异步化。
thenRun 和 thenRunAsync 也是一样
不带 Async：
future.thenRun(() -> {
System.out.println("流程结束");
});
通常由完成上一步的线程执行。
带 Async：
future.thenRunAsync(() -> {
System.out.println("流程结束");
}, executor);
会重新提交到指定线程池。
如果只是打印一句日志，通常没有必要切线程。
所以简单场景更适合：
thenRun(...)
一个完整的线程观察示例
新建类：
com.succos.completablefuture.AsyncDifferenceDemo
代码如下：
package com.succos.completablefuture;
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 AsyncDifferenceDemo {
public static void main(String[] args) {
ThreadPoolExecutor pdfExecutor = new ThreadPoolExecutor(
2,
2,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory("pdf-worker"),
new ThreadPoolExecutor.CallerRunsPolicy()
);
ThreadPoolExecutor uploadExecutor = new ThreadPoolExecutor(
2,
2,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory("upload-worker"),
new ThreadPoolExecutor.CallerRunsPolicy()
);
CompletableFuture<Void> future = CompletableFuture
.supplyAsync(() -> {
printThread("添加 PDF 水印");
sleep(2000);
return "output/test-watermark.pdf";
}, pdfExecutor)
.thenApply(path -> {
printThread("生成本地下载地址");
return "http://localhost/download?file=" + path;
})
.thenApplyAsync(url -> {
printThread("上传远程存储");
sleep(2000);
return "https://file.example.com/test-watermark.pdf";
}, uploadExecutor)
.thenAccept(url -> {
printThread("打印最终地址");
System.out.println("最终地址：" + url);
});
future.join();
pdfExecutor.shutdown();
uploadExecutor.shutdown();
}
private static void printThread(String step) {
System.out.println(step
+ "，执行线程："
+ Thread.currentThread().getName());
}
private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
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 水印，执行线程：pdf-worker-1
生成本地下载地址，执行线程：pdf-worker-1
上传远程存储，执行线程：upload-worker-1
打印最终地址，执行线程：upload-worker-1
这里可以看到：
thenApply 没有主动切线程，所以继续由 pdf-worker-1 执行；
thenApplyAsync 指定了 uploadExecutor，所以切到 upload-worker-1；
后面的 thenAccept 不带 Async，所以继续由 upload-worker-1 执行。
这个例子基本把执行规律串起来了。
不要把 Async 理解成一定并行
还有一个容易误解的地方。
写了：
thenApplyAsync(...)
不代表它会和上一步并行执行。
因为它仍然依赖上一步的结果。
比如：
supplyAsync(() -> processPdf())
.thenApplyAsync(path -> upload(path));
upload(path) 必须等 processPdf() 返回路径以后才能执行。
所以这两个步骤还是前后依赖关系。
Async 只是表示：
后续步骤由执行器异步调度；
不是直接在完成上一步的线程里继续执行。
它没有改变任务依赖关系。
每一步都加 Async 可能更慢
比如：
CompletableFuture
.supplyAsync(() -> step1(), executor)
.thenApplyAsync(result -> step2(result), executor)
.thenApplyAsync(result -> step3(result), executor)
.thenAcceptAsync(result -> step4(result), executor);
如果每一步都非常轻，而且还使用同一个线程池，那么每一步都重新入队、重新调度，可能只是增加开销。
这种情况下，不带 Async 反而更自然：
CompletableFuture
.supplyAsync(() -> step1(), executor)
.thenApply(result -> step2(result))
.thenApply(result -> step3(result))
.thenAccept(result -> step4(result));
让同一个工作线程顺着流程往下执行，减少不必要的切换。
所以 Async 不是越多越好。
我现在怎么选
我会先看后续步骤的性质。
如果只是：
格式转换；
字符串拼接；
组装结果对象；
简单日志；
轻量状态判断。
通常不带 Async。
比如：
thenApply(path -> buildDownloadUrl(path))
如果后续步骤：
耗时明显；
可能阻塞；
需要使用另一个专用线程池；
不希望占用当前线程池。
可以考虑带 Async。
比如：
thenApplyAsync(path -> uploadToStorage(path), uploadExecutor)
所以我的判断不是：
异步代码就全部加 Async。
而是：
是否需要把这一步重新交给线程池调度？
这一节小结
这一节我主要记住几点：
1. 不带 Async 的后续任务，通常由完成上一步的线程继续执行；
2. 如果上一步已经完成，不带 Async 的回调也可能由当前调用线程执行；
3. 带 Async 的方法会把后续步骤重新提交给执行器调度；
4. 带 Async 但不指定线程池时，通常使用公共线程池；
5. 要控制执行位置，最好显式传入自定义线程池；
6. 简单转换没必要加 Async，否则只是增加调度开销；
7. 耗时任务、阻塞任务或需要线程池隔离时，可以使用 Async；
8. Async 只改变调度方式，不会改变任务之间的依赖关系。
用一句话总结：
不带 Async 是“当前线程顺手继续做”，带 Async 是“把下一步重新交给线程池安排”。
下一节继续看 exceptionally、handle 和 whenComplete。
这三个方法都和异常有关，但用途不一样：一个适合失败兜底，一个可以同时转换成功和失败结果，一个更适合记录日志和观察任务完成情况。
