Java8中CompletableFuture的用法全解
更新時間：2022年01月26日 09:25:21 作者：孫大圣666
這篇文章主要給大家介紹了關(guān)于Java8中CompletableFuture用法的相關(guān)資料,文中通過實例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友可以參考下
前言
CompletableFuture實現(xiàn)了CompletionStage接口和Future接口，前者是對后者的一個擴展，增加了異步回調(diào)、流式處理、多個Future組合處理的能力，使Java在處理多任務(wù)的協(xié)同工作時更加順暢便利。
一、創(chuàng)建異步任務(wù)
1、Future.submit
通常的線程池接口類ExecutorService，其中execute方法的返回值是void，即無法獲取異步任務(wù)的執(zhí)行狀態(tài)，3個重載的submit方法的返回值是Future，可以據(jù)此獲取任務(wù)執(zhí)行的狀態(tài)和結(jié)果，示例如下：
@Test
public void test3() throws Exception {
// 創(chuàng)建異步執(zhí)行任務(wù):
ExecutorService executorService= Executors.newSingleThreadExecutor();
Future<Double> cf = executorService.submit(()->{
System.out.println(Thread.currentThread()+" start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(false){
throw new RuntimeException("test");
}else{
System.out.println(Thread.currentThread()+" exit,time->"+System.currentTimeMillis());
return 1.2;
}
});
System.out.println("main thread start,time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成,如果已完成則直接返回結(jié)果
//如果執(zhí)行任務(wù)異常，則get方法會把之前捕獲的異常重新拋出
System.out.println("run result->"+cf.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}執(zhí)行結(jié)果如下：
子線程是異步執(zhí)行的，主線程休眠等待子線程執(zhí)行完成，子線程執(zhí)行完成后喚醒主線程，主線程獲取任務(wù)執(zhí)行結(jié)果后退出。
很多博客說使用不帶等待時間限制的get方法時，如果子線程執(zhí)行異常了會導(dǎo)致主線程長期阻塞，這其實是錯誤的，子線程執(zhí)行異常時其異常會被捕獲，然后修改任務(wù)的狀態(tài)為異常結(jié)束并喚醒等待的主線程，get方法判斷任務(wù)狀態(tài)發(fā)生變更，就終止等待了，并拋出異常，可參考《Java8 AbstractExecutorService 和 FutureTask 源碼解析》中FutureTask的實現(xiàn)。將上述用例中if(false)改成if(true) ，執(zhí)行結(jié)果如下：
get方法拋出異常導(dǎo)致主線程異常終止。
2、supplyAsync / runAsync
supplyAsync表示創(chuàng)建帶返回值的異步任務(wù)的，相當于ExecutorService submit(Callable<T> task) 方法，runAsync表示創(chuàng)建無返回值的異步任務(wù)，相當于ExecutorService submit(Runnable task)方法，這兩方法的效果跟submit是一樣的，測試用例如下：
@Test
public void test2() throws Exception {
// 創(chuàng)建異步執(zhí)行任務(wù)，有返回值
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(true){
throw new RuntimeException("test");
}else{
System.out.println(Thread.currentThread()+" exit,time->"+System.currentTimeMillis());
return 1.2;
}
});
System.out.println("main thread start,time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成
System.out.println("run result->"+cf.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}
@Test
public void test4() throws Exception {
// 創(chuàng)建異步執(zhí)行任務(wù)，無返回值
CompletableFuture cf = CompletableFuture.runAsync(()->{
System.out.println(Thread.currentThread()+" start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(false){
throw new RuntimeException("test");
}else{
System.out.println(Thread.currentThread()+" exit,time->"+System.currentTimeMillis());
}
});
System.out.println("main thread start,time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成
System.out.println("run result->"+cf.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}這兩方法各有一個重載版本，可以指定執(zhí)行異步任務(wù)的Executor實現(xiàn)，如果不指定，默認使用ForkJoinPool.commonPool()，如果機器是單核的，則默認使用ThreadPerTaskExecutor，該類是一個內(nèi)部類，每次執(zhí)行execute都會創(chuàng)建一個新線程。測試用例如下：
@Test
public void test2() throws Exception {
ForkJoinPool pool=new ForkJoinPool();
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(true){
throw new RuntimeException("test");
}else{
System.out.println(Thread.currentThread()+" exit,time->"+System.currentTimeMillis());
return 1.2;
}
},pool);
System.out.println("main thread start,time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成
System.out.println("run result->"+cf.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}
@Test
public void test4() throws Exception {
ExecutorService executorService= Executors.newSingleThreadExecutor();
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture cf = CompletableFuture.runAsync(()->{
System.out.println(Thread.currentThread()+" start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(false){
throw new RuntimeException("test");
}else{
System.out.println(Thread.currentThread()+" exit,time->"+System.currentTimeMillis());
}
},executorService);
System.out.println("main thread start,time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成
System.out.println("run result->"+cf.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}二、異步回調(diào)
1、thenApply / thenApplyAsync
thenApply 表示某個任務(wù)執(zhí)行完成后執(zhí)行的動作，即回調(diào)方法，會將該任務(wù)的執(zhí)行結(jié)果即方法返回值作為入?yún)鬟f到回調(diào)方法中，測試用例如下：
@Test
public void test5() throws Exception {
ForkJoinPool pool=new ForkJoinPool();
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
return 1.2;
},pool);
//cf關(guān)聯(lián)的異步任務(wù)的返回值作為方法入?yún)?，傳入到thenApply的方法中
//thenApply這里實際創(chuàng)建了一個新的CompletableFuture實例
CompletableFuture<String> cf2=cf.thenApply((result)->{
System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
return "test:"+result;
});
System.out.println("main thread start cf.get(),time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成
System.out.println("run result->"+cf.get());
System.out.println("main thread start cf2.get(),time->"+System.currentTimeMillis());
System.out.println("run result->"+cf2.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}其執(zhí)行結(jié)果如下：
job1執(zhí)行結(jié)束后，將job1的方法返回值作為入?yún)鬟f到j(luò)ob2中并立即執(zhí)行job2。thenApplyAsync與thenApply的區(qū)別在于，前者是將job2提交到線程池中異步執(zhí)行，實際執(zhí)行job2的線程可能是另外一個線程，后者是由執(zhí)行job1的線程立即執(zhí)行job2，即兩個job都是同一個線程執(zhí)行的。將上述測試用例中thenApply改成thenApplyAsync后，執(zhí)行結(jié)果如下：
從輸出可知，執(zhí)行job1和job2是兩個不同的線程。thenApplyAsync有一個重載版本，可以指定執(zhí)行異步任務(wù)的Executor實現(xiàn)，如果不指定，默認使用ForkJoinPool.commonPool()。 下述的多個方法，每個方法都有兩個以Async結(jié)尾的方法，一個使用默認的Executor實現(xiàn)，一個使用指定的Executor實現(xiàn)，不帶Async的方法是由觸發(fā)該任務(wù)的線程執(zhí)行該任務(wù)，帶Async的方法是由觸發(fā)該任務(wù)的線程將任務(wù)提交到線程池，執(zhí)行任務(wù)的線程跟觸發(fā)任務(wù)的線程不一定是同一個。
2、thenAccept / thenRun
thenAccept 同 thenApply 接收上一個任務(wù)的返回值作為參數(shù)，但是無返回值；thenRun 的方法沒有入?yún)?，也買有返回值，測試用例如下：
@Test
public void test6() throws Exception {
ForkJoinPool pool=new ForkJoinPool();
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
return 1.2;
},pool);
//cf關(guān)聯(lián)的異步任務(wù)的返回值作為方法入?yún)ⅲ瑐魅氲絫henApply的方法中
CompletableFuture cf2=cf.thenApply((result)->{
System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
return "test:"+result;
}).thenAccept((result)-> { //接收上一個任務(wù)的執(zhí)行結(jié)果作為入?yún)?，但是沒有返回值
System.out.println(Thread.currentThread()+" start job3,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(result);
System.out.println(Thread.currentThread()+" exit job3,time->"+System.currentTimeMillis());
}).thenRun(()->{ //無入?yún)?，也沒有返回值
System.out.println(Thread.currentThread()+" start job4,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println("thenRun do something");
System.out.println(Thread.currentThread()+" exit job4,time->"+System.currentTimeMillis());
});
System.out.println("main thread start cf.get(),time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成
System.out.println("run result->"+cf.get());
System.out.println("main thread start cf2.get(),time->"+System.currentTimeMillis());
//cf2 等待最后一個thenRun執(zhí)行完成
System.out.println("run result->"+cf2.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}其執(zhí)行結(jié)果如下：
3、 exceptionally
exceptionally方法指定某個任務(wù)執(zhí)行異常時執(zhí)行的回調(diào)方法，會將拋出異常作為參數(shù)傳遞到回調(diào)方法中，如果該任務(wù)正常執(zhí)行則會exceptionally方法返回的CompletionStage的result就是該任務(wù)正常執(zhí)行的結(jié)果，測試用例如下：
@Test
public void test2() throws Exception {
ForkJoinPool pool=new ForkJoinPool();
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+"job1 start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(true){
throw new RuntimeException("test");
}else{
System.out.println(Thread.currentThread()+"job1 exit,time->"+System.currentTimeMillis());
return 1.2;
}
},pool);
//cf執(zhí)行異常時，將拋出的異常作為入?yún)鬟f給回調(diào)方法
CompletableFuture<Double> cf2= cf.exceptionally((param)->{
System.out.println(Thread.currentThread()+" start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println("error stack trace->");
param.printStackTrace();
System.out.println(Thread.currentThread()+" exit,time->"+System.currentTimeMillis());
return -1.1;
});
//cf正常執(zhí)行時執(zhí)行的邏輯，如果執(zhí)行異常則不調(diào)用此邏輯
CompletableFuture cf3=cf.thenAccept((param)->{
System.out.println(Thread.currentThread()+"job2 start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println("param->"+param);
System.out.println(Thread.currentThread()+"job2 exit,time->"+System.currentTimeMillis());
});
System.out.println("main thread start,time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成,此處無論是job2和job3都可以實現(xiàn)job2退出，主線程才退出，如果是cf，則主線程不會等待job2執(zhí)行完成自動退出了
//cf2.get時，沒有異常，但是依然有返回值，就是cf的返回值
System.out.println("run result->"+cf2.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}其輸出如下：
拋出異常后，只有cf2執(zhí)行了，cf3沒有執(zhí)行。將上述示例中的if(true) 改成if(false)，其輸出如下：
cf2沒有指定，其result就是cf執(zhí)行的結(jié)果，理論上cf2.get應(yīng)該立即返回的，此處是等待了cf3，即job2執(zhí)行完成后才返回，具體原因且待下篇源碼分析時再做探討。
4、whenComplete
whenComplete是當某個任務(wù)執(zhí)行完成后執(zhí)行的回調(diào)方法，會將執(zhí)行結(jié)果或者執(zhí)行期間拋出的異常傳遞給回調(diào)方法，如果是正常執(zhí)行則異常為null，回調(diào)方法對應(yīng)的CompletableFuture的result和該任務(wù)一致，如果該任務(wù)正常執(zhí)行，則get方法返回執(zhí)行結(jié)果，如果是執(zhí)行異常，則get方法拋出異常。測試用例如下：
@Test
public void test10() throws Exception {
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+"job1 start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(false){
throw new RuntimeException("test");
}else{
System.out.println(Thread.currentThread()+"job1 exit,time->"+System.currentTimeMillis());
return 1.2;
}
});
//cf執(zhí)行完成后會將執(zhí)行結(jié)果和執(zhí)行過程中拋出的異常傳入回調(diào)方法，如果是正常執(zhí)行的則傳入的異常為null
CompletableFuture<Double> cf2=cf.whenComplete((a,b)->{
System.out.println(Thread.currentThread()+"job2 start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(b!=null){
System.out.println("error stack trace->");
b.printStackTrace();
}else{
System.out.println("run succ,result->"+a);
}
System.out.println(Thread.currentThread()+"job2 exit,time->"+System.currentTimeMillis());
});
//等待子任務(wù)執(zhí)行完成
System.out.println("main thread start wait,time->"+System.currentTimeMillis());
//如果cf是正常執(zhí)行的，cf2.get的結(jié)果就是cf執(zhí)行的結(jié)果
//如果cf是執(zhí)行異常，則cf2.get會拋出異常
System.out.println("run result->"+cf2.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}執(zhí)行結(jié)果如下：
將上述示例中的if(false) 改成if(true)，其輸出如下：
5、handle
跟whenComplete基本一致，區(qū)別在于handle的回調(diào)方法有返回值，且handle方法返回的CompletableFuture的result是回調(diào)方法的執(zhí)行結(jié)果或者回調(diào)方法執(zhí)行期間拋出的異常，與原始CompletableFuture的result無關(guān)了。測試用例如下：
@Test
public void test10() throws Exception {
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+"job1 start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(true){
throw new RuntimeException("test");
}else{
System.out.println(Thread.currentThread()+"job1 exit,time->"+System.currentTimeMillis());
return 1.2;
}
});
//cf執(zhí)行完成后會將執(zhí)行結(jié)果和執(zhí)行過程中拋出的異常傳入回調(diào)方法，如果是正常執(zhí)行的則傳入的異常為null
CompletableFuture<String> cf2=cf.handle((a,b)->{
System.out.println(Thread.currentThread()+"job2 start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(b!=null){
System.out.println("error stack trace->");
b.printStackTrace();
}else{
System.out.println("run succ,result->"+a);
}
System.out.println(Thread.currentThread()+"job2 exit,time->"+System.currentTimeMillis());
if(b!=null){
return "run error";
}else{
return "run succ";
}
});
//等待子任務(wù)執(zhí)行完成
System.out.println("main thread start wait,time->"+System.currentTimeMillis());
//get的結(jié)果是cf2的返回值，跟cf沒關(guān)系了
System.out.println("run result->"+cf2.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}其執(zhí)行結(jié)果如下：
將上述示例中的if(true) 改成if(false)，其輸出如下：
三、組合處理
1、thenCombine / thenAcceptBoth / runAfterBoth
這三個方法都是將兩個CompletableFuture組合起來，只有這兩個都正常執(zhí)行完了才會執(zhí)行某個任務(wù)，區(qū)別在于，thenCombine會將兩個任務(wù)的執(zhí)行結(jié)果作為方法入?yún)鬟f到指定方法中，且該方法有返回值；thenAcceptBoth同樣將兩個任務(wù)的執(zhí)行結(jié)果作為方法入?yún)?，但是無返回值；runAfterBoth沒有入?yún)ⅲ矝]有返回值。注意兩個任務(wù)中只要有一個執(zhí)行異常，則將該異常信息作為指定任務(wù)的執(zhí)行結(jié)果。測試用例如下：
@Test
public void test7() throws Exception {
ForkJoinPool pool=new ForkJoinPool();
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
return 1.2;
});
CompletableFuture<Double> cf2 = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
return 3.2;
});
//cf和cf2的異步任務(wù)都執(zhí)行完成后，會將其執(zhí)行結(jié)果作為方法入?yún)鬟f給cf3,且有返回值
CompletableFuture<Double> cf3=cf.thenCombine(cf2,(a,b)->{
System.out.println(Thread.currentThread()+" start job3,time->"+System.currentTimeMillis());
System.out.println("job3 param a->"+a+",b->"+b);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job3,time->"+System.currentTimeMillis());
return a+b;
});
//cf和cf2的異步任務(wù)都執(zhí)行完成后，會將其執(zhí)行結(jié)果作為方法入?yún)鬟f給cf3,無返回值
CompletableFuture cf4=cf.thenAcceptBoth(cf2,(a,b)->{
System.out.println(Thread.currentThread()+" start job4,time->"+System.currentTimeMillis());
System.out.println("job4 param a->"+a+",b->"+b);
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job4,time->"+System.currentTimeMillis());
});
//cf4和cf3都執(zhí)行完成后，執(zhí)行cf5，無入?yún)?，無返回值
CompletableFuture cf5=cf4.runAfterBoth(cf3,()->{
System.out.println(Thread.currentThread()+" start job5,time->"+System.currentTimeMillis());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
System.out.println("cf5 do something");
System.out.println(Thread.currentThread()+" exit job5,time->"+System.currentTimeMillis());
});
System.out.println("main thread start cf.get(),time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成
System.out.println("cf run result->"+cf.get());
System.out.println("main thread start cf5.get(),time->"+System.currentTimeMillis());
System.out.println("cf5 run result->"+cf5.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}其運行結(jié)果如下：
job1 和 job2幾乎同時運行，job2比job1先執(zhí)行完成，等job1退出后，job3和job4幾乎同時開始運行，job4先退出，等job3執(zhí)行完成，job5開始了，等job5執(zhí)行完成后，主線程退出。
2、applyToEither / acceptEither / runAfterEither
這三個方法都是將兩個CompletableFuture組合起來，只要其中一個執(zhí)行完了就會執(zhí)行某個任務(wù)，其區(qū)別在于applyToEither會將已經(jīng)執(zhí)行完成的任務(wù)的執(zhí)行結(jié)果作為方法入?yún)?，并有返回值；acceptEither同樣將已經(jīng)執(zhí)行完成的任務(wù)的執(zhí)行結(jié)果作為方法入?yún)?，但是沒有返回值；runAfterEither沒有方法入?yún)?，也沒有返回值。注意兩個任務(wù)中只要有一個執(zhí)行異常，則將該異常信息作為指定任務(wù)的執(zhí)行結(jié)果。測試用例如下：
@Test
public void test8() throws Exception {
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
return 1.2;
});
CompletableFuture<Double> cf2 = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
return 3.2;
});
//cf和cf2的異步任務(wù)都執(zhí)行完成后，會將其執(zhí)行結(jié)果作為方法入?yún)鬟f給cf3,且有返回值
CompletableFuture<Double> cf3=cf.applyToEither(cf2,(result)->{
System.out.println(Thread.currentThread()+" start job3,time->"+System.currentTimeMillis());
System.out.println("job3 param result->"+result);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job3,time->"+System.currentTimeMillis());
return result;
});
//cf和cf2的異步任務(wù)都執(zhí)行完成后，會將其執(zhí)行結(jié)果作為方法入?yún)鬟f給cf3,無返回值
CompletableFuture cf4=cf.acceptEither(cf2,(result)->{
System.out.println(Thread.currentThread()+" start job4,time->"+System.currentTimeMillis());
System.out.println("job4 param result->"+result);
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job4,time->"+System.currentTimeMillis());
});
//cf4和cf3都執(zhí)行完成后，執(zhí)行cf5，無入?yún)?，無返回值
CompletableFuture cf5=cf4.runAfterEither(cf3,()->{
System.out.println(Thread.currentThread()+" start job5,time->"+System.currentTimeMillis());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
System.out.println("cf5 do something");
System.out.println(Thread.currentThread()+" exit job5,time->"+System.currentTimeMillis());
});
System.out.println("main thread start cf.get(),time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成
System.out.println("cf run result->"+cf.get());
System.out.println("main thread start cf5.get(),time->"+System.currentTimeMillis());
System.out.println("cf5 run result->"+cf5.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}其運行結(jié)果如下：
job1 和job2 同時開始運行，job2先執(zhí)行完成，然后job4開始執(zhí)行，理論上job3和job4應(yīng)該同時開始運行，但是此時只有job4開始執(zhí)行了，job3是等待job1執(zhí)行完成后才開始執(zhí)行，job4先于job3執(zhí)行完成，然后job5開始執(zhí)行，等job5執(zhí)行完成后，主線程退出。上述差異且到下篇源碼分析時再做探討。
3、thenCompose
thenCompose方法會在某個任務(wù)執(zhí)行完成后，將該任務(wù)的執(zhí)行結(jié)果作為方法入?yún)⑷缓髨?zhí)行指定的方法，該方法會返回一個新的CompletableFuture實例，如果該CompletableFuture實例的result不為null，則返回一個基于該result的新的CompletableFuture實例；如果該CompletableFuture實例為null，則，然后執(zhí)行這個新任務(wù)，測試用例如下：
@Test
public void test9() throws Exception {
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
return 1.2;
});
CompletableFuture<String> cf2= cf.thenCompose((param)->{
System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
return CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job3,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job3,time->"+System.currentTimeMillis());
return "job3 test";
});
});
System.out.println("main thread start cf.get(),time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成
System.out.println("cf run result->"+cf.get());
System.out.println("main thread start cf2.get(),time->"+System.currentTimeMillis());
System.out.println("cf2 run result->"+cf2.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}其輸出如下：
job1執(zhí)行完成后job2開始執(zhí)行，等job2執(zhí)行完成后會把job3返回，然后執(zhí)行job3，等job3執(zhí)行完成后，主線程退出。
4、allOf / anyOf
allOf返回的CompletableFuture是多個任務(wù)都執(zhí)行完成后才會執(zhí)行，只有有一個任務(wù)執(zhí)行異常，則返回的CompletableFuture執(zhí)行g(shù)et方法時會拋出異常，如果都是正常執(zhí)行，則get返回null。
@Test
public void test11() throws Exception {
// 創(chuàng)建異步執(zhí)行任務(wù):
CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
return 1.2;
});
CompletableFuture<Double> cf2 = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
return 3.2;
});
CompletableFuture<Double> cf3 = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start job3,time->"+System.currentTimeMillis());
try {
Thread.sleep(1300);
} catch (InterruptedException e) {
}
// throw new RuntimeException("test");
System.out.println(Thread.currentThread()+" exit job3,time->"+System.currentTimeMillis());
return 2.2;
});
//allof等待所有任務(wù)執(zhí)行完成才執(zhí)行cf4，如果有一個任務(wù)異常終止，則cf4.get時會拋出異常，都是正常執(zhí)行，cf4.get返回null
//anyOf是只有一個任務(wù)執(zhí)行完成，無論是正常執(zhí)行或者執(zhí)行異常，都會執(zhí)行cf4，cf4.get的結(jié)果就是已執(zhí)行完成的任務(wù)的執(zhí)行結(jié)果
CompletableFuture cf4=CompletableFuture.allOf(cf,cf2,cf3).whenComplete((a,b)->{
if(b!=null){
System.out.println("error stack trace->");
b.printStackTrace();
}else{
System.out.println("run succ,result->"+a);
}
});
System.out.println("main thread start cf4.get(),time->"+System.currentTimeMillis());
//等待子任務(wù)執(zhí)行完成
System.out.println("cf4 run result->"+cf4.get());
System.out.println("main thread exit,time->"+System.currentTimeMillis());
}其輸出如下：
主線程等待最后一個job1執(zhí)行完成后退出。anyOf返回的CompletableFuture是多個任務(wù)只要其中一個執(zhí)行完成就會執(zhí)行，其get返回的是已經(jīng)執(zhí)行完成的任務(wù)的執(zhí)行結(jié)果，如果該任務(wù)執(zhí)行異常，則拋出異常。將上述測試用例中allOf改成anyOf后，其輸出如下：
總結(jié)
到此這篇關(guān)于Java8中CompletableFuture用法的文章就介紹到這了,更多相關(guān)Java8 CompletableFuture用法內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Java?spring?mvc請求詳情，mvc是spring源碼中的一個子模塊，下文關(guān)于spring?mvc請求的相關(guān)資料做簡單介紹，需要的小伙伴可以參考一下,希望對你有所幫助2022-03-03
這篇文章主要給大家介紹了關(guān)于Java中跳出多重循環(huán)嵌套的三種方法,很多復(fù)雜的運算以及邏輯可能用到嵌套循環(huán),但是如何跳出當前的多重嵌套循環(huán),很多同學(xué)特別是新手都操作的不是很清楚,需要的朋友可以參考下2023-07-07
這篇文章主要介紹了Spring中的InitializingBean接口的使用,需要的朋友可以參考下2018-04-04
這篇文章主要介紹了Java中的join線程、后臺線程、線程睡眠、線程讓步以及線程的優(yōu)先級，非常的詳細，希望能對大家有所幫助2014-10-10
springboot應(yīng)用訪問zookeeper的流程
這篇文章主要介紹了springboot應(yīng)用訪問zookeeper的流程，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-01-01
這篇文章主要為大家介紹了DispatcherServlet的處理器的適配和處理流程，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2023-06-06
Java中Dijkstra算法求解最短路徑的實現(xiàn)
Dijkstra算法是一種解決最短路徑問題的常用算法,本文主要介紹了Java中Dijkstra算法求解最短路徑的實現(xiàn),具有一定的參考價值,感興趣的可以了解一下2023-09-09
這篇文章主要介紹了SpringBoot框架打包體積簡化過程圖解,文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-05-05
SpringSecurity跨域請求偽造(CSRF)的防護實現(xiàn)
本文主要介紹了SpringSecurity跨域請求偽造(CSRF)的防護實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-07-07
idea 查看一個類的所有子類以及子類的子類并以層級關(guān)系顯示
這篇文章主要介紹了idea 查看一個類的所有子類以及子類的子類并以層級關(guān)系顯示，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-02-02