













根据Oracle官方出具的Java文档说明,创建线程的方式只有两种:继承Thread或者实现Runnable接口。但是这两种方法都存在一个缺陷,没有返回值,也就是说我们无法得知线程执行结果。虽然简单场景下已经满足,但是当我们需要返回值的时候怎么办呢? Java 1.5 以后的Callable和Future接口就解决了这个问题,我们可以通过向线程池提交一个Callable来获取一个包含返回值的Future对象,从此,我们的程序逻辑就不再是同步顺序。
在Java多线程体系中,Future接口(位于java.util.concurrent包)是早期用于处理异步任务结果的重要工具,它提供了获取异步计算结果、取消任务等能力。但随着并发场景的复杂化,Future的局限性逐渐显现,主要体现在以下几个方面:
Future的get()方法是获取异步结果的核心方式,但它存在严重的阻塞问题:
举例:若要实现“任务A完成后自动执行任务B”,使用Future需要手动轮询isDone()判断任务A是否完成,再触发任务B,代码冗余且效率低。
在实际场景中,常需要组合多个异步任务的结果(如“等待所有任务完成”或“只要有一个任务完成”),但Future缺乏原生支持:
对比:Java 8引入的CompletableFuture提供了allOf()(等待所有完成)、anyOf()(任一完成)等方法,直接解决了这一问题。
Future的cancel(boolean mayInterruptIfRunning)方法用于取消任务,但存在明显局限:
异步任务执行过程中若抛出异常,Future的处理方式非常被动:
举例:若任务因网络异常失败,Future不会即时通知,直到调用get()才会抛出异常,可能导致错误处理延迟。
Future仅提供isDone()判断任务是否完成,没有任何API获取任务的执行进度(如“完成了50%”)。
对于长时间运行的任务(如下载大文件、复杂计算),无法实时反馈进度,用户体验差。
Future的接口设计简单,仅包含get()、cancel()、isDone()等基础方法,不支持任务的链式编排。
在复杂业务场景中(如“任务A→任务B→任务C”的依赖关系),需要手动嵌套Future,代码可读性和可维护性极差。
Future是Java早期异步编程的基础工具,但其设计过于简单,无法满足复杂并发场景的需求(如回调、组合、进度跟踪等)。正因为这些局限性,Java 8引入了CompletableFuture,它实现了Future接口并扩展了丰富的异步编程能力(回调、链式操作、组合任务等),成为现代Java异步编程的首选。
简单来说:Future适合简单的“提交任务→获取结果”场景,而复杂的异步流程(尤其是需要多任务协作、回调触发的场景)需使用CompletableFuture。
CompletableFuture是 Java 8 引入的一个异步编程工具类(位于java.util.concurrent包),它实现了Future和CompletionStage接口,旨在解决传统Future接口的局限性,提供更强大、更灵活的异步编程能力。
CompletableFuture不仅可以像Future一样表示一个异步计算的结果,还支持通过回调机制自动处理结果、组合多个异步任务、异常处理等复杂操作,无需手动轮询或阻塞等待,大幅简化了异步代码的编写。
CompletableFuture提供了一系列以then开头的方法(如thenApply、thenAccept、thenRun等),允许在异步任务完成后自动触发后续操作,无需主动调用get()阻塞等待。
这些方法支持链式编排,形成“任务流水线”,代码可读性更高。
示例:
// 异步计算一个数字的平方,然后加1,最后打印结果
CompletableFuture<Integer> future = CompletableFuture
.supplyAsync(() -> 3 * 3) // 异步任务:计算9
.thenApply(result -> result + 1) // 任务完成后自动执行:9+1=10
.thenAccept(result -> System.out.println("结果:" + result)); // 最终处理:打印10
CompletableFuture提供了原生方法组合多个异步任务,解决了Future无法高效组合任务的问题:
示例:
// 三个异步任务
CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> "任务1完成");
CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> "任务2完成");
CompletableFuture<String> task3 = CompletableFuture.supplyAsync(() -> "任务3完成");
// 等待所有任务完成后打印"全部完成"
CompletableFuture.allOf(task1, task2, task3)
.thenRun(() -> System.out.println("全部任务完成"));
CompletableFuture提供了专门的异常处理方法,无需等到调用get()时才捕获异常,可在任务执行过程中即时处理:
示例:
CompletableFuture<Integer> future = CompletableFuture
.supplyAsync(() -> {
if (true) {
throw new RuntimeException("计算失败"); // 模拟异常
}
return 100;
})
.exceptionally(ex -> { // 捕获异常,返回默认值
System.out.println("发生异常:" + ex.getMessage());
return 0; // 异常时返回0
})
.thenAccept(result -> System.out.println("最终结果:" + result)); // 输出:最终结果:0
CompletableFuture的异步方法(如supplyAsync、runAsync)默认使用ForkJoinPool.commonPool()作为线程池,无需手动创建线程或线程池;也可显式传入自定义线程池,灵活控制资源。
示例:
// 使用自定义线程池执行异步任务
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletableFuture<String> future = CompletableFuture.supplyAsync(
() -> {
return "使用自定义线程池";
},
executor
); // 传入自定义线程池
CompletableFuture允许通过complete(T value)或completeExceptionally(Throwable ex)方法手动设置结果或异常,无需等待异步任务执行完成。这在测试或需要外部触发的场景中非常有用。
示例:
CompletableFuture<String> future = new CompletableFuture<>();
// 另一个线程手动完成任务
new Thread(() -> {
try {
Thread.sleep(1000);
future.complete("手动设置的结果"); // 主动设置成功结果
} catch (InterruptedException e) {
future.completeExceptionally(e); // 主动设置异常
}
}).start();
System.out.println(future.get()); // 输出:手动设置的结果
CompletableFuture是 Java 异步编程的“瑞士军刀”,其核心特点是:非阻塞回调、链式操作、任务组合、灵活异常处理,完美解决了传统Future的阻塞、无法组合、异常处理简陋等问题。它广泛应用于需要高效处理异步任务的场景(如分布式服务调用、IO操作、并行计算等),是现代Java并发编程的核心工具之一。
虽然CompletableFuture 的异步方法(如 supplyAsync、runAsync)默认使用 ForkJoinPool.commonPool() 作为线程池,无需手动创建线程或线程池。
但是使用ForkJoinPool创建线程会产生许多问题,在实际项目中一般都会创建池,CompletableFuture执行时由线程池进行托管。
在使用CompletableFuture时,若依赖默认的ForkJoinPool.commonPool()(公共线程池)创建线程,可能会产生以下问题:
ForkJoinPool.commonPool()的设计更适合轻量、短时的并行计算任务,而非复杂的业务异步场景。在使用CompletableFuture时,建议根据业务需求自定义线程池(如ThreadPoolExecutor),通过控制线程数、设置非守护线程、自定义命名等方式,避免上述问题,提升异步任务的稳定性和可维护性。
在application.yml中添加如下配置:
spring:
application:
name: others
task:
execution:
thread-name-prefix: thread_pool_name_
pool:
core-size: 10
max-size: 20
queue-capacity: 10000
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.aop.interceptor.SimpleAsyncUncaughtExceptionHandler;
import org.springframework.boot.autoconfigure.task.TaskExecutionProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
/**
* 线程池配置类定义
* @author gary.wang
*
*/
@Configuration
@EnableAsync
public class AsyncConfiguration implements AsyncConfigurer {
private final Logger log = LoggerFactory.getLogger(AsyncConfiguration.class);
private final TaskExecutionProperties taskExecutionProperties;
public AsyncConfiguration(TaskExecutionProperties taskExecutionProperties) {
this.taskExecutionProperties = taskExecutionProperties;
}
@Override
@Bean(name = "taskExecutor")
public Executor getAsyncExecutor() {
log.info("创建异步执行线程池开始");
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(taskExecutionProperties.getPool().getCoreSize());
executor.setMaxPoolSize(taskExecutionProperties.getPool().getMaxSize());
executor.setQueueCapacity(taskExecutionProperties.getPool().getQueueCapacity());
executor.setThreadNamePrefix(taskExecutionProperties.getThreadNamePrefix());
log.info("创建异步执行线程池结束");
return executor;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return new SimpleAsyncUncaughtExceptionHandler();
}
}
实例化方法:
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
public static <U> CompletableFuture<U> supplyAsync(
Supplier<U> supplier,
Executor executor
);
public static CompletableFuture<Void> runAsync(Runnable runnable);
public static CompletableFuture<Void> runAsync(
Runnable runnable,
Executor executor
);
有两种格式:
示例:
@Test
void init() {
CompletableFuture<String> future = CompletableFuture.supplyAsync(
() -> {
log.info("supplyAsync 线程: {}", Thread.currentThread().getName());
return "hello world";
},
taskExecutor
);
log.info("future result: {}", future.join());
}
public T get()
public T get(long timeout, TimeUnit unit)
public T getNow(T valueIfAbsent)
public T join()
1:get():get()方法会阻塞线程直到任务完成。
如果使用下面的代码创建一个CompletableFuture,并执行get方法将会一直阻塞下去,不会看到log.info打印的输出。
/**
* get方法阻塞线程知道任务完成测试
* 主线程会一直阻塞,因为这种方式创建的CompletableFuture从未完成,因此不会执行到log.info输出任何内容
* @throws ExecutionException
* @throws InterruptedException
*/
@Test
void get() throws InterruptedException, ExecutionException {
CompletableFuture<Integer> future = new CompletableFuture<>();
Integer integer = future.get();
log.info("get 方法阻塞线程知道任务完成测试, 结果: {}", integer);//不会看到此句日志打印
}
2:get(long timeout, TimeUnit unit):同样会阻塞线程直到任务完成或者超过指定的超时时间。
/**
* get方法超时阻塞线程直到任务完成或者超过指定的超时时间 测试
*/
@Test
void getTimeout() {
CompletableFuture<Integer> future = new CompletableFuture<>();
try {
Integer integer = future.get(2, TimeUnit.SECONDS);
log.info("getTimeout 方法阻塞线程知道任务完成测试, 结果: {}", integer);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
log.error(
"getTimeout 方法阻塞线程直到任务完成测试, 异常: {}",
e.getMessage()
);
}
}
3:getNow(T valueIfAbsent):参数valueIfAbsent的意思是当计算结果不存在或者Now时刻没有完成任务,则返回一个默认值。
4:join():join与get的区别在于join返回计算结果或者一个unchecked异常(不需要明确的try-catch捕获),而get()返回一个具体的异常(InterruptedException, ExecutionException)。
特点:接收上一步的结果和异常,可内部处理异常做兜底操作,无返回结果。
public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)
/**
* 4. 计算完成后操作-complete
* whenCompleteAsync方法测试
*/
@Test
void whenCompleteAsync() {
CompletableFuture<String> future = CompletableFuture
.supplyAsync(
() -> {
log.info("supplyAsync 线程: {}", Thread.currentThread().getName());
return "hello world";
},
taskExecutor
)
.whenCompleteAsync(
(result, throwable) -> {
log.info(
"whenCompleteAsync 线程: {}",
Thread.currentThread().getName()
);
if (throwable != null) {
log.error(
"whenCompleteAsync 方法异常测试, 异常: {}",
throwable.getMessage()
);
} else {
log.info("whenCompleteAsync 方法测试, 入参: {}", result);
}
},
taskExecutor
);
log.info("whenCompleteAsync 方法测试, 结果: {}", future.join());
}
输出,可以看到supplyAsync和whenCompleteAsync方法对应的是不同的线程。
INFO 28576 --- [r_thread_pool_1] c.s.others.CompletableFutureComposeTest : supplyAsync 线程: other_thread_pool_1
INFO 28576 --- [r_thread_pool_2] c.s.others.CompletableFutureComposeTest : whenCompleteAsync 线程: other_thread_pool_2
INFO 28576 --- [r_thread_pool_2] c.s.others.CompletableFutureComposeTest : whenCompleteAsync 方法测试, 入参: hello world
INFO 28576 --- [ main] c.s.others.CompletableFutureComposeTest : whenCompleteAsync 方法测试, 结果: hello world
handle开头的方法和上面complete方法入参没有区别,都接收一个返回结果和一个可抛出异常,区别就在于handle相关的方法具有返回值。
特点:接收上一步的结果和异常,可内部处理异常做兜底操作,有返回结果。
public <U> CompletableFuture<U> handle(BiFunction<? super T,Throwable,? extends U> fn)
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T,Throwable,? extends U> fn)
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T,Throwable,? extends U> fn, Executor executor)
handle开头的方法和complete方法由于入参都接收一个可抛出的异常,因此可以在代码中捕获此异常做兜底操作。
handle使用示例:
/**
* 5. 计算完成后操作-handle
*/
@Test
void handle() {
CompletableFuture<String> future = CompletableFuture
.supplyAsync(
() -> {
log.info("supplyAsync 线程: {}", Thread.currentThread().getName());
return "hello world";
},
taskExecutor
)
.handleAsync(
(result, throwable) -> {
log.info("handleAsync 线程: {}", Thread.currentThread().getName());
if (throwable != null) {
log.error(
"handleAsync 方法异常测试, 异常: {}",
throwable.getMessage()
);
return "handleAsync 方法异常测试, 异常: " + throwable.getMessage();
} else {
log.info("handleAsync 方法测试, 入参: {}", result);
return result + "!";
}
},
taskExecutor
);
log.info("handleAsync 方法测试, 结果: {}", future.join());
}
特点:接收上一步的结果,执行时如果发生异常直接向上抛出,有返回结果。
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
特点:只消费,无返回
public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)
accept使用示例:
@Test
void accept() {
CompletableFuture<Void> future = CompletableFuture
.supplyAsync(
() -> {
log.info("supplyAsync 线程: {}", Thread.currentThread().getName());
return "hello world";
},
taskExecutor
)
.thenAcceptAsync(
result -> {
log.info("acceptAsync 线程: {}", Thread.currentThread().getName());
log.info("acceptAsync 方法测试, 入参: {}", result);
},
taskExecutor
);
log.info("acceptAsync 方法测试, 结果: {}", future.join());
}
打印结果:
\[r_thread_pool_1\] c.s.others.CompletableFutureComposeTest : supplyAsync 线程: other_thread_pool_1
\[r_thread_pool_2\] c.s.others.CompletableFutureComposeTest : acceptAsync 线程: other_thread_pool_2
\[r_thread_pool_2\] c.s.others.CompletableFutureComposeTest : acceptAsync 方法测试, 入参: hello world
\[ main\] c.s.others.CompletableFutureComposeTest : acceptAsync 方法测试, 结果: null
小贴士:向线程池提交任务的时候发生的异常属于外部异常,是无法进行捕捉的。exceptionally无法捕捉RejectedExecutionException()
exceptionally可以帮我们捕捉到所有中间过程的异常。方法会给我们一个异常作为参数,我们可以处理这个异常,同时返回一个默认值,相当于是返回兜底值,默认值的类型和上一个操作的返回值相同。
exceptionally使用示例:
@Test
void exceptionally() {
CompletableFuture<BigDecimal> future = CompletableFuture
.supplyAsync(
() -> {
log.info("supplyAsync 线程: {}", Thread.currentThread().getName());
return BigDecimal.ZERO;
},
taskExecutor
)
.thenApplyAsync(
result -> {
log.info("thenApplyAsync 线程: {}", Thread.currentThread().getName());
return BigDecimal.valueOf(1).divide(result, RoundingMode.HALF_UP);
},
taskExecutor
)
.thenApplyAsync(
result -> {
log.info("thenApplyAsync 线程: {}", Thread.currentThread().getName());
log.info("不会执行到这里,因为上一步就报错了。");
return result;
},
taskExecutor
)
.exceptionally(error -> {
log.error("exceptionally 方法异常测试, 异常: {}", error.getMessage());
return BigDecimal.ZERO;
});
log.info("exceptionally 方法测试, 结果: {}", future.join());
}
日志输出:
\[r_thread_pool_1\] c.s.others.CompletableFutureComposeTest : supplyAsync 线程: other_thread_pool_1
\[r_thread_pool_2\] c.s.others.CompletableFutureComposeTest : thenApplyAsync 线程: other_thread_pool_2
\[r_thread_pool_2\] c.s.others.CompletableFutureComposeTest : exceptionally 方法异常测试, 异常: java.lang.ArithmeticException: / by zero
\[ main\] c.s.others.CompletableFutureComposeTest : exceptionally 方法测试, 结果: 0
| 方法 | 触发时机 | 输入 | 输出 | 是否改变结果 | 典型用途 |
| complete(value) | 主动调用 | value(外部提供) | boolean(是否成功设置结果) | ✅(设置最终结果,若未完成) | 手动结束任务,常用于超时兜底、外部条件满足时提前返回 |
| thenApply(fn) | 正常完成时 | 前一步结果 | 转换后的新结果 | ✅(结果被替换) | 对结果进行转换/加工,如字符串转对象 |
| thenAccept(consumer) | 正常完成时 | 前一步结果 | void | ❌(不影响结果) | 消费结果(打印、保存数据库),不需要返回值 |
| handle((res, ex)) | 正常/异常完成时 | 结果或异常 | 新结果 | ✅(结果可被替换) | 统一收尾处理,无论成功/失败都能拿到,适合做异常修复 |
| exceptionally(ex -> …) | 异常完成时 | 异常对象 | 替代结果 | ✅(仅异常时替换结果) | 异常兜底,仅在失败时提供默认值或降级方案 |
假设有一个场景,需要查询一名学生每天的课程情况,需要首先需要根据学号查询学生详细信息,然后根据此详细信息查询学生所上的课程。
/**
* 8. thenApply使用示例
*/
@Test
void thenApply() {
String studentNo = "S001";
CompletableFuture<List<Map<String, Object>>> future = CompletableFuture
.supplyAsync(
() -> {
log.info("supplyAsync 线程: {}", Thread.currentThread().getName());
Map<String, Object> studentMap = new HashMap<>();
studentMap.put("studentNo", studentNo);
return studentMap;
},
taskExecutor
)
.thenApplyAsync(
studentMap -> {
log.info("thenApplyAsync 线程: {}", Thread.currentThread().getName());
//模拟根据学生信息查询学生相关的课程信息
List<Map<String, Object>> lessonMapList = new ArrayList<>();
Map<String, Object> lessonMap = new HashMap<>();
lessonMap.put("lessonNo", "L001");
lessonMap.put("lessonName", "Java 基础");
lessonMapList.add(lessonMap);
return lessonMapList;
},
taskExecutor
);
log.info("thenApply 方法测试, 结果: {}", future.join());
}
thenApply和thenCompose看起来很像,都是拿上一步的结果做点事,然后返回一个新的CompletableFuture。但他们的核心区别如下:
| 方法 | 输入 | 输出 | 返回类型 | 主要用途 | 类比 |
| thenApply(fn) | 上一步结果T | 转换结果U | CompletableFuture<U> | 普通转换:把结果加工成一个新值 | map(值映射) |
| thenCompose(fn) | 上一步结果T | 一个CompletableFuture<U> | 直接展开成CompletableFuture<U> | 任务扁平化:避免CompletableFuture<CompletableFuture<U>> | flatMap(展平映射) |
二者对比示例:
/**
* 9. thenApply和thenCompose的区别
*/
@Test
void thenApplyAndThenCompose() {
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "hello");
// thenApply:返回 CompletableFuture<CompletableFuture<Integer>>
CompletableFuture<CompletableFuture<Integer>> r1 = cf.thenApply(s ->
CompletableFuture.supplyAsync(s::length)
);
log.info("thenApply 方法测试, 结果: {}", r1.join().join());
// thenCompose:返回 CompletableFuture<Integer>
CompletableFuture<Integer> r2 = cf.thenCompose(s ->
CompletableFuture.supplyAsync(s::length)
);
log.info("thenCompose 方法测试, 结果: {}", r2.join());
}
输出结果:
c.s.others.CompletableFutureComposeTest : thenApply 方法测试, 结果: 5
c.s.others.CompletableFutureComposeTest : thenCompose 方法测试, 结果: 5
<U,V> CompletableFuture<V> thenCombine(
CompletionStage<? extends U> other,
BiFunction<? super T,? super U,? extends V> fn)
| 方法 | 任务个数 | 输入 | 输出 | 适用场景 |
| thenApply | 1 个 | 上一步结果T | 普通结果U | 单任务结果转换 |
| thenCompose | 1 个 | 上一步结果T | CompletableFuture<U>(被展开) | 链式依赖(结果触发下个异步任务) |
| thenCombine | 2 个 | this 结果T+ other 结果U | 普通结果V | 多任务并行后合并结果 |
thenCombine使用示例:
/**
* 10. thenCombine:组合两个CompletableFuture的结果
*/
@Test
void thenCombine() {
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "hello");
CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> 123);
CompletableFuture<String> r = cf1.thenCombine(
cf2,
(result1, result2) -> result1 + " -> " + result2
);
log.info("thenCombine 方法测试, 结果: {}", r.join());
}
输出结果:
thenCombine 方法测试, 结果: hello -> 123
allOf方法,当所有给定的任务完成后,返回一个全新的已完成CompletableFuture。
使用示例:
/**
* 11. allOf:等待所有CompletableFuture完成
*/
@Test
void allOf() {
List<CompletableFuture<Integer>> cfList = new ArrayList<>();
for (int i = 1; i <= 10; i++) {
int finalI = i;
CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(
() -> {
log.info("supplyAsync 线程: {}", Thread.currentThread().getName());
return finalI * finalI;
},
taskExecutor
);
cfList.add(cf);
}
CompletableFuture<Void> allFuture = CompletableFuture.allOf(
cfList.toArray(new CompletableFuture[0])
);
Integer result = 0;
log.info("allOf 方法测试, 等待全部线程完成,结果: {}", allFuture.join());
for (CompletableFuture<Integer> cf : cfList) {
result += cf.join();
}
log.info("allOf 方法测试, 结果: {}", result);
}
输出结果:
supplyAsync 线程: other_thread_pool_3
supplyAsync 线程: other_thread_pool_1
supplyAsync 线程: other_thread_pool_4
supplyAsync 线程: other_thread_pool_2
supplyAsync 线程: other_thread_pool_5
supplyAsync 线程: other_thread_pool_6
supplyAsync 线程: other_thread_pool_7
supplyAsync 线程: other_thread_pool_8
supplyAsync 线程: other_thread_pool_9
supplyAsync 线程: other_thread_pool_10
allOf 方法测试, 等待全部线程完成,结果: null
allOf 方法测试, 结果: 385
如果遇到需要通过多线程并行处理多个独立的耗时子任务,再汇总结果以提升整体执行效率的情景可以利用allOf函数。
当需要处理一批相互独立、逻辑相似但执行耗时的子任务(如批量数据计算、多源数据查询、并行接口调用等)时,为避免串行执行导致的总时间过长,可采用 “并行化处理 + 统一等待 + 结果汇总” 的方案:
仅等待Future集合中最快结束的任务完成,比如两个任务试图通过不同的方式计算同一个值,并返回结果。
那个任务执行过程中先遇到了异常也会先返回异常,因此可以使用exceptionally进行兜底。
anyOf使用示例:
/**
* 12. anyOf:等待任意一个CompletableFuture完成
*/
@Test
void anyOf() {
CompletableFuture<String> cf1 = CompletableFuture
.supplyAsync(
() -> {
long sleepTime = RandomUtil.randomLong(1000, 2000);
log.info(
"supplyAsync1 线程: {},本次睡眠时间为:{}",
Thread.currentThread().getName(),
sleepTime
);
try {
Thread.sleep(sleepTime); //随机睡眠一段时间
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return "hello world";
},
taskExecutor
)
.exceptionally(error -> "supplyAsync1 异常: " + error.getMessage());
CompletableFuture<String> cf2 = CompletableFuture
.supplyAsync(
() -> {
long sleepTime = RandomUtil.randomLong(1000, 2000);
log.info(
"supplyAsync2 线程: {},本次睡眠时间为:{}",
Thread.currentThread().getName(),
sleepTime
);
try {
Thread.sleep(sleepTime); //随机睡眠一段时间
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return "hello everyone";
},
taskExecutor
)
.exceptionally(error -> "supplyAsync2 异常: " + error.getMessage());
CompletableFuture<Object> r = CompletableFuture
.anyOf(cf1, cf2)
.exceptionally(error -> "anyOf 异常: " + error.getMessage());
log.info("anyOf 方法测试, 结果: {}", r.join());
}
输出结果:
supplyAsync1 线程: other_thread_pool_1,本次睡眠时间为:1575
supplyAsync2 线程: other_thread_pool_2,本次睡眠时间为:1792
anyOf 方法测试, 结果: hello world
@Test
void shopping() {
StopWatch stopWatch = new StopWatch();
String userId = "u001"; //用户ID
String goodId = "g001"; //商品ID
//可并行:查询用户信息
stopWatch.start("查询用户信息");
CompletableFuture<Map<String, Object>> userInfoFuture = CompletableFuture
.supplyAsync(
() -> {
log.info("查询用户信息 线程: {}", Thread.currentThread().getName());
Map<String, Object> userMap = new HashMap<>();
userMap.put("userId", userId);
userMap.put("userName", "张三");
return userMap;
},
taskExecutor
)
.exceptionally(error -> {
log.error("查询用户信息异常: {}", error.getMessage());
throw new RuntimeException("查询用户信息异常: " + error.getMessage());
});
stopWatch.stop();
//可并行:查询库存信息
stopWatch.start("查询库存信息");
CompletableFuture<Map<String, Object>> stockInfoFuture = CompletableFuture
.supplyAsync(
() -> {
log.info("查询库存信息 线程: {}", Thread.currentThread().getName());
Map<String, Object> stockMap = new HashMap<>();
stockMap.put("goodId", goodId);
stockMap.put("stock", 100);
return stockMap;
},
taskExecutor
)
.exceptionally(error -> {
log.error("查询库存信息异常: {}", error.getMessage());
throw new RuntimeException("查询库存信息异常: " + error.getMessage());
});
stopWatch.stop();
//可并行:查询商品信息
stopWatch.start("查询商品信息");
CompletableFuture<Map<String, Object>> goodInfoFuture = CompletableFuture
.supplyAsync(
() -> {
log.info("查询商品信息 线程: {}", Thread.currentThread().getName());
Map<String, Object> goodMap = new HashMap<>();
goodMap.put("goodId", goodId);
goodMap.put("goodName", "商品名称");
goodMap.put("price", 100); //商品价格
return goodMap;
},
taskExecutor
)
.exceptionally(error -> {
log.error("查询商品信息异常: {}", error.getMessage());
throw new RuntimeException("查询商品信息异常: " + error.getMessage());
});
stopWatch.stop();
//根据用户信息查询用户的优惠券信息
stopWatch.start("根据用户信息查询用户的优惠券信息");
CompletableFuture<Map<String, Object>> couponInfoFuture = userInfoFuture
.thenComposeAsync(
userMap -> {
log.info(
"根据用户信息查询用户的优惠券信息 线程: {}",
Thread.currentThread().getName()
);
Map<String, Object> couponMap = new HashMap<>();
couponMap.put("userId", userMap.get("userId"));
couponMap.put("coupon", "满100减10");
return CompletableFuture.completedFuture(couponMap);
},
taskExecutor
)
.exceptionally(error -> {
log.error("根据用户信息查询用户的优惠券信息异常: {}", error.getMessage());
throw new RuntimeException(
"根据用户信息查询用户的优惠券信息异常: " + error.getMessage()
);
});
stopWatch.stop();
//等待所有任务完成
stopWatch.start("等待所有任务完成");
CompletableFuture<Void> allFuture = CompletableFuture
.allOf(userInfoFuture, stockInfoFuture, goodInfoFuture, couponInfoFuture)
.thenAccept(v -> {
log.info("所有任务完成, 线程: {}", Thread.currentThread().getName());
});
allFuture.join(); //先调用join方法,等待所有任务完成
stopWatch.stop();
//获取所有任务的结果
Map<String, Object> userMap = userInfoFuture.join();
Map<String, Object> stockMap = stockInfoFuture.join();
Map<String, Object> goodMap = goodInfoFuture.join();
Map<String, Object> couponMap = couponInfoFuture.join();
log.info(
"所有任务完成, 用户信息: {}, 库存信息: {}, 商品信息: {}, 优惠券信息: {}",
userMap,
stockMap,
goodMap,
couponMap
);
log.info("购物流程耗时: {}ms", stopWatch.prettyPrint(TimeUnit.MICROSECONDS)); //微秒
}
在<获取所有完成结果-allOf>章节中已经给出过循环创建任务的示例了。这里再给出一个示例:
/**
* 14. forTest 测试
*/
@Test
void forTest() {
StopWatch stopWatch = new StopWatch();
stopWatch.start("forTest");
// 循环创建10个CompletableFuture
List<CompletableFuture<Integer>> collect = IntStream
.range(1, 10)
.mapToObj(i ->
CompletableFuture
.supplyAsync(
() -> {
// 在i=5的时候抛出一个NPE
if (i == 5) {
throw new NullPointerException();
}
log.info(
"CompletableFuture 线程睡眠开始: {}, 结果: {}",
Thread.currentThread().getName(),
i
);
try {
// 每个依次睡眠1-9s,模拟线程耗时
TimeUnit.SECONDS.sleep(i);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.info(
"CompletableFuture 线程睡眠结束: {}, 结果: {}",
Thread.currentThread().getName(),
i
);
return i;
},
taskExecutor
)
.exceptionally(error -> {
// 这里处理一下i=5时出现的NPE
// 如果这里不处理异常,那么异常会在所有任务完成后抛出,小伙伴可自行测试
log.error("exceptionally 异常: {}", error.getMessage());
log.info(
"exceptionally 线程睡眠开始: {}, 结果: {}",
Thread.currentThread().getName(),
i
);
try {
TimeUnit.SECONDS.sleep(5);
System.out.println(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.info(
"exceptionally 线程睡眠结束: {}, 结果: {}",
Thread.currentThread().getName(),
i
);
return 100;
})
)
.collect(Collectors.toList());
// List列表转成CompletableFuture的Array数组,使其可以作为allOf()的参数
// 使用join()方法使得主线程阻塞,并等待所有并行线程完成
CompletableFuture.allOf(collect.toArray(new CompletableFuture[] {})).join();
stopWatch.stop();
log.info("forTest 耗时: {}ms", stopWatch.prettyPrint(TimeUnit.SECONDS)); //秒
}
输出结果:
2025-09-25 15:47:50.200 INFO 38440 --- \[r_thread_pool_1\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠开始: other_thread_pool_1, 结果: 1
2025-09-25 15:47:50.202 INFO 38440 --- \[r_thread_pool_6\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠开始: other_thread_pool_6, 结果: 6
2025-09-25 15:47:50.202 INFO 38440 --- \[r_thread_pool_3\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠开始: other_thread_pool_3, 结果: 3
2025-09-25 15:47:50.202 INFO 38440 --- \[r_thread_pool_2\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠开始: other_thread_pool_2, 结果: 2
2025-09-25 15:47:50.202 INFO 38440 --- \[r_thread_pool_7\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠开始: other_thread_pool_7, 结果: 7
2025-09-25 15:47:50.202 INFO 38440 --- \[r_thread_pool_4\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠开始: other_thread_pool_4, 结果: 4
2025-09-25 15:47:50.202 INFO 38440 --- \[r_thread_pool_9\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠开始: other_thread_pool_9, 结果: 9
2025-09-25 15:47:50.202 INFO 38440 --- \[r_thread_pool_8\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠开始: other_thread_pool_8, 结果: 8
2025-09-25 15:47:50.203 ERROR 38440 --- \[r_thread_pool_5\] c.s.others.CompletableFutureComposeTest : exceptionally 异常: java.lang.NullPointerException
2025-09-25 15:47:50.203 INFO 38440 --- \[r_thread_pool_5\] c.s.others.CompletableFutureComposeTest : exceptionally 线程睡眠开始: other_thread_pool_5, 结果: 5
2025-09-25 15:47:51.201 INFO 38440 --- \[r_thread_pool_1\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠结束: other_thread_pool_1, 结果: 1
2025-09-25 15:47:52.209 INFO 38440 --- \[r_thread_pool_2\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠结束: other_thread_pool_2, 结果: 2
2025-09-25 15:47:53.205 INFO 38440 --- \[r_thread_pool_3\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠结束: other_thread_pool_3, 结果: 3
2025-09-25 15:47:54.212 INFO 38440 --- \[r_thread_pool_4\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠结束: other_thread_pool_4, 结果: 4
100
2025-09-25 15:47:55.207 INFO 38440 --- \[r_thread_pool_5\] c.s.others.CompletableFutureComposeTest : exceptionally 线程睡眠结束: other_thread_pool_5, 结果: 5
2025-09-25 15:47:56.215 INFO 38440 --- \[r_thread_pool_6\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠结束: other_thread_pool_6, 结果: 6
2025-09-25 15:47:57.204 INFO 38440 --- \[r_thread_pool_7\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠结束: other_thread_pool_7, 结果: 7
2025-09-25 15:47:58.210 INFO 38440 --- \[r_thread_pool_8\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠结束: other_thread_pool_8, 结果: 8
2025-09-25 15:47:59.213 INFO 38440 --- \[r_thread_pool_9\] c.s.others.CompletableFutureComposeTest : CompletableFuture 线程睡眠结束: other_thread_pool_9, 结果: 9
2025-09-25 15:47:59.224 INFO 38440 --- \[ main\] c.s.others.CompletableFutureComposeTest : forTest 耗时: StopWatch '': running time = 9 s
\---------------------------------------------
s % Task name
\---------------------------------------------
000000009 100% forTest
ms
CompletableFuture user = CompletableFuture.supplyAsync(() -> getUser());
CompletableFuture stock = CompletableFuture.supplyAsync(() -> getStock());
CompletableFuture product = CompletableFuture.supplyAsync(() -> getProduct());
CompletableFuture all = CompletableFuture.allOf(user, stock, product);
all.join(); // 等所有完成
CompletableFuture userFuture = CompletableFuture.supplyAsync(() -> getUser());
CompletableFuture couponFuture = userFuture.thenCompose(user ->
CompletableFuture.supplyAsync(() -> getCoupon(user))
);
CompletableFuture productFuture = CompletableFuture.supplyAsync(() -> getProduct());
CompletableFuture dtoFuture = productFuture.thenApply(this::convertToDTO);
productFuture.thenAccept(product -> log.info("商品: {}", product));
CompletableFuture priceFuture = productFuture.thenCombine(couponFuture,
(product, coupon) -> product.getPrice() - coupon.getDiscount()
);
CompletableFuture fast = serviceA.applyToEither(serviceB, res -> res);
CompletableFuture safeFuture = remoteCall.exceptionally(ex -> {
log.error("远程调用失败: {}", ex.getMessage());
return "默认值";
});
CompletableFuture future = new CompletableFuture<>();
scheduler.schedule(() -> future.complete("超时返回"), 2, TimeUnit.SECONDS);
future.whenComplete((res, ex) -> {
long cost = System.currentTimeMillis() - start;
log.info("耗时: {} ms", cost);
});
此内容由惯性聚合(RSS阅读器)自动聚合整理,仅供阅读参考。 原文来自 — 版权归原作者所有。