Java8 CompletableFuture使用及說明
更新時間：2026年04月02日 17:07:26 作者：Jerry的技術博客
文章介紹了Future和CompletableFuture接口及其方法,解釋了它們的異同、使用場景與應用場景,示例了Future和CompletableFuture的使用方法和場景,強調了CompletableFuture的函數(shù)式編程特性
一、Future接口
1.1 Runnable與Callable
Runnable接口源自JDK1.1，它只有一個run()方法，該方法沒有返回結果：
public interface Runnable {
public abstract void run();
}
Callable接口是JDK1.5中添加，只有一個call()方法，該方法支持結果返回且可以拋出異常：
public interface Callable<V> {
V call() throws Exception;
}
無論是Runnalble還是Callable任務實例，直接調用時不會新開啟子線程，是在主線程中運行。
如果要在子線程中運行這些任務，需要將任務提交到Thread實例或線程池執(zhí)行。
值得注意的是通過Thread實例執(zhí)行Callable任務，Thread沒有接受Callable入?yún)⒌臉嬙旌瘮?shù)，因此，不能直接構造Callable任務的Thread實例。
可以通過構建FutureTask任務將Callable任務提供到Thread實例執(zhí)行，因為FutureTask實現(xiàn)了Runnable接口，且其構造函數(shù)入?yún)⒅С諧allable類型。
@Test
public void testRunable() throws Exception {
Runnable r = new Runnable() {
@Override
public void run() {
try {
Thread.sleep(1000L);
System.out.println(Thread.currentThread().getName());
} catch (InterruptedException e) {
}
}
};
Callable<Integer> c = new Callable<Integer>() {
@Override
public Integer call() throws Exception {
try {
Thread.sleep(1000L);
System.out.println(Thread.currentThread().getName());
} catch (InterruptedException e) {
}
return ThreadLocalRandom.current().nextInt(100);
}
};
r.run();
System.out.println(c.call());
Thread t = new Thread(r);
t.start();
t.join(); // 不阻塞時，如果主線程比子線程先完成，則子線程中任務不會完整執(zhí)行
FutureTask<Integer> f = new FutureTask<>(c);
Thread t2 = new Thread(f);
t2.start();
t2.join(); // 不阻塞時，如果主線程比子線程先完成，則子線程中任務不會完整執(zhí)行
System.out.println(f.get());
ExecutorService executorService = Executors.newFixedThreadPool(1);
executorService.submit(r);
Future<Integer> future = executorService.submit(c);
System.out.println(future.get()); // 不阻塞時，如果主線程比子線程先完成，則子線程中任務不會完整執(zhí)行
}
---
main
main
62
Thread-0
Thread-1
6
pool-1-thread-1
pool-1-thread-1
36
注意：
- 通過junit對以上代碼進行測試時，如果主線程比子線程先執(zhí)行完成，則主線程執(zhí)行完成后會程序會退出，意味著沒有完成的子線程任務不會再繼續(xù)執(zhí)行。
- 因此，如果需要子線程完整執(zhí)行，則需要阻塞主線程。
1.2 Future接口
Future接口也是在JDK1.5中添加的，用于描述一個異步Callable任務的計算結果，如下所示。
Future接口一共有5個方法：
- 通過cancel()方法停止任務；
- 通過isCanceld()方法判斷任務是否被停止；
- 通過isDone()方法檢查任務是否完成；
- 提供了兩個重載版本的get方法，用于阻塞調用線程，直到任務完成或超時。
public interface Future<V> {
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
Future只能通過阻塞或輪訓的方式獲取任務結果，阻塞違背異步變成的初衷，輪訓的方式又耗費了CPU資源，通過Future獲取異步任務結果的示里如下：
@Test
public void testFuture() throws InterruptedException, ExecutionException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
List<Future<Integer>> futures = new ArrayList<>();
List<Integer> results = new ArrayList<>();
for(int i=0; i <= 3; i++) {
final int j = i;
Future<Integer> future = executorService.submit(() -> {
try {
Thread.sleep(j * 100L);
} catch (InterruptedException e) {
}
return ThreadLocalRandom.current().nextInt(100);
});
futures.add(future);
}
for(Future<Integer> future : futures) {
while(true) {
if(future.isDone() && !future.isCancelled()) {
results.add(future.get());
break;
}
}
}
System.out.println(results);
}
二、CompletableFuture
JDK8中添加了ComletableFuture類，它實現(xiàn)了Future接口和CompletionStage接口，其中CompletionStage可以看做是一個異步任務執(zhí)行過程的抽象，如下。
它簡化了異步變成的復雜性，并提供了函數(shù)式變成的能力。
ComletableFuture提供了同步和異步的執(zhí)行方式，其異步方式又支持外部聲明的線程池，如果不提供外部線程池，則異步時默認使用ForkJoinPool.commonPool()。
public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
...
}
CompletableFuture實現(xiàn)了Future接口，因此，也可以通過以前的方式阻塞或輪訓獲取結果：
// 同F(xiàn)uture
public T get() throws InterruptedException, ExecutionException {
...
}
// 同F(xiàn)uture
public T get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException {
...
}
// 返回執(zhí)行結果或拋出一個unchecked異常
public T join() {
...
}
// 如果執(zhí)行結束則返回執(zhí)行結果，否則返回valueIfAbsent
public T getNow(T valueIfAbsent) {
...
}
CompetableFuture.completedFuture是一個靜態(tài)輔助類，用于返回一個計算好的CompletableFuture，示例：
@Test
public void testComletedFuture() throws InterruptedException, ExecutionException {
CompletableFuture<String> future = CompletableFuture.completedFuture("result");
System.out.println(future.get());
}
---
result
通過以下四個靜態(tài)方法為異步執(zhí)行代碼創(chuàng)建CompletableFuture對象：
// 接受Runnable任務，返回結果為空
public static CompletableFuture<Void> runAsync(Runnable runnable) {
...
}
// 接受Runnable任務，返回結果為空，指定線程池
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) {
...
}
// 接受Supplier對象，可指定返回類型
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
...
}
// 接受Supplier對象，可指定返回類型，指定線程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) {
...
}
CompletableFuture任務執(zhí)行示例如下，可指定執(zhí)行線程池，如不指定則使用ForkJoinPool.commonPool()。
@Test
public void testCreateSyncTask() throws InterruptedException, ExecutionException {
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName());
return "supplyAsync result";
});
ExecutorService executorService = Executors.newFixedThreadPool(1);
CompletableFuture<Void> f2 = CompletableFuture.runAsync(() -> {
System.out.println(Thread.currentThread().getName());
}, executorService);
System.out.println(f1.get());
}
---
ForkJoinPool.commonPool-worker-1
supplyAsync result
pool-1-thread-1
Future和CompletableFuture提供的get()方法是阻塞的，為了獲取任務結果同時不阻塞當前線程，可使用CompletionStage提供的方法實現(xiàn)任務異步處理，有以下4中處理方式：
// 上游任務完成后在當前主線程中同步執(zhí)行處理，向下傳遞上游處理結果或異常
public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {
...
}
// 上游任務完成后在當前子線程中異步執(zhí)行處理，向下傳遞上游處理結果或異常
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) {
...
}
// 上游任務完成后在當前主線程中同步執(zhí)行處理，指定線程池，向下傳遞上游處理結果或異常
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {
...
}
// 上游任務異常時處理，在主線程中完成
public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn) {
...
}
示例代碼如下：
@Test
public void testWhenComplete() throws InterruptedException, ExecutionException {
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
return "test";
}).whenCompleteAsync((result, exception) -> {
if (result != null) {
System.out.println(result);
} else {
exception.printStackTrace();
}
});
System.out.println(f1.get());
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
int i = 1 / 0;
return "test";
}).exceptionally((e) -> {
e.printStackTrace();
return "error";
});
System.out.println(f2.get());
}
---
test
error
java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
...
whenComplete方法入?yún)⑹荁iConsumer，它是一個純消費者函數(shù)，不會修改返回值及其類型。
下面一組接口除了whenComplete的功能外，同時具備轉換結果的功能，通過參數(shù)BiFunction實現(xiàn)：
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) {
...
}
示例代碼如下：
@Test
public void testHandle() throws InterruptedException, ExecutionException {
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
return "100";
});
CompletableFuture<Integer> f2 = f1.handleAsync((result, exception) -> {
if(exception != null) {
exception.printStackTrace();
}
return Integer.parseInt(result) * 10;
});
System.out.println(f2.get());
}
---
1000
handle方法依然保留了對異常請款的處理，在BiFunction中指定其第二個參數(shù)類型是Throwble。
接下來一組方法thenApply與handle類似，也可以對上游結果進行轉換，同時忽略對異常情況的處理。
public <U> CompletableFuture<U> thenApplyFunction<? 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) {
...
}
示例代碼：
@Test
public void testThenApply() throws InterruptedException, ExecutionException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
//int i= 1/0;
return "100";
});
CompletableFuture<Integer> f = future.thenApply(t -> Integer.parseInt(t) * 10).thenApply(t -> t * 10);
System.out.println(f.get());
}
---
10000
上述方法處理完成后，都會返回計算結果，CompletableFuture還提供了一組處理方法，只對上游處理結果進行消費，且沒有返回，如果上游發(fā)生異常，則不執(zhí)行該方法；同時，CompletableFuture還提供了一組加強版方法，提供兩個CompletableFuture任務都完成或完成一個時執(zhí)行的方法。
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) {
...
}
public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action) {
...
}
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action) {
...
}
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action, Executor executor) {
...
}
public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) {
...
}
public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) {
...
}
public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action,Executor executor) {
...
}
示例代碼：
@Test
public void testThenAccept() throws InterruptedException, ExecutionException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
int i = 1/0;
return "1";
});
CompletableFuture<Void> f = future.thenAccept(t -> {
System.out.println(t);
});
}
---
上述方法都依賴上游處理的結果，CompletableFuture還提供了一組方法，不依賴上游的結果，如下
public CompletableFuture<Void> thenRun(Runnable action) {
...
}
public CompletableFuture<Void> thenRunAsync(Runnable action) {
...
}
public CompletableFuture<Void> thenRunAsync(Runnable action,Executor executor) {
...
}
示例代碼：
@Test
public void testThenRun() throws InterruptedException, ExecutionException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
return "result";
});
CompletableFuture<Void> f = future.thenRunAsync(() -> {
System.out.println("future then run");
});
}
---
future then run
thenCompose這組方法與thenApply類似，都可以將上游執(zhí)行結果作為本stage入?yún)⒗^續(xù)計算，并轉換返回類型，不同的是thenCompose在一個新CompletableFuture對象執(zhí)行計算。
public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) {
...
}
public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) {
...
}
public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,Executor executor) {
...
}
示例代碼：
@Test
public void testThenCompose() throws InterruptedException, ExecutionException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
return "10";
});
CompletableFuture<Integer> f = future.thenCompose(v -> {
return CompletableFuture.supplyAsync(() -> {
return Integer.parseInt(v) * 100;
});
});
System.out.println(f.get());
}
---
1000
thenCombine這組方法與thenAccepBoth類似，不同的是thenCombine有返回值：
public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn) {
...
}
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn) {
...
}
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
...
}
示例代碼：
@Test
public void testThenCombine() throws InterruptedException, ExecutionException {
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
return "abc";
});
CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> {
return 123;
});
CompletableFuture<String> f3 = f1.thenCombine(f2, (m,n) -> {
return m + " is " + n;
});
System.out.println(f3.get());
}
---
abc is 123
最后兩個方法用于組合多個CompletableFuture，allOf方法是當所有CompletableFuture執(zhí)行完成后執(zhí)行計算，anyOf方法是任意一個CompletableFuture執(zhí)行完成后執(zhí)行計算。
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
...
}
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
...
}
示例代碼：
@Test
public void testAllof() throws InterruptedException, ExecutionException {
List<CompletableFuture<Integer>> list = new ArrayList<>();
Random random = new Random();
for (int i=1; i < 10; i++) {
list.add(CompletableFuture.supplyAsync(() -> {
return random.nextInt(100);
}));
}
CompletableFuture<Void> fn = CompletableFuture.allOf(list.toArray(new CompletableFuture[list.size()]));
CompletableFuture<List<Integer>> fr = fn.thenApply(v -> {
return list.stream().map(CompletableFuture::join).collect(Collectors.toList());
});
System.out.println(fr.get());
}
---
[3, 97, 11, 93, 22, 77, 68, 26, 89]
總結
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關文章
基于SpringBoot+Mybatis實現(xiàn)Mysql分表
這篇文章主要為大家詳細介紹了基于SpringBoot+Mybatis實現(xiàn)Mysql分表的相關知識,文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學習一下2025-04-04
本文給大家介紹的是java使用動態(tài)代理類實現(xiàn)動態(tài)代理的方法和示例，這里推薦給大家，有需要的小伙伴參考下吧2015-02-02
SpringBoot基于HttpMessageConverter實現(xiàn)全局日期格式化
這篇文章主要介紹了SpringBoot基于HttpMessageConverter實現(xiàn)全局日期格式化，使用Jackson消息轉換器，非常具有實用價值，需要的朋友可以參考下2018-12-12
這篇文章主要為大家詳細介紹了Java實現(xiàn)24點小游戲，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-01-01
JNDI是Java EE的重要部分，本文主要介紹了Java之JNDI注入的實現(xiàn)，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-11-11
這篇文章主要介紹了Java應用CPU使用率過高排查方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-07-07
一不小心就讓Java開發(fā)踩坑的fail-fast是個什么鬼？(推薦)
這篇文章主要介紹了Java fail-fast，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-04-04
這篇文章主要給大家介紹了關于Java查詢MongoDB數(shù)據(jù)庫的一些相關案例,Java可以使用MongoDB的官方Java驅動程序來連接和操作MongoDB數(shù)據(jù)庫,需要的朋友可以參考下2023-07-07
這篇文章主要為大家詳細介紹了Java實現(xiàn)一行一行讀取文本的多種方法,文中的示例代碼講解詳細,具有一定的借鑒價值,有需要的小伙伴可以了解下2025-10-10
這篇文章主要介紹了,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2019-08-08