Java異步編程工具 CompletableFuture

時間 2019-11-12

標籤 java 異步編程工具 completablefuture 欄目 Java 简体版

原文原文鏈接

CompletableFuture簡介

JDK 1.8 提供了CompletableFuture來支持異步編程，咱們能夠用CompletableFuture來很快的實現異步編程，CompletableFuture提供了串行，並行，匯聚3種模式提供給咱們使用java

使用方法

建立

public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor)

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

咱們能夠經過上面4個API來建立CompletableFuture對象，API分爲兩大類，一類是無返回值的runAsync，一類是有返回值的supplyAsync,每一個大類下面有分紅了兩個小類，一種是使用默認的Fork/Join線程池，一種是使用本身定義的線程池編程

串行調用

CompletableFuture<U> thenApply(Function<? super T,? extends U> fn);
CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn);
CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

CompletableFuture<Void> thenAccept(Consumer<? super T> action);
CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action);
CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor)
                                                
CompletableFuture<Void> thenRun(Runnable action);
CompletableFuture<Void> thenRunAsync(Runnable action);
CompletableFuture<Void> thenRunAsync(Runnable action,Executor executor)

CompletableFuture<R> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn);
CompletableFuture<R> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn);
CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,Executor executor)

串行調用提供了上述的8個API，分爲4大類，逐一介紹一下：app

thenApply系列須要傳入一個Function<? super T,? extends U>參數，T表明入參，U表明出參，因此thenApply系列能夠傳入參數也能夠返回結果異步
thenAccept系列會傳入一個Consumer<? super T>，T是入參，因此thenAccept能夠傳入參數，可是不會返回結果異步編程
thenRun系列須要傳入一個Runnale，因此這個系列既不能有入參也不會有結果線程
thenCompose系列和thenApply系列結果相同，可是須要開啓一個子任務去執行，從傳入的參數也能夠看出，參數一接受一個CompletionStage的Function，CompletionStage就是CompletableFuture實現的接口，具體到實現類就是在接收一個CompletableFuture對象code

每一個大類都有* 和 *Async兩種API，區別就在於帶Async的任務會在丟給Fork/Join線程池執行，不帶Async就直接由前面任務的線程來執行，帶Async還能夠本身指定線程池對象

並行

並行比較好理解，就是同時建立多個CompletableFuture，讓任務去並行執行接口

匯聚

匯聚又分紅兩種，一種的AND匯聚，一個是OR匯聚，簡單的說就是AND匯聚須要匯聚的任務都完成才能夠執行匯聚以後的方法，而OR匯聚只要其中一個任務完成就能夠往下執行了，匯聚API能夠將並行執行的CompletableFuture匯聚成一個CompletableFutureget

AND匯聚

CompletableFuture<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn);
CompletableFuture<V> thenCombineAsync( CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn, Executor executor)

CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action));
CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action);
CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action, Executor executor)      

CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action);
CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action);
CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor)

AND匯聚提供了3類API，API和串行的API功能相似，thenCombine提供了有入參和出參的能力，thenAcceptBoth只提供了入參的能力，沒有返回值，runAfterBoth既沒有入參也沒有出參

OR匯聚

CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)
CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn)
CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn,Executor executor)

CompletableFuture<Void> acceptEither( CompletionStage<? extends T> other, Consumer<? super T> action)
CompletableFuture<Void> acceptEitherAsync( CompletionStage<? extends T> other, Consumer<? super T> action)
CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action,Executor executor)

CompletableFuture<Void> runAfterEither(CompletionStage<?> other,Runnable action)
CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action)
CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action,Executor executor)

OR匯聚也和AND匯聚相似，提供了3類API，功能方法也能夠參考AND匯聚執行的方法

異常處理

CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn);
        
CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action));
CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action);
CompletableFuture<T> whenCompleteAsync( BiConsumer<? super T, ? super Throwable> action, Executor executor)
        
CompletableFuture<U> handle( BiFunction<? super T, Throwable, ? extends U> fn);
CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn);
CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)

異常處理提供了3類API：

exceptionally相似try-catch，若是有異常咱們能夠經過入參獲取到異常
whenComplete能夠獲取CompletableFuture的結果，而且能夠經過第二個參數異常（若是有的話），而且這個異常在主線程也能夠捕獲
handle和whenComplete相似，可是他還能夠返回一個結果，和whenComplete不一樣的是，裏面的異常在主線程不能捕獲

例子

package com.demo;

import java.util.concurrent.CompletableFuture;

public class Test {
    
    public static void main(String[] args){

        CompletableFuture<String> f1 = CompletableFuture.runAsync(()->{
            System.out.println("T1：start");
            sleep(1000);
            System.out.println("T1: doing sth");
            sleep(5000);
        }).thenRunAsync(()-> System.out.println("T1 : next task")).thenApply((__)-> {
            System.out.println("T1 task end");
            return " T1 result";
        });

        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(()->{
            System.out.println("T2: start");
            sleep(1000);
            System.out.println("T2: doing sth");
            sleep(2000);
            return " T2：result";
        }).thenApply(s-> s+ "!!!").thenCompose(s -> CompletableFuture.supplyAsync(s::toUpperCase));

        CompletableFuture<String> f3 = f1.thenCombine(f2,(r1,r2)->{
            System.out.println("T1 result :" + r1);
            System.out.println("T2 result：" + r2);
            return "t1 t2 end";
        });


        System.out.println(f3.join());

        System.out.println("--------------");

        /**
         * exceptionally處理異常狀況
         * result:
         * java.lang.ArithmeticException: / by zero
         * 0
         */
        CompletableFuture<Integer> f4 = CompletableFuture.supplyAsync(()->1/0)
                .thenApply(i->i*i)
                .exceptionally((throwable -> {
                    System.out.println(throwable.getMessage());
                    return 0;
                }));
        System.out.println(f4.join());
        System.out.println("--------------");
        /**
         * whenComplete處理異常狀況
         * result : null, error : java.lang.ArithmeticException: / by zero
         * enter exception block
         *
         * Process finished with exit code 0
         *
         */
        try {
            CompletableFuture<Integer> f5 = CompletableFuture.supplyAsync(()->1/0)
                    .thenApply(i->i*i)
                    .whenComplete((i,t)-> {
                        System.out.println("result : " +i+ ", error : " + t.getMessage());
                    });

            System.out.println(f5.join());
        }catch (Exception e){
            System.out.println("enter exception block");
        }
        System.out.println("--------------");
        /**
         * handle處理異常狀況
         * result : null, error : java.lang.ArithmeticException: / by zero
         * 0
         *
         * Process finished with exit code 0
         *
         */
        try {
            CompletableFuture<Integer> f6 = CompletableFuture.supplyAsync(()->1/0)
                    .thenApply(i->i*i)
                    .handle((i,t)-> {
                        System.out.println("result : " +i+ ", error : " + t.getMessage());
                        return 0;
                    });

            System.out.println(f6.join());
        }catch (Exception e){
            System.out.println("enter exception block");
        }
    }

    private static void sleep(long time){
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

一次結果

T1：start
T2: start
T1: doing sth
T2: doing sth
T1 : next task
T1 task end
T1 result : T1 result
T2 result： T2：RESULT!!!
t1 t2 end
--------------
java.lang.ArithmeticException: / by zero
0
--------------
result : null, error : java.lang.ArithmeticException: / by zero
enter exception block
--------------
result : null, error : java.lang.ArithmeticException: / by zero
0

Process finished with exit code 0