前面我們不止一次提到,用多線程優(yōu)化性能,其實(shí)不過就是將串行操作變成并行操作。如果仔細(xì)觀察,你還會發(fā)現(xiàn)在串行轉(zhuǎn)換成并行的過程中,一定會涉及到異步化,例如下面的示例代碼,現(xiàn)在是串行的,為了提升性能,我們得把它們并行化。
創(chuàng)新互聯(lián)專注于企業(yè)成都全網(wǎng)營銷推廣、網(wǎng)站重做改版、永康網(wǎng)站定制設(shè)計、自適應(yīng)品牌網(wǎng)站建設(shè)、成都h5網(wǎng)站建設(shè)、商城開發(fā)、集團(tuán)公司官網(wǎng)建設(shè)、成都外貿(mào)網(wǎng)站制作、高端網(wǎng)站制作、響應(yīng)式網(wǎng)頁設(shè)計等建站業(yè)務(wù),價格優(yōu)惠性價比高,為永康等各大城市提供網(wǎng)站開發(fā)制作服務(wù)。
// 以下兩個方法都是耗時操作 doBizA(); doBizB(); //創(chuàng)建兩個子線程去執(zhí)行就可以了,兩個操作已經(jīng)被異步化了。 new Thread(()->doBizA()) .start(); new Thread(()->doBizB()) .start();
異步化
,是并行方案得以實(shí)施的基礎(chǔ),更深入地講其實(shí)就是:利用多線程優(yōu)化性能這個核心方案得以實(shí)施的基礎(chǔ)
。Java 在 1.8 版本提供了 CompletableFuture 來支持異步編程。
CompletableFuture 的核心優(yōu)勢
為了領(lǐng)略 CompletableFuture 異步編程的優(yōu)勢,這里我們用 CompletableFuture 重新實(shí)現(xiàn)前面曾提及的燒水泡茶程序。首先還是需要先完成分工方案,在下面的程序中,我們分了 3 個任務(wù):任務(wù) 1 負(fù)責(zé)洗水壺、燒開水,任務(wù) 2 負(fù)責(zé)洗茶壺、洗茶杯和拿茶葉,任務(wù) 3 負(fù)責(zé)泡茶。其中任務(wù) 3 要等待任務(wù) 1 和任務(wù) 2 都完成后才能開始。這個分工如下圖所示。
燒水泡茶分工方案
// 任務(wù) 1:洗水壺 -> 燒開水 CompletableFuture<Void> f1 = CompletableFuture.runAsync(()->{ System.out.println("T1: 洗水壺..."); sleep(1, TimeUnit.SECONDS); System.out.println("T1: 燒開水..."); sleep(15, TimeUnit.SECONDS); }); // 任務(wù) 2:洗茶壺 -> 洗茶杯 -> 拿茶葉 CompletableFuture<String> f2 = CompletableFuture.supplyAsync(()->{ System.out.println("T2: 洗茶壺..."); sleep(1, TimeUnit.SECONDS); System.out.println("T2: 洗茶杯..."); sleep(2, TimeUnit.SECONDS); System.out.println("T2: 拿茶葉..."); sleep(1, TimeUnit.SECONDS); return " 龍井 "; }); // 任務(wù) 3:任務(wù) 1 和任務(wù) 2 完成后執(zhí)行:泡茶 CompletableFuture<String> f3 = f1.thenCombine(f2, (__, tf)->{ System.out.println("T1: 拿到茶葉:" + tf); System.out.println("T1: 泡茶..."); return " 上茶:" + tf; }); // 等待任務(wù) 3 執(zhí)行結(jié)果 System.out.println(f3.join()); void sleep(int t, TimeUnit u) { try { u.sleep(t); }catch(InterruptedException e){} } // 一次執(zhí)行結(jié)果: T1: 洗水壺... T2: 洗茶壺... T1: 燒開水... T2: 洗茶杯... T2: 拿茶葉... T1: 拿到茶葉: 龍井 T1: 泡茶... 上茶: 龍井
從整體上來看,我們會發(fā)現(xiàn)
f3 = f1.thenCombine(f2, ()->{})
能夠清晰地表述“任務(wù) 3 要等待任務(wù) 1 和任務(wù) 2 都完成后才能開始”;領(lǐng)略 CompletableFuture 異步編程的優(yōu)勢之后,下面我們詳細(xì)介紹 CompletableFuture 的使用。
創(chuàng)建 CompletableFuture 對象
創(chuàng)建 CompletableFuture 對象主要靠下面代碼中展示的這 4 個靜態(tài)方法,我們先看前兩個。在燒水泡茶的例子中,我們已經(jīng)使用了runAsync(Runnable runnable)
和 supplyAsync(Supplier<U> supplier)
,它們之間的區(qū)別是:Runnable 接口的 run() 方法沒有返回值,而 Supplier 接口的 get() 方法是有返回值的。
前兩個方法和后兩個方法的區(qū)別在于:后兩個方法可以指定線程池參數(shù)。
默認(rèn)情況下 CompletableFuture 會使用公共的 ForkJoinPool 線程池,這個線程池默認(rèn)創(chuàng)建的線程數(shù)是 CPU 的核數(shù)(也可以通過 JVM option:-Djava.util.concurrent.ForkJoinPool.common.parallelism
來設(shè)置 ForkJoinPool 線程池的線程數(shù))。如果所有 CompletableFuture 共享一個線程池,那么一旦有任務(wù)執(zhí)行一些很慢的 I/O 操作,就會導(dǎo)致線程池中所有線程都阻塞在 I/O 操作上,從而造成線程饑餓,進(jìn)而影響整個系統(tǒng)的性能。所以,強(qiáng)烈建議你要根據(jù)不同的業(yè)務(wù)類型創(chuàng)建不同的線程池,以避免互相干擾
。
// 使用默認(rèn)線程池 static CompletableFuture<Void> runAsync(Runnable runnable) static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) // 可以指定線程池 static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
創(chuàng)建完 CompletableFuture 對象之后,會自動地異步執(zhí)行 runnable.run() 方法或者 supplier.get() 方法,對于一個異步操作,你需要關(guān)注兩個問題:一個是異步操作什么時候結(jié)束,另一個是如何獲取異步操作的執(zhí)行結(jié)果。因?yàn)?CompletableFuture 類實(shí)現(xiàn)了 Future 接口,所以這兩個問題你都可以通過 Future 接口來解決。另外,CompletableFuture 類還實(shí)現(xiàn)了 CompletionStage 接口,這個接口內(nèi)容實(shí)在是太豐富了,在 1.8 版本里有 40 個方法,這些方法我們該如何理解呢?
理解 CompletionStage 接口
可以站在分工的角度類比一下工作流。任務(wù)是有時序關(guān)系的,比如有串行關(guān)系、并行關(guān)系、匯聚關(guān)系
等。這樣說可能有點(diǎn)抽象,這里還舉前面燒水泡茶的例子,其中洗水壺和燒開水就是串行關(guān)系,洗水壺、燒開水和洗茶壺、洗茶杯這兩組任務(wù)之間就是并行關(guān)系,而燒開水、拿茶葉和泡茶就是匯聚關(guān)系。
串行關(guān)系
并行關(guān)系
匯聚關(guān)系
CompletionStage 接口可以清晰地描述任務(wù)之間的這種時序關(guān)系,例如前面提到的f3 = f1.thenCombine(f2, ()->{})
描述的就是一種匯聚關(guān)系。燒水泡茶程序中的匯聚關(guān)系是一種 AND 聚合關(guān)系,這里的 AND 指的是所有依賴的任務(wù)(燒開水和拿茶葉)都完成后才開始執(zhí)行當(dāng)前任務(wù)(泡茶)。既然有 AND 聚合關(guān)系,那就一定還有 OR 聚合關(guān)系,所謂 OR 指的是依賴的任務(wù)只要有一個完成就可以執(zhí)行當(dāng)前任務(wù)。
最后就是異常,CompletionStage 接口也可以方便地描述異常處理。
下面我們就來一一介紹,CompletionStage 接口如何描述串行關(guān)系、AND 聚合關(guān)系、OR 聚合關(guān)系以及異常處理。
1. 描述串行關(guān)系
CompletionStage 接口里面描述串行關(guān)系,主要是 thenApply、thenAccept、thenRun 和 thenCompose 這四個系列的接口。
thenApply 系列函數(shù)里參數(shù) fn 的類型是接口 Function<T, R>,這個接口里與 CompletionStage 相關(guān)的方法是R apply(T t)
,這個方法既能接收參數(shù)也支持返回值,所以 thenApply 系列方法返回的是CompletionStage<R>
。
而 thenAccept 系列方法里參數(shù) consumer 的類型是接口Consumer<T>
,這個接口里與 CompletionStage 相關(guān)的方法是void accept(T t)
,這個方法雖然支持參數(shù),但卻不支持回值,所以 thenAccept 系列方法返回的是CompletionStage<Void>
thenRun 系列方法里 action 的參數(shù)是 Runnable,所以 action 既不能接收參數(shù)也不支持返回值,所以 thenRun 系列方法返回的也是CompletionStage<Void>
這些方法里面 Async 代表的是異步執(zhí)行 fn、consumer 或者 action。其中,需要你注意的是 thenCompose 系列方法,這個系列的方法會新創(chuàng)建出一個子流程,最終結(jié)果和 thenApply 系列是相同的。
CompletionStage<R> thenApply(fn); CompletionStage<R> thenApplyAsync(fn); CompletionStage<Void> thenAccept(consumer); CompletionStage<Void> thenAcceptAsync(consumer); CompletionStage<Void> thenRun(action); CompletionStage<Void> thenRunAsync(action); CompletionStage<R> thenCompose(fn); CompletionStage<R> thenComposeAsync(fn);
通過下面的示例代碼,你可以看一下 thenApply() 方法是如何使用的。首先通過 supplyAsync() 啟動一個異步流程,之后是兩個串行操作,整體看起來還是挺簡單的。不過,雖然這是一個異步流程,但任務(wù)①②③卻是串行執(zhí)行的,②依賴①的執(zhí)行結(jié)果,③依賴②的執(zhí)行結(jié)果。
CompletableFuture<String> f0 = CompletableFuture.supplyAsync( () -> "Hello World") //① .thenApply(s -> s + " QQ") //② .thenApply(String::toUpperCase);//③ System.out.println(f0.join()); // 輸出結(jié)果 HELLO WORLD QQ
2. 描述 AND 匯聚關(guān)系
CompletionStage 接口里面描述 AND 匯聚關(guān)系,主要是 thenCombine、thenAcceptBoth 和 runAfterBoth 系列的接口,這些接口的區(qū)別也是源自 fn、consumer、action 這三個核心參數(shù)不同。
CompletionStage<R> thenCombine(other, fn); CompletionStage<R> thenCombineAsync(other, fn); CompletionStage<Void> thenAcceptBoth(other, consumer); CompletionStage<Void> thenAcceptBothAsync(other, consumer); CompletionStage<Void> runAfterBoth(other, action); CompletionStage<Void> runAfterBothAsync(other, action);
3. 描述 OR 匯聚關(guān)系
CompletionStage 接口里面描述 OR 匯聚關(guān)系,主要是 applyToEither、acceptEither 和 runAfterEither 系列的接口,這些接口的區(qū)別也是源自 fn、consumer、action 這三個核心參數(shù)不同。
CompletionStage applyToEither(other, fn); CompletionStage applyToEitherAsync(other, fn); CompletionStage acceptEither(other, consumer); CompletionStage acceptEitherAsync(other, consumer); CompletionStage runAfterEither(other, action); CompletionStage runAfterEitherAsync(other, action);
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(()->{ int t = getRandom(5, 10); sleep(t, TimeUnit.SECONDS); return String.valueOf(t); }); CompletableFuture<String> f2 = CompletableFuture.supplyAsync(()->{ int t = getRandom(5, 10); sleep(t, TimeUnit.SECONDS); return String.valueOf(t); }); CompletableFuture<String> f3 = f1.applyToEither(f2,s -> s); System.out.println(f3.join());
4. 異常處理
雖然上面我們提到的 fn、consumer、action 它們的核心方法都不允許拋出可檢查異常,但是卻無法限制它們拋出運(yùn)行時異常
,例如下面的代碼,執(zhí)行
CompletableFuture<Integer> f0 = CompletableFuture. .supplyAsync(()->(7/0)) .thenApply(r->r*10); System.out.println(f0.join());
CompletionStage 接口給我們提供的方案非常簡單,比 try{}catch{}還要簡單,下面是相關(guān)的方法,使用這些方法進(jìn)行異常處理和串行操作是一樣的,都支持鏈?zhǔn)骄幊谭绞健?/p>
CompletionStage exceptionally(fn); CompletionStage<R> whenComplete(consumer); CompletionStage<R> whenCompleteAsync(consumer); CompletionStage<R> handle(fn); CompletionStage<R> handleAsync(fn);
下面的示例代碼展示了如何使用 exceptionally() 方法來處理異常,exceptionally() 的使用非常類似于 try{}catch{}中的 catch{},但是由于支持鏈?zhǔn)骄幊谭绞?,所以相對更簡單?/p>
whenComplete() 和 handle() 系列方法就類似于 try{}finally{}中的 finally{},無論是否發(fā)生異常都會執(zhí)行 whenComplete() 中的回調(diào)函數(shù) consumer 和 handle() 中的回調(diào)函數(shù) fn。
whenComplete() 和 handle() 的區(qū)別在于 whenComplete() 不支持返回結(jié)果,而 handle() 是支持返回結(jié)果的。
CompletableFuture<Integer> f0 = CompletableFuture .supplyAsync(()->7/0)) .thenApply(r->r*10) .exceptionally(e->0); System.out.println(f0.join());
總結(jié)
不過最近幾年,伴隨著 ReactiveX
的發(fā)展(Java 語言的實(shí)現(xiàn)版本是 RxJava),回調(diào)地獄已經(jīng)被完美解決了,Java 語言也開始官方支持異步編程:在 1.8 版本提供了 CompletableFuture,在 Java 9 版本則提供了更加完備的 Flow API,異步編程目前已經(jīng)完全工業(yè)化。
CompletableFuture 已經(jīng)能夠滿足簡單的異步編程需求,如果你對異步編程感興趣,可以重點(diǎn)關(guān)注 RxJava 這個項目,利用 RxJava,即便在 Java 1.6 版本也能享受異步編程的樂趣。
以上就是本文的全部內(nèi)容,希望對大家的學(xué)習(xí)有所幫助,也希望大家多多支持創(chuàng)新互聯(lián)。
當(dāng)前標(biāo)題:Java并發(fā)CompletableFuture異步編程的實(shí)現(xiàn)
網(wǎng)站地址:http://jinyejixie.com/article12/ipipdc.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供網(wǎng)站制作、外貿(mào)建站、小程序開發(fā)、企業(yè)網(wǎng)站制作、網(wǎng)站改版、做網(wǎng)站
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)