Future系列（CompletableFuture的使用）

提莫队长

发布于 2021-03-09 12:00:44

5920

发布于 2021-03-09 12:00:44

文章被收录于专栏：刘晓杰刘晓杰

1.创建CompletableFuture对象

CompletableFuture提供了四个静态方法用来创建CompletableFuture对象：

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)

Asynsc表示异步,而supplyAsync与runAsync不同在与前者异步返回一个结果,后者是void.第二个函数第二个参数表示是用我们自己创建的线程池,否则采用默认的ForkJoinPool.commonPool()作为它的线程池

        CompletableFuture<String> future = CompletableFuture
                .supplyAsync(() -> {
                    try {
                        Thread.sleep(1000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    return Thread.currentThread().getName();
                });
        System.out.println(future.get());//输出ForkJoinPool.commonPool-worker-1

2.计算结果

public T    get()
public T    get(long timeout, TimeUnit unit)
public T    getNow(T valueIfAbsent)
public T    join()

getNow有点特殊，如果结果已经计算完则返回结果或者抛出异常，否则返回给定的valueIfAbsent值。 join()与get()区别在于join()返回计算的结果或者抛出一个unchecked异常(CompletionException)，而get()返回一个具体的异常.

3.主动触发计算

public boolean complete(T  value)
public boolean completeExceptionally(Throwable ex)

上面方法表示当调用CompletableFuture.get()被阻塞的时候,那么这个方法就是结束阻塞,并且get()获取设置的value.

        CompletableFuture<String> future = CompletableFuture
                .supplyAsync(() -> {
                    try {
                        Thread.sleep(1000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    return Thread.currentThread().getName();
                });
        
        new Thread(new Runnable() {
            
            @Override
            public void run() {
                try {
                    Thread.sleep(1500);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                future.complete("hello");
            }
        }).start();
        
        System.out.println(future.get());

上面的例子，如果sleep小于1000，输出hello，大于1000ms，则输出ForkJoinPool.commonPool-worker-1

4.值转化

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

可以通过Function转化数据类型。直接上示例代码

        CompletableFuture<String> future = CompletableFuture
                .supplyAsync(() -> {
                    System.out.println("supplyAsync" + Thread.currentThread().getName());
                    return "step1";
                }).thenApply(new Function<String, String>() {
                    @Override
                    public String apply(String t) {
                        System.out.println("thenApply" + Thread.currentThread().getName());
                        return "step2";
                    }
                    
                }).thenApplyAsync(new Function<String, String>() {
                    @Override
                    public String apply(String t) {
                        System.out.println("thenApplyAsync" + Thread.currentThread().getName());
                        return "step3";
                    }
                    
                });
        System.out.println(future.get());

输出

supplyAsyncForkJoinPool.commonPool-worker-1
thenApplymain
thenApplyAsyncForkJoinPool.commonPool-worker-1
step3

thenApply会自动从默认的子线程中切换成主线程

5.消费处理结果

public CompletionStage<Void> thenAccept(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);

注意没有返回值，一般作为最后一步来做。示例代码

        CompletableFuture<Void> future = CompletableFuture
                .supplyAsync(() -> {
                    System.out.println("supplyAsync" + Thread.currentThread().getName());
                    return "step1";
                }).thenApply(new Function<String, String>() {
                    @Override
                    public String apply(String t) {
                        System.out.println("thenApply" + Thread.currentThread().getName());
                        return "step2";
                    }
                    
                }).thenAccept(new Consumer<String>() {
                    @Override
                    public void accept(String t) {
                        System.out.println("accept" + Thread.currentThread().getName());
                    }
                });
        System.out.println(future.get());

输出

supplyAsyncForkJoinPool.commonPool-worker-1
thenApplymain
acceptmain
null

如果换成 thenAcceptAsync，那么就会打印 ForkJoinPool.commonPool-worker-1

6.thenRun 方法

public CompletionStage<Void> thenRun(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);

它只是接着上面继续往下运行

        CompletableFuture<Void> future = CompletableFuture
                .supplyAsync(() -> {
                    System.out.println("supplyAsync" + Thread.currentThread().getName());
                    return "step1";
                }).thenApply(new Function<String, String>() {
                    @Override
                    public String apply(String t) {
                        System.out.println("thenApply" + Thread.currentThread().getName());
                        return "step2";
                    }
                    
                }).thenRunAsync(new Runnable() {
                    
                    @Override
                    public void run() {
                        System.out.println("thenRunAsync" + Thread.currentThread().getName());
                    }
                });
        System.out.println(future.get());

输出

supplyAsyncForkJoinPool.commonPool-worker-1
thenApplymain
thenRunAsyncForkJoinPool.commonPool-worker-1
null

如果改成thenRun，就是main

7.合并任务系列

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

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

当两个CompletionStage都执行完成后，把结果一块处理。thenAcceptBoth 系列没有返回值。还有后面的 runAfterBoth 系列的也是同样的功能

        CompletableFuture<String> future = CompletableFuture
                .supplyAsync(() -> {
                    System.out.println("supplyAsync" + Thread.currentThread().getName());
                    return "step1";
                }).thenApply(new Function<String, String>() {
                    @Override
                    public String apply(String t) {
                        System.out.println("thenApply" + Thread.currentThread().getName());
                        return "step2";
                    }
                    
                });
        
        CompletableFuture<String> future2 = CompletableFuture
                .supplyAsync(() -> {
                    return "future2-step1";
                }).thenApplyAsync(new Function<String, String>() {
                    @Override
                    public String apply(String t) {
                        return "future2-step2";
                    }
                    
                });
        
        future.thenCombine(future2, new BiFunction<String, String, String>() {

            @Override
            public String apply(String t, String u) {
                System.out.println("thenCombine" + Thread.currentThread().getName());
                return "thenCombine";
            }
            
        });
        
        System.out.println(future.get());

输出

supplyAsyncForkJoinPool.commonPool-worker-1
thenApplymain
thenCombinemain
step2

8.either系列

两个CompletionStage，谁执行返回的结果快，我就用那个CompletionStage的结果进行下一步的转化操作。

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

例子我就不写了

9.thenCompose

thenCompose 方法允许你对两个 CompletionStage 进行流水线操作，第一个操作完成时，将其结果作为参数传递给第二个操作。

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

直接上代码

        CompletableFuture<String> future = CompletableFuture.supplyAsync(
                () -> {
                    System.out.println("supplyAsync" + Thread.currentThread().getName());
                    return "step1";
                }).thenCompose(new Function<String, CompletionStage<String>>() {

            @Override
            public CompletionStage<String> apply(String t) {

                return CompletableFuture.supplyAsync(() -> {
                    return "future2-step1";
                });
            }
        });

        System.out.println(future.get());

输出

supplyAsyncForkJoinPool.commonPool-worker-1
future2-step1

感觉和前面的没啥区别呀

10.whenComplete

计算结果完成时的处理

public CompletableFuture<T>     whenComplete(BiConsumer<? super T,? super Throwable> action)
public CompletableFuture<T>     whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
public CompletableFuture<T>     whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)
public CompletableFuture<T>     exceptionally(Function<Throwable,? extends T> fn)

直接上代码

        CompletableFuture<String> future = CompletableFuture
                .supplyAsync(() -> {
                    double a = 1/0;
                    return "step1";
                }).whenComplete(new BiConsumer<String, Throwable>() {

                    @Override
                    public void accept(String t, Throwable throwable) {
                        System.out.println("t = " + t + ", throwable = " + throwable);
                    }
                    
                });

        System.out.println(future.get());

输出

t = null, throwable = java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.ArithmeticException: / by zero
    at java.util.concurrent.CompletableFuture.reportGet(Unknown Source)
    at java.util.concurrent.CompletableFuture.get(Unknown Source)
    at completableFuture.Main.main(Main.java:28)
Caused by: java.lang.ArithmeticException: / by zero
    at completableFuture.Main.lambda$0(Main.java:17)
    at java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source)
    at java.util.concurrent.CompletableFuture$AsyncSupply.exec(Unknown Source)
    at java.util.concurrent.ForkJoinTask.doExec(Unknown Source)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(Unknown Source)
    at java.util.concurrent.ForkJoinPool.runWorker(Unknown Source)
    at java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source)

可以看到其实就算accept，整个程序依然会崩溃

11.handle

功能和上面一点一样，计算结果完成时的处理

        CompletableFuture<String> future = CompletableFuture
                .supplyAsync(() -> {
                    double a = 1/0;
                    return "step1";
                }).handle(new BiFunction<String, Throwable, String>() {

                    @Override
                    public String apply(String t, Throwable throwable) {
                        System.out.println("t = " + t + ", throwable = " + throwable);
                        return "handle return";
                    }
                });

        System.out.println(future.get());

输出

t = null, throwable = java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
handle return

明显看到程序并没有崩溃，也就是说handle里面把崩溃捕捉了

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

如有侵权请联系 cloudcommunity@tencent.com 删除

http

unix

javascript

node.js

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度