【CompletableFuture】常用方法(三)

1. 线程池的使用

默认情况下通过静态方法 runAsync 、supplyAsync 创建的 CompletableFuture 任务会使用公
共的 ForkJoinPool 线程池，其默认的线程数是 CPU 的核数。当然，其线程数可以通过以下 JVM
参数去设置：
option:-Djava.util.concurrent.ForkJoinPool.common.parallelism

问题是：如果所有 CompletableFuture 共享一个线程池，那么一旦有任务执行一些很慢的 IO
操作，就会导致线程池中所有线程都阻塞在 IO 操作上，从而造成线程饥饿，进而影响整个系统
的性能。所以，强烈建议大家根据不同的业务类型创建不同的线程池，以避免互相干扰。

所以，建议大家在生产环境使用时，根据不同的业务类型创建不同的线程池，以避免互相影
响。前面第一章为大家介绍了三种线程池：IO 密集型任务线程池、CPU 密集型任务线程池、混合
型任务线程池。大家可以根据不同的任务类型，确定线程池的类型和线程数。

作为演示，这里使用“混合型任务线程池”执行CompletableFuture任务，具体的代码如下：

2. 异步任务的串行执行

如果两个异步任务需要串行（当一个任务依赖另一个任务）执行，可以通过 CompletionStage
接口的 thenApply、thenAccept、thenRun 和 thenCompose 四个方法实现。

2.1 thenApply 方法

thenApply 方法有三个重载版本，三个版本的声明如下：

// 后一个任务与前一个任务在同一个线程中执行
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)// 后一个任务与前一个任务可以不在同一个线程中执行
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)// 后一个任务在指定的 executor 线程池中执行
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

thenApply 三个重载版本有一个共同的参数 fn，该参数表示待串行执行的第二个异步任务，
其类型为 Function。fn 的类型声明涉及到两个范型参数，具体如下：

• 范型参数 T：上一个任务所返回结果的类型。
• 范型参数 U：当前任务的返回值类型。

作为示例，使用 thenApply 分两步计算（10+10）*2，代码如下：

Function<T, R>接口既能接收参数也支持返回值，所以 thenApply 可以将前一个任务的结果，
通过 Function 的 R apply(T t) 方法传递给到第二个任务，并且能输出第二个任务的执行结果。

2.1.1 thenApply(Function<? super T, ? extends U> fn)

再举一个🌰

同步执行：后一个任务和前一个任务 在同一个线程中执行（如果前一个任务是异步的，使用的是那个线程）。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {System.out.println("任务1线程: " + Thread.currentThread().getName());return "hello";
}).thenApply(result -> {System.out.println("任务2线程: " + Thread.currentThread().getName());return result.toUpperCase();
});
future.join();

输出的结果：

任务1线程: ForkJoinPool.commonPool-worker-1
任务2线程: ForkJoinPool.commonPool-worker-1

✅ 任务2复用了任务1的线程。

2.1.2 thenApplyAsync(Function<? super T, ? extends U> fn)

异步执行（默认线程池）：后一个任务在公共线程池中异步执行，不一定和前一个任务在同一个线程。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {System.out.println("任务1线程: " + Thread.currentThread().getName());return "hello";
}).thenApplyAsync(result -> {System.out.println("任务2线程: " + Thread.currentThread().getName());return result.toUpperCase();
});
future.join();

任务1线程: ForkJoinPool.commonPool-worker-1
任务2线程: ForkJoinPool.commonPool-worker-3

✅ 任务2可能使用不同的线程（ForkJoinPool）执行。

2.1.3 thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor)

异步执行（指定线程池）：后一个任务在你提供的 Executor 线程池中执行，更灵活控制线程来源。

ExecutorService executor = Executors.newFixedThreadPool(2);CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {System.out.println("任务1线程: " + Thread.currentThread().getName());return "hello";
}).thenApplyAsync(result -> {System.out.println("任务2线程: " + Thread.currentThread().getName());return result.toUpperCase();
}, executor);future.join();
executor.shutdown();

输出可能：

任务1线程: ForkJoinPool.commonPool-worker-1
任务2线程: pool-1-thread-1

✅ 任务2明确使用了你自己创建的线程池中的线程。

2.1.4

方法名	是否异步	使用线程	是否可指定线程池
thenApply	否	同前任务线程	否
thenApplyAsync(fn)	是	默认线程池（ForkJoinPool）	否
thenApplyAsync(fn, executor)	是	指定线程池	✅ 是

2.2 thenRun 方法