Java 8中CompletableFuture如何使用,相信很多沒有經(jīng)驗的人對此束手無策,為此本文總結(jié)了問題出現(xiàn)的原因和解決方法,通過這篇文章希望你能解決這個問題。
成都創(chuàng)新互聯(lián)是一家專注于網(wǎng)站建設(shè)、做網(wǎng)站與策劃設(shè)計,曹縣網(wǎng)站建設(shè)哪家好?成都創(chuàng)新互聯(lián)做網(wǎng)站,專注于網(wǎng)站建設(shè)10余年,網(wǎng)設(shè)計領(lǐng)域的專業(yè)建站公司;建站業(yè)務(wù)涵蓋:曹縣等地區(qū)。曹縣做網(wǎng)站價格咨詢:18982081108
最簡單的例子就是使用一個預(yù)定義的結(jié)果創(chuàng)建一個完成的CompletableFuture,通常我們會在計算的開始階段使用它。
static void completedFutureExample() { CompletableFuture cf = CompletableFuture.completedFuture("message"); assertTrue(cf.isDone()); assertEquals("message", cf.getNow(null)); }
getNow(null)方法在future完成的情況下會返回結(jié)果,就比如上面這個例子,否則返回null (傳入的參數(shù))。
這個例子創(chuàng)建一個一個異步執(zhí)行的階段:
static void runAsyncExample() { CompletableFuture cf = CompletableFuture.runAsync(() -> { assertTrue(Thread.currentThread().isDaemon()); randomSleep(); }); assertFalse(cf.isDone()); sleepEnough(); assertTrue(cf.isDone()); }
通過這個例子可以學(xué)到兩件事情:
CompletableFuture的方法如果以Async結(jié)尾,它會異步的執(zhí)行(沒有指定executor的情況下), 異步執(zhí)行通過ForkJoinPool實現(xiàn), 它使用守護線程去執(zhí)行任務(wù)。注意這是CompletableFuture的特性, 其它CompletionStage可以override這個默認(rèn)的行為。
參考閱讀:任務(wù)并行執(zhí)行神器:Fork&Join框架
下面這個例子使用前面 #1 的完成的CompletableFuture, #1返回結(jié)果為字符串message,然后應(yīng)用一個函數(shù)把它變成大寫字母。
static void thenApplyExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApply(s -> { assertFalse(Thread.currentThread().isDaemon()); return s.toUpperCase(); }); assertEquals("MESSAGE", cf.getNow(null)); }
注意thenApply方法名稱代表的行為。
then意味著這個階段的動作發(fā)生當(dāng)前的階段正常完成之后。本例中,當(dāng)前節(jié)點完成,返回字符串message。
Apply意味著返回的階段將會對結(jié)果前一階段的結(jié)果應(yīng)用一個函數(shù)。
函數(shù)的執(zhí)行會被阻塞,這意味著getNow()只有打斜操作被完成后才返回。
另外,關(guān)注公眾號Java技術(shù)棧,在后臺回復(fù):面試,可以獲取我整理的 Java 并發(fā)多線程系列面試題和答案,非常齊全。
通過調(diào)用異步方法(方法后邊加Async后綴),串聯(lián)起來的CompletableFuture可以異步地執(zhí)行(使用ForkJoinPool.commonPool())。
static void thenApplyAsyncExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> { assertTrue(Thread.currentThread().isDaemon()); randomSleep(); return s.toUpperCase(); }); assertNull(cf.getNow(null)); assertEquals("MESSAGE", cf.join()); }
異步方法一個非常有用的特性就是能夠提供一個Executor來異步地執(zhí)行CompletableFuture?!毒€程池全面解析》推薦看下。
這個例子演示了如何使用一個固定大小的線程池來應(yīng)用大寫函數(shù)。
static ExecutorService executor = Executors.newFixedThreadPool(3, new ThreadFactory() { int count = 1; @Override public Thread newThread(Runnable runnable) { return new Thread(runnable, "custom-executor-" + count++); } }); static void thenApplyAsyncWithExecutorExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> { assertTrue(Thread.currentThread().getName().startsWith("custom-executor-")); assertFalse(Thread.currentThread().isDaemon()); randomSleep(); return s.toUpperCase(); }, executor); assertNull(cf.getNow(null)); assertEquals("MESSAGE", cf.join()); }
如果下一階段接收了當(dāng)前階段的結(jié)果,但是在計算的時候不需要返回值(它的返回類型是void), 那么它可以不應(yīng)用一個函數(shù),而是一個消費者, 調(diào)用方法也變成了thenAccept:
static void thenAcceptExample() { StringBuilder result = new StringBuilder(); CompletableFuture.completedFuture("thenAccept message") .thenAccept(s -> result.append(s)); assertTrue("Result was empty", result.length() > 0); }
本例中消費者同步地執(zhí)行,所以我們不需要在CompletableFuture調(diào)用join方法。
同樣,可以使用thenAcceptAsync方法, 串聯(lián)的CompletableFuture可以異步地執(zhí)行。
static void thenAcceptAsyncExample() { StringBuilder result = new StringBuilder(); CompletableFuture cf = CompletableFuture.completedFuture("thenAcceptAsync message") .thenAcceptAsync(s -> result.append(s)); cf.join(); assertTrue("Result was empty", result.length() > 0); }
現(xiàn)在我們來看一下異步操作如何顯式地返回異常,用來指示計算失敗。我們簡化這個例子,操作處理一個字符串,把它轉(zhuǎn)換成答謝,我們模擬延遲一秒。
我們使用thenApplyAsync(Function, Executor)方法,第一個參數(shù)傳入大寫函數(shù), executor是一個delayed executor,在執(zhí)行前會延遲一秒。
static void completeExceptionallyExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)); CompletableFuture exceptionHandler = cf.handle((s, th) -> { return (th != null) ? "message upon cancel" : ""; }); cf.completeExceptionally(new RuntimeException("completed exceptionally")); assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally()); try { cf.join(); fail("Should have thrown an exception"); } catch(CompletionException ex) { // just for testing assertEquals("completed exceptionally", ex.getCause().getMessage()); } assertEquals("message upon cancel", exceptionHandler.join()); }
讓我們看一下細(xì)節(jié)。
首先我們創(chuàng)建了一個CompletableFuture, 完成后返回一個字符串message,接著我們調(diào)用thenApplyAsync方法,它返回一個CompletableFuture。這個方法在第一個函數(shù)完成后,異步地應(yīng)用轉(zhuǎn)大寫字母函數(shù)。
這個例子還演示了如何通過delayedExecutor(timeout, timeUnit)延遲執(zhí)行一個異步任務(wù)。
我們創(chuàng)建了一個分離的handler階段:exceptionHandler, 它處理異常異常,在異常情況下返回message upon cancel。
下一步我們顯式地用異常完成第二個階段。在階段上調(diào)用join方法,它會執(zhí)行大寫轉(zhuǎn)換,然后拋出CompletionException(正常的join會等待1秒,然后得到大寫的字符串。不過我們的例子還沒等它執(zhí)行就完成了異常), 然后它觸發(fā)了handler階段。
和完成異常類似,我們可以調(diào)用cancel(boolean mayInterruptIfRunning)取消計算。對于CompletableFuture類,布爾參數(shù)并沒有被使用,這是因為它并沒有使用中斷去取消操作,相反,cancel等價于completeExceptionally(new CancellationException())。
static void cancelExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)); CompletableFuture cfcf2 = cf.exceptionally(throwable -> "canceled message"); assertTrue("Was not canceled", cf.cancel(true)); assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally()); assertEquals("canceled message", cf2.join()); }
下面的例子創(chuàng)建了CompletableFuture, applyToEither處理兩個階段, 在其中之一上應(yīng)用函數(shù)(包保證哪一個被執(zhí)行)。本例中的兩個階段一個是應(yīng)用大寫轉(zhuǎn)換在原始的字符串上, 另一個階段是應(yīng)用小些轉(zhuǎn)換。
static void applyToEitherExample() { String original = "Message"; CompletableFuture cf1 = CompletableFuture.completedFuture(original) .thenApplyAsync(s -> delayedUpperCase(s)); CompletableFuture cf2 = cf1.applyToEither( CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)), s -> s + " from applyToEither"); assertTrue(cf2.join().endsWith(" from applyToEither")); }
和前一個例子很類似了,只不過我們調(diào)用的是消費者函數(shù) (Function變成Consumer):
static void acceptEitherExample() { String original = "Message"; StringBuilder result = new StringBuilder(); CompletableFuture cf = CompletableFuture.completedFuture(original) .thenApplyAsync(s -> delayedUpperCase(s)) .acceptEither(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)), s -> result.append(s).append("acceptEither")); cf.join(); assertTrue("Result was empty", result.toString().endsWith("acceptEither")); }
這個例子演示了依賴的CompletableFuture如果等待兩個階段完成后執(zhí)行了一個Runnable。
注意下面所有的階段都是同步執(zhí)行的,第一個階段執(zhí)行大寫轉(zhuǎn)換,第二個階段執(zhí)行小寫轉(zhuǎn)換。
static void runAfterBothExample() { String original = "Message"; StringBuilder result = new StringBuilder(); CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).runAfterBoth( CompletableFuture.completedFuture(original).thenApply(String::toLowerCase), () -> result.append("done")); assertTrue("Result was empty", result.length() > 0); }
上面的例子還可以通過BiConsumer來實現(xiàn):
static void thenAcceptBothExample() { String original = "Message"; StringBuilder result = new StringBuilder(); CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).thenAcceptBoth( CompletableFuture.completedFuture(original).thenApply(String::toLowerCase), (s1, s2) -> result.append(s1 + s2)); assertEquals("MESSAGEmessage", result.toString()); }
如果CompletableFuture依賴兩個前面階段的結(jié)果, 它復(fù)合兩個階段的結(jié)果再返回一個結(jié)果,我們就可以使用thenCombine()函數(shù)。整個流水線是同步的,所以getNow()會得到最終的結(jié)果,它把大寫和小寫字符串連接起來。
static void thenCombineExample() { String original = "Message"; CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s)) .thenCombine(CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s)), (s1, s2) -> s1 + s2); assertEquals("MESSAGEmessage", cf.getNow(null)); }
類似上面的例子,但是有一點不同:依賴的前兩個階段異步地執(zhí)行,所以thenCombine()也異步地執(zhí)行,即時它沒有Async后綴。
Javadoc中有注釋:
Actions supplied for dependent completions of non-async methods may be performed by the thread that completes the current CompletableFuture, or by any other caller of a completion method
所以我們需要join方法等待結(jié)果的完成。
static void thenCombineAsyncExample() { String original = "Message"; CompletableFuture cf = CompletableFuture.completedFuture(original) .thenApplyAsync(s -> delayedUpperCase(s)) .thenCombine(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)), (s1, s2) -> s1 + s2); assertEquals("MESSAGEmessage", cf.join()); }
我們可以使用thenCompose()完成上面兩個例子。這個方法等待第一個階段的完成(大寫轉(zhuǎn)換), 它的結(jié)果傳給一個指定的返回CompletableFuture函數(shù),它的結(jié)果就是返回的CompletableFuture的結(jié)果。
有點拗口,但是我們看例子來理解。函數(shù)需要一個大寫字符串做參數(shù),然后返回一個CompletableFuture, 這個CompletableFuture會轉(zhuǎn)換字符串變成小寫然后連接在大寫字符串的后面。
static void thenComposeExample() { String original = "Message"; CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s)) .thenCompose(upper -> CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s)) .thenApply(s -> upper + s)); assertEquals("MESSAGEmessage", cf.join()); }
下面的例子演示了當(dāng)任意一個CompletableFuture完成后, 創(chuàng)建一個完成的CompletableFuture.
待處理的階段首先創(chuàng)建, 每個階段都是轉(zhuǎn)換一個字符串為大寫。因為本例中這些階段都是同步地執(zhí)行(thenApply), 從anyOf中創(chuàng)建的CompletableFuture會立即完成,這樣所有的階段都已完成,我們使用whenComplete(BiConsumer<? super Object, ? super Throwable> action)處理完成的結(jié)果。
static void anyOfExample() { StringBuilder result = new StringBuilder(); List messages = Arrays.asList("a", "b", "c"); List<CompletableFuture> futures = messages.stream() .map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s))) .collect(Collectors.toList()); CompletableFuture.anyOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((res, th) -> { if(th == null) { assertTrue(isUpperCase((String) res)); result.append(res); } }); assertTrue("Result was empty", result.length() > 0); }
上一個例子是當(dāng)任意一個階段完成后接著處理,接下來的兩個例子演示當(dāng)所有的階段完成后才繼續(xù)處理, 同步地方式和異步地方式兩種。
static void allOfExample() { StringBuilder result = new StringBuilder(); List messages = Arrays.asList("a", "b", "c"); List<CompletableFuture> futures = messages.stream() .map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s))) .collect(Collectors.toList()); CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((v, th) -> { futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null)))); result.append("done"); }); assertTrue("Result was empty", result.length() > 0); }
使用thenApplyAsync()替換那些單個的CompletableFutures的方法,allOf()會在通用池中的線程中異步地執(zhí)行。所以我們需要調(diào)用join方法等待它完成。
static void allOfAsyncExample() { StringBuilder result = new StringBuilder(); List messages = Arrays.asList("a", "b", "c"); List<CompletableFuture> futures = messages.stream() .map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(s -> delayedUpperCase(s))) .collect(Collectors.toList()); CompletableFuture allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])) .whenComplete((v, th) -> { futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null)))); result.append("done"); }); allOf.join(); assertTrue("Result was empty", result.length() > 0); }
現(xiàn)在你已經(jīng)了解了CompletionStage 和 CompletableFuture 的一些函數(shù)的功能,下面的例子是一個實踐場景:
鴻蒙官方戰(zhàn)略合作共建——HarmonyOS技術(shù)社區(qū)
首先異步調(diào)用cars方法獲得Car的列表,它返回CompletionStage場景。cars消費一個遠(yuǎn)程的REST API。
然后我們復(fù)合一個CompletionStage填寫每個汽車的評分,通過rating(manufacturerId)返回一個CompletionStage, 它會異步地獲取汽車的評分(可能又是一個REST API調(diào)用)
當(dāng)所有的汽車填好評分后,我們結(jié)束這個列表,所以我們調(diào)用allOf得到最終的階段, 它在前面階段所有階段完成后才完成。
在最終的階段調(diào)用whenComplete(),我們打印出每個汽車和它的評分。
cars().thenCompose(cars -> { List<CompletionStage> updatedCars = cars.stream() .map(car -> rating(car.manufacturerId).thenApply(r -> { car.setRating(r); return car; })).collect(Collectors.toList()); CompletableFuture done = CompletableFuture .allOf(updatedCars.toArray(new CompletableFuture[updatedCars.size()])); return done.thenApply(v -> updatedCars.stream().map(CompletionStage::toCompletableFuture) .map(CompletableFuture::join).collect(Collectors.toList())); }).whenComplete((cars, th) -> { if (th == null) { cars.forEach(System.out::println); } else { throw new RuntimeException(th); } }).toCompletableFuture().join();
看完上述內(nèi)容,你們掌握J(rèn)ava 8中CompletableFuture如何使用的方法了嗎?如果還想學(xué)到更多技能或想了解更多相關(guān)內(nèi)容,歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道,感謝各位的閱讀!
當(dāng)前標(biāo)題:Java8中CompletableFuture如何使用
文章出自:http://aaarwkj.com/article26/pcdhcg.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供營銷型網(wǎng)站建設(shè)、企業(yè)建站、域名注冊、網(wǎng)頁設(shè)計公司、網(wǎng)站設(shè)計公司、網(wǎng)站導(dǎo)航
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)