阅读 218

CompletableFuture API(全)

原文:https://blog.csdn.net/u012129558/article/details/78962759

CompletableFuture类实现了CompletionStage和Future接口。Future是Java 5添加的类,用来描述一个异步计算的结果,但是获取一个结果时方法较少,要么通过轮询isDone,确认完成后,调用get()获取值,要么调用get()设置一个超时时间。但是这个get()方法会阻塞住调用线程,这种阻塞的方式显然和我们的异步编程的初衷相违背。

为了解决这个问题,JDK吸收了guava的设计思想,加入了Future的诸多扩展功能形成了CompletableFuture。

CompletionStage是一个接口,从命名上看得知是一个完成的阶段,它里面的方法也标明是在某个运行阶段得到了结果之后要做的事情。

  1. 进行变换
    public  CompletionStage thenApply(Functionsuper T,? extends U> fn);
    public  CompletionStage thenApplyAsync(Functionsuper T,? extends U> fn);
    public  CompletionStage thenApplyAsync(Functionsuper T,? extends U> fn,Executor executor);

     

  2. 首先说明一下已Async结尾的方法都是可以异步执行的,如果指定了线程池,会在指定的线程池中执行,如果没有指定,默认会在ForkJoinPool.commonPool()中执行,下文中将会有好多类似的,都不详细解释了。关键的入参只有一个Function,它是函数式接口,所以使用Lambda表示起来会更加优雅。它的入参是上一个阶段计算后的结果,返回值是经过转化后结果。
    例如:

      @Test
        public void thenApply() {
            String result = CompletableFuture.supplyAsync(() -> "hello").thenApply(s -> s + " world").join();
            System.out.println(result);
        }

    结果为:

    hello world
    
    1. 进行消耗
    public CompletionStage thenAccept(Consumersuper T> action);
    public CompletionStage thenAcceptAsync(Consumersuper T> action);
    public CompletionStage thenAcceptAsync(Consumersuper T> action,Executor executor);

    thenRun它的入参是一个Runnable的实例,表示当得到上一步的结果时的操作。
    例如:

     @Test
        public void thenRun(){
            CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "hello";
            }).thenRun(() -> System.out.println("hello world"));
            while (true){}
        }

    结果为:

    hello world

 

4.结合两个CompletionStage的结果,进行转化后返回

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

它需要原来的处理返回值,并且other代表的CompletionStage也要返回值之后,利用这两个返回值,进行转换后返回指定类型的值。
例如:

  @Test
    public void thenCombine() {
        String result = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "world";
        }), (s1, s2) -> s1 + " " + s2).join();
        System.out.println(result);
    }

 

结果为:

hello world



5.结合两个CompletionStage的结果,进行消耗

public  CompletionStage thenAcceptBoth(CompletionStageextends U> other,BiConsumersuper T, ? super U> action);
public  CompletionStage thenAcceptBothAsync(CompletionStageextends U> other,BiConsumersuper T, ? super U> action);
public  CompletionStage thenAcceptBothAsync(CompletionStageextends U> other,BiConsumersuper T, ? super U> action,     Executor executor);

它需要原来的处理返回值,并且other代表的CompletionStage也要返回值之后,利用这两个返回值,进行消耗。
例如:

 @Test
    public void thenAcceptBoth() {
        CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        }).thenAcceptBoth(CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "world";
        }), (s1, s2) -> System.out.println(s1 + " " + s2));
        while (true){}
    }

结果为:

hello world


6在两个CompletionStage都运行完执行。

public CompletionStage runAfterBoth(CompletionStage other,Runnable action);
public CompletionStage runAfterBothAsync(CompletionStage other,Runnable action);
public CompletionStage runAfterBothAsync(CompletionStage other,Runnable action,Executor executor);

不关心这两个CompletionStage的结果,只关心这两个CompletionStage执行完毕,之后在进行操作(Runnable)。
例如:

@Test
    public void runAfterBoth(){
        CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "s1";
        }).runAfterBothAsync(CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "s2";
        }), () -> System.out.println("hello world"));
        while (true){}
    }

结果为

   hello world



7.两个CompletionStage,谁计算的快,我就用那个CompletionStage的结果进行下一步的转化操作。
public  CompletionStage applyToEither(CompletionStageextends T> other,Functionsuper T, U> fn);
public  CompletionStage applyToEitherAsync(CompletionStageextends T> other,Functionsuper T, U> fn);
public  CompletionStage applyToEitherAsync(CompletionStageextends T> other,Functionsuper T, U> fn,Executor executor);

我们现实开发场景中,总会碰到有两种渠道完成同一个事情,所以就可以调用这个方法,找一个最快的结果进行处理。
例如:

@Test
    public void applyToEither() {
        String result = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "s1";
        }).applyToEither(CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello world";
        }), s -> s).join();
        System.out.println(result);
    }

结果为:

hello world


8.两个CompletionStage,谁计算的快,我就用那个CompletionStage的结果进行下一步的消耗操作。

public CompletionStage acceptEither(CompletionStageextends T> other,Consumersuper T> action);
public CompletionStage acceptEitherAsync(CompletionStageextends T> other,Consumersuper T> action);
public CompletionStage acceptEitherAsync(CompletionStageextends T> other,Consumersuper T> action,Executor executor);

 

 @Test
    public void acceptEither() {
        CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "s1";
        }).acceptEither(CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello world";
        }), System.out::println);
        while (true){}
    }

 

结果为:

hello world


9两个CompletionStage,任何一个完成了都会执行下一步的操作(Runnable)。

public CompletionStage runAfterEither(CompletionStage other,Runnable action);
public CompletionStage runAfterEitherAsync(CompletionStage other,Runnable action);
public CompletionStage runAfterEitherAsync(CompletionStage other,Runnable action,Executor executor);
@Test
    public void runAfterEither() {
        CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "s1";
        }).runAfterEither(CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "s2";
        }), () -> System.out.println("hello world"));
        while (true) {
        }
    }

结果为:

hello world


10 当运行时出现了异常,可以通过exceptionally进行补偿。

public CompletionStage exceptionally(Functionextends T> fn);
@Test
    public void exceptionally() {
        String result = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (1 == 1) {
                throw new RuntimeException("测试一下异常情况");
            }
            return "s1";
        }).exceptionally(e -> {
            System.out.println(e.getMessage());
            return "hello world";
        }).join();
        System.out.println(result);
    }

结果为:

  1.  java.lang.RuntimeException: 测试一下异常情况
  2.  hello world

11当运行完成时,对结果的记录。这里的完成时有两种情况,一种是正常执行,返回值。另外一种是遇到异常抛出造成程序的中断。

这里为什么要说成记录,因为这几个方法都会返回CompletableFuture,当Action执行完毕后它的结果返回原始的CompletableFuture的计算结果或者返回异常。所以不会对结果产生任何的作用

public CompletionStage whenComplete(BiConsumersuper T, ? super Throwable> action);
public CompletionStage whenCompleteAsync(BiConsumersuper T, ? super Throwable> action);
public CompletionStage whenCompleteAsync(BiConsumersuper T, ? super Throwable> action,Executor executor);
@Test
    public void whenComplete() {
        String result = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (1 == 1) {
                throw new RuntimeException("测试一下异常情况");
            }
            return "s1";
        }).whenComplete((s, t) -> {
            System.out.println(s);
            System.out.println(t.getMessage());
        }).exceptionally(e -> {
            System.out.println(e.getMessage());
            return "hello world";
        }).join();
        System.out.println(result);
    }

结果为:

  1.  null
  2.  java.lang.RuntimeException: 测试一下异常情况
  3.  java.lang.RuntimeException: 测试一下异常情况
  4.  hello world

 

这里也可以看出,如果使用了exceptionally,就会对最终的结果产生影响,它没有口子返回如果没有异常时的正确的值,这也就引出下面我们要介绍的handle。

 

12 运行完成时,对结果的处理。这里的完成时有两种情况,一种是正常执行,返回值。另外一种是遇到异常抛出造成程序的中断。

public  CompletionStage handle(BiFunctionsuper T, Throwable, ? extends U> fn);
public  CompletionStage handleAsync(BiFunctionsuper T, Throwable, ? extends U> fn);
public  CompletionStage handleAsync(BiFunctionsuper T, Throwable, ? extends U> fn,Executor executor);

例如:
出现异常时

@Test
    public void handle() {
        String result = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //出现异常
            if (1 == 1) {
                throw new RuntimeException("测试一下异常情况");
            }
            return "s1";
        }).handle((s, t) -> {
            if (t != null) {
                return "hello world";
            }
            return s;
        }).join();
        System.out.println(result);
    }

结果为:

hello world

未出现异常时
 @Test
    public void handle() {
        String result = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "s1";
        }).handle((s, t) -> {
            if (t != null) {
                return "hello world";
            }
            return s;
        }).join();
        System.out.println(result);
    }

结果为:

s1

上面就是CompletionStage接口中方法的使用实例,CompletableFuture同样也同样实现了Future,所以也同样可以使用get进行阻塞获取值,总的来说,CompletableFuture使用起来还是比较爽的,看起来也比较优雅一点。

作者:数齐
链接:https://www.jianshu.com/p/6f3ee90ab7d3
来源:简书
简书著作权归作者所有,任何形式的转载都请联系作者获得授权并注明出处。

 

原文:https://www.cnblogs.com/lshan/p/14442908.html

文章分类
代码人生
版权声明:本站是系统测试站点,无实际运营。本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容, 请发送邮件至 XXXXXXo@163.com 举报,一经查实,本站将立刻删除。
相关推荐