異步編程利器:CompletableFuture 詳解
前言
最近剛好使用 CompeletableFuture 優化了項目中的代碼,所以跟大家一起學習 CompletableFuture。
- 公衆號: 撿田螺的小男孩
一個例子回顧 Future
因爲 CompletableFuture 實現了Future
接口,我們先來回顧 Future 吧。
Future 是 Java5 新加的一個接口,它提供了一種異步並行計算的功能。如果主線程需要執行一個很耗時的計算任務,我們就可以通過 future 把這個任務放到異步線程中執行。主線程繼續處理其他任務,處理完成後,再通過 Future 獲取計算結果。
來看個簡單例子吧,假設我們有兩個任務服務,一個查詢用戶基本信息,一個是查詢用戶勳章信息。如下,
public class UserInfoService {
public UserInfo getUserInfo(Long userId) throws InterruptedException {
Thread.sleep(300);//模擬調用耗時
return new UserInfo("666", "撿田螺的小男孩", 27); //一般是查數據庫,或者遠程調用返回的
}
}
public class MedalService {
public MedalInfo getMedalInfo(long userId) throws InterruptedException {
Thread.sleep(500); //模擬調用耗時
return new MedalInfo("666", "守護勳章");
}
}
接下來,我們來演示下,在主線程中是如何使用 Future 來進行異步調用的。
public class FutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
UserInfoService userInfoService = new UserInfoService();
MedalService medalService = new MedalService();
long userId =666L;
long startTime = System.currentTimeMillis();
//調用用戶服務獲取用戶基本信息
FutureTask<UserInfo> userInfoFutureTask = new FutureTask<>(new Callable<UserInfo>() {
@Override
public UserInfo call() throws Exception {
return userInfoService.getUserInfo(userId);
}
});
executorService.submit(userInfoFutureTask);
Thread.sleep(300); //模擬主線程其它操作耗時
FutureTask<MedalInfo> medalInfoFutureTask = new FutureTask<>(new Callable<MedalInfo>() {
@Override
public MedalInfo call() throws Exception {
return medalService.getMedalInfo(userId);
}
});
executorService.submit(medalInfoFutureTask);
UserInfo userInfo = userInfoFutureTask.get();//獲取個人信息結果
MedalInfo medalInfo = medalInfoFutureTask.get();//獲取勳章信息結果
System.out.println("總共用時" + (System.currentTimeMillis() - startTime) + "ms");
}
}
運行結果:
總共用時806ms
如果我們不使用 Future 進行並行異步調用,而是在主線程串行進行的話,耗時大約爲 300+500+300 = 1100 ms。可以發現,future + 線程池異步配合,提高了程序的執行效率。
但是 Future 對於結果的獲取,不是很友好,只能通過阻塞或者輪詢的方式得到任務的結果。
-
Future.get() 就是阻塞調用,在線程獲取結果之前 get 方法會一直阻塞。
-
Future 提供了一個 isDone 方法,可以在程序中輪詢這個方法查詢執行結果。
阻塞的方式和異步編程的設計理念相違背,而輪詢的方式會耗費無謂的 CPU 資源。因此,JDK8 設計出 CompletableFuture。CompletableFuture 提供了一種觀察者模式類似的機制,可以讓任務執行完成後通知監聽的一方。
一個例子走進 CompletableFuture
我們還是基於以上 Future 的例子,改用 CompletableFuture 來實現
public class FutureTest {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
UserInfoService userInfoService = new UserInfoService();
MedalService medalService = new MedalService();
long userId =666L;
long startTime = System.currentTimeMillis();
//調用用戶服務獲取用戶基本信息
CompletableFuture<UserInfo> completableUserInfoFuture = CompletableFuture.supplyAsync(() -> userInfoService.getUserInfo(userId));
Thread.sleep(300); //模擬主線程其它操作耗時
CompletableFuture<MedalInfo> completableMedalInfoFuture = CompletableFuture.supplyAsync(() -> medalService.getMedalInfo(userId));
UserInfo userInfo = completableUserInfoFuture.get(2,TimeUnit.SECONDS);//獲取個人信息結果
MedalInfo medalInfo = completableMedalInfoFuture.get();//獲取勳章信息結果
System.out.println("總共用時" + (System.currentTimeMillis() - startTime) + "ms");
}
}
可以發現,使用 CompletableFuture,代碼簡潔了很多。CompletableFuture 的 supplyAsync 方法,提供了異步執行的功能,線程池也不用單獨創建了。實際上,它 CompletableFuture 使用了默認線程池是 ForkJoinPool.commonPool。
CompletableFuture 提供了幾十種方法,輔助我們的異步任務場景。這些方法包括創建異步任務、任務異步回調、多個任務組合處理等方面。我們一起來學習吧
CompletableFuture 使用場景
創建異步任務
CompletableFuture 創建異步任務,一般有 supplyAsync 和 runAsync 兩個方法
創建異步任務
-
supplyAsync 執行 CompletableFuture 任務,支持返回值
-
runAsync 執行 CompletableFuture 任務,沒有返回值。
supplyAsync 方法
//使用默認內置線程池ForkJoinPool.commonPool(),根據supplier構建執行任務
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
//自定義線程,根據supplier構建執行任務
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
runAsync 方法
//使用默認內置線程池ForkJoinPool.commonPool(),根據runnable構建執行任務
public static CompletableFuture<Void> runAsync(Runnable runnable)
//自定義線程,根據runnable構建執行任務
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
實例代碼如下:
public class FutureTest {
public static void main(String[] args) {
//可以自定義線程池
ExecutorService executor = Executors.newCachedThreadPool();
//runAsync的使用
CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> System.out.println("run,關注公衆號:撿田螺的小男孩"), executor);
//supplyAsync的使用
CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> {
System.out.print("supply,關注公衆號:撿田螺的小男孩");
return "撿田螺的小男孩"; }, executor);
//runAsync的future沒有返回值,輸出null
System.out.println(runFuture.join());
//supplyAsync的future,有返回值
System.out.println(supplyFuture.join());
executor.shutdown(); // 線程池需要關閉
}
}
//輸出
run,關注公衆號:撿田螺的小男孩
null
supply,關注公衆號:撿田螺的小男孩撿田螺的小男孩
任務異步回調
1. thenRun/thenRunAsync
public CompletableFuture<Void> thenRun(Runnable action);
public CompletableFuture<Void> thenRunAsync(Runnable action);
CompletableFuture 的 thenRun 方法,通俗點講就是,做完第一個任務後,再做第二個任務。某個任務執行完成後,執行回調方法;但是前後兩個任務沒有參數傳遞,第二個任務也沒有返回值
public class FutureThenRunTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("先執行第一個CompletableFuture方法任務");
return "撿田螺的小男孩";
}
);
CompletableFuture thenRunFuture = orgFuture.thenRun(() -> {
System.out.println("接着執行第二個任務");
});
System.out.println(thenRunFuture.get());
}
}
//輸出
先執行第一個CompletableFuture方法任務
接着執行第二個任務
null
thenRun 和 thenRunAsync 有什麼區別呢?可以看下源碼哈:
private static final Executor asyncPool = useCommonPool ?
ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
public CompletableFuture<Void> thenRun(Runnable action) {
return uniRunStage(null, action);
}
public CompletableFuture<Void> thenRunAsync(Runnable action) {
return uniRunStage(asyncPool, action);
}
如果你執行第一個任務的時候,傳入了一個自定義線程池:
-
調用 thenRun 方法執行第二個任務時,則第二個任務和第一個任務是共用同一個線程池。
-
調用 thenRunAsync 執行第二個任務時,則第一個任務使用的是你自己傳入的線程池,第二個任務使用的是 ForkJoin 線程池
TIPS: 後面介紹的 thenAccept 和 thenAcceptAsync,thenApply 和 thenApplyAsync 等,它們之間的區別也是這個哈。
2.thenAccept/thenAcceptAsync
CompletableFuture 的 thenAccept 方法表示,第一個任務執行完成後,執行第二個回調方法任務,會將該任務的執行結果,作爲入參,傳遞到回調方法中,但是回調方法是沒有返回值的。
public class FutureThenAcceptTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("原始CompletableFuture方法任務");
return "撿田螺的小男孩";
}
);
CompletableFuture thenAcceptFuture = orgFuture.thenAccept((a) -> {
if ("撿田螺的小男孩".equals(a)) {
System.out.println("關注了");
}
System.out.println("先考慮考慮");
});
System.out.println(thenAcceptFuture.get());
}
}
3. thenApply/thenApplyAsync
CompletableFuture 的 thenApply 方法表示,第一個任務執行完成後,執行第二個回調方法任務,會將該任務的執行結果,作爲入參,傳遞到回調方法中,並且回調方法是有返回值的。
public class FutureThenApplyTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("原始CompletableFuture方法任務");
return "撿田螺的小男孩";
}
);
CompletableFuture<String> thenApplyFuture = orgFuture.thenApply((a) -> {
if ("撿田螺的小男孩".equals(a)) {
return "關注了";
}
return "先考慮考慮";
});
System.out.println(thenApplyFuture.get());
}
}
//輸出
原始CompletableFuture方法任務
關注了
4. exceptionally
CompletableFuture 的 exceptionally 方法表示,某個任務執行異常時,執行的回調方法; 並且有拋出異常作爲參數,傳遞到回調方法。
public class FutureExceptionTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("當前線程名稱:" + Thread.currentThread().getName());
throw new RuntimeException();
}
);
CompletableFuture<String> exceptionFuture = orgFuture.exceptionally((e) -> {
e.printStackTrace();
return "你的程序異常啦";
});
System.out.println(exceptionFuture.get());
}
}
//輸出
當前線程名稱:ForkJoinPool.commonPool-worker-1
java.util.concurrent.CompletionException: java.lang.RuntimeException
at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592)
at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.lang.RuntimeException
at cn.eovie.future.FutureWhenTest.lambda$main$0(FutureWhenTest.java:13)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
... 5 more
你的程序異常啦
5. whenComplete 方法
CompletableFuture 的 whenComplete 方法表示,某個任務執行完成後,執行的回調方法,無返回值;並且 whenComplete 方法返回的 CompletableFuture 的 result 是上個任務的結果。
public class FutureWhenTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("當前線程名稱:" + Thread.currentThread().getName());
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "撿田螺的小男孩";
}
);
CompletableFuture<String> rstFuture = orgFuture.whenComplete((a, throwable) -> {
System.out.println("當前線程名稱:" + Thread.currentThread().getName());
System.out.println("上個任務執行完啦,還把" + a + "傳過來");
if ("撿田螺的小男孩".equals(a)) {
System.out.println("666");
}
System.out.println("233333");
});
System.out.println(rstFuture.get());
}
}
//輸出
當前線程名稱:ForkJoinPool.commonPool-worker-1
當前線程名稱:ForkJoinPool.commonPool-worker-1
上個任務執行完啦,還把撿田螺的小男孩傳過來
666
233333
撿田螺的小男孩
6. handle 方法
CompletableFuture 的 handle 方法表示,某個任務執行完成後,執行回調方法,並且是有返回值的; 並且 handle 方法返回的 CompletableFuture 的 result 是回調方法執行的結果。
public class FutureHandlerTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("當前線程名稱:" + Thread.currentThread().getName());
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "撿田螺的小男孩";
}
);
CompletableFuture<String> rstFuture = orgFuture.handle((a, throwable) -> {
System.out.println("上個任務執行完啦,還把" + a + "傳過來");
if ("撿田螺的小男孩".equals(a)) {
System.out.println("666");
return "關注了";
}
System.out.println("233333");
return null;
});
System.out.println(rstFuture.get());
}
}
//輸出
當前線程名稱:ForkJoinPool.commonPool-worker-1
上個任務執行完啦,還把撿田螺的小男孩傳過來
666
關注了
多個任務組合處理
AND 組合關係
thenCombine / thenAcceptBoth / runAfterBoth 都表示:將兩個 CompletableFuture 組合起來,只有這兩個都正常執行完了,纔會執行某個任務。
區別在於:
-
thenCombine:會將兩個任務的執行結果作爲方法入參,傳遞到指定方法中,且有返回值
-
thenAcceptBoth: 會將兩個任務的執行結果作爲方法入參,傳遞到指定方法中,且無返回值
-
runAfterBoth 不會把執行結果當做方法入參,且沒有返回值。
public class ThenCombineTest {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<String> first = CompletableFuture.completedFuture("第一個異步任務");
ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture
//第二個異步任務
.supplyAsync(() -> "第二個異步任務", executor)
// (w, s) -> System.out.println(s) 是第三個任務
.thenCombineAsync(first, (s, w) -> {
System.out.println(w);
System.out.println(s);
return "兩個異步任務的組合";
}, executor);
System.out.println(future.join());
executor.shutdown();
}
}
//輸出
第一個異步任務
第二個異步任務
兩個異步任務的組合
OR 組合的關係
applyToEither / acceptEither / runAfterEither 都表示:將兩個 CompletableFuture 組合起來,只要其中一個執行完了, 就會執行某個任務。
區別在於:
-
applyToEither:會將已經執行完成的任務,作爲方法入參,傳遞到指定方法中,且有返回值
-
acceptEither: 會將已經執行完成的任務,作爲方法入參,傳遞到指定方法中,且無返回值
-
runAfterEither:不會把執行結果當做方法入參,且沒有返回值。
public class AcceptEitherTest {
public static void main(String[] args) {
//第一個異步任務,休眠2秒,保證它執行晚點
CompletableFuture<String> first = CompletableFuture.supplyAsync(()->{
try{
Thread.sleep(2000L);
System.out.println("執行完第一個異步任務");}
catch (Exception e){
return "第一個任務異常";
}
return "第一個異步任務";
});
ExecutorService executor = Executors.newSingleThreadExecutor();
CompletableFuture<Void> future = CompletableFuture
//第二個異步任務
.supplyAsync(() -> {
System.out.println("執行完第二個任務");
return "第二個任務";}
, executor)
//第三個任務
.acceptEitherAsync(first, System.out::println, executor);
executor.shutdown();
}
}
//輸出
執行完第二個任務
第二個任務
AllOf
所有任務都執行完成後,才執行 allOf 返回的 CompletableFuture。如果任意一個任務異常,allOf 的 CompletableFuture,執行 get 方法,會拋出異常
public class allOfFutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Void> a = CompletableFuture.runAsync(()->{
System.out.println("我執行完了");
});
CompletableFuture<Void> b = CompletableFuture.runAsync(() -> {
System.out.println("我也執行完了");
});
CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(a, b).whenComplete((m,k)->{
System.out.println("finish");
});
}
}
//輸出
我執行完了
我也執行完了
finish
AnyOf
任意一個任務執行完,就執行 anyOf 返回的 CompletableFuture。如果執行的任務異常,anyOf 的 CompletableFuture,執行 get 方法,會拋出異常
public class AnyOfFutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Void> a = CompletableFuture.runAsync(()->{
try {
Thread.sleep(3000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("我執行完了");
});
CompletableFuture<Void> b = CompletableFuture.runAsync(() -> {
System.out.println("我也執行完了");
});
CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(a, b).whenComplete((m,k)->{
System.out.println("finish");
// return "撿田螺的小男孩";
});
anyOfFuture.join();
}
}
//輸出
我也執行完了
finish
thenCompose
thenCompose 方法會在某個任務執行完成後,將該任務的執行結果, 作爲方法入參, 去執行指定的方法。該方法會返回一個新的 CompletableFuture 實例
-
如果該 CompletableFuture 實例的 result 不爲 null,則返回一個基於該 result 新的 CompletableFuture 實例;
-
如果該 CompletableFuture 實例爲 null,然後就執行這個新任務
public class ThenComposeTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> f = CompletableFuture.completedFuture("第一個任務");
//第二個異步任務
ExecutorService executor = Executors.newSingleThreadExecutor();
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> "第二個任務", executor)
.thenComposeAsync(data -> {
System.out.println(data); return f; //使用第一個任務作爲返回
}, executor);
System.out.println(future.join());
executor.shutdown();
}
}
//輸出
第二個任務
第一個任務
CompletableFuture 使用有哪些注意點
CompletableFuture 使我們的異步編程更加便利的、代碼更加優雅的同時,我們也要關注下它,使用的一些注意點。
1. Future 需要獲取返回值,才能獲取異常信息
ExecutorService executorService = new ThreadPoolExecutor(5, 10, 5L,
TimeUnit.SECONDS, new ArrayBlockingQueue<>(10));
CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
int a = 0;
int b = 666;
int c = b / a;
return true;
},executorService).thenAccept(System.out::println);
//如果不加 get()方法這一行,看不到異常信息
//future.get();
Future 需要獲取返回值,才能獲取到異常信息。如果不加 get()/join() 方法,看不到異常信息。小夥伴們使用的時候,注意一下哈, 考慮是否加 try...catch... 或者使用 exceptionally 方法。
2. CompletableFuture 的 get() 方法是阻塞的。
CompletableFuture 的 get() 方法是阻塞的,如果使用它來獲取異步調用的返回值,需要添加超時時間~
//反例
CompletableFuture.get();
//正例
CompletableFuture.get(5, TimeUnit.SECONDS);
3. 默認線程池的注意點
CompletableFuture 代碼中又使用了默認的線程池,處理的線程個數是電腦 CPU 核數 - 1。在大量請求過來的時候,處理邏輯複雜的話,響應會很慢。一般建議使用自定義線程池,優化線程池配置參數。
4. 自定義線程池時,注意飽和策略
CompletableFuture 的 get() 方法是阻塞的,我們一般建議使用future.get(3, TimeUnit.SECONDS)
。並且一般建議使用自定義線程池。
但是如果線程池拒絕策略是DiscardPolicy
或者DiscardOldestPolicy
,當線程池飽和時,會直接丟棄任務,不會拋棄異常。因此建議,CompletableFuture 線程池策略最好使用 AbortPolicy,然後耗時的異步線程,做好線程池隔離哈。
參考資料
[1]
Java8 CompletableFuture 用法全解: https://blog.csdn.net/qq_31865983/article/details/106137777
[2]
基礎篇:異步編程不會?我教你啊!: https://juejin.cn/post/6902655550031413262#heading-5
[3]
CompletableFuture get 方法一直阻塞或拋出 TimeoutException: https://blog.csdn.net/xiaolyuh123/article/details/85023269
[4]
編程老司機帶你玩轉 CompletableFuture 異步編程: https://zhuanlan.zhihu.com/p/111841508
[5]
解決 CompletableFuture 異常阻塞: https://blog.csdn.net/weixin_42742643/article/details/111638260
本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源:https://mp.weixin.qq.com/s/Q5GJkaHucMEtNPJ-XO7A4A