java8 CompletableFuture

阅读数:137 评论数:0

分类

python/Java

正文

在JAVA5已经提供了Future和Callable的实现，可以用于阻塞式获取结果，如果想要异步获取结果，通常都会以轮询的方式去获取结果，如下：

//定义一个异步任务
Future<String> future = executor.submit(()->{
       Thread.sleep(2000);
       return "hello world";
});
//轮询获取结果
while (true){
    if(future.isDone()) {
         System.out.println(future.get());
         break;
     }
 }

这种方式会耗费CPU资源，而且也不能及时得到计算结果，在JAVA8中CompletableFuture提供了更强大的Future扩展功能。

创建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)

supplyAsync与runAsync不同在于前者异步返回一个结果，后者是void，第二个参数表示使用我们自己创建的线程，否则采用默认的ForkJoinPool.commonPool()作为它的线程池，其中Supplier是一个函数式接口（所以可以使用lamda写法），传入0个参数，返回一个结果。

CompletableFuture<String> future = CompletableFuture.supplyAsync(()->{
            return "hello world";
  });
System.out.println(future.get());  //阻塞的获取结果  ''helllo world"

主动计算

以下方法用于获取结果

//同步获取结果
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返回一个具体的异常。

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

主动调用上面的方法，可以结束获取结果时的阻塞。

计算结果完成时的处理

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)
// 处理task的异常
public CompletableFuture<T>     exceptionally(Function<Throwable,? extends T> fn)

以上方法是当计算结束的时候触发，BiConsumer有两个入参，分别代表计算返回值，另外一个是异常。无返回值的方法不以Async结尾。

// 两个参数，task结果和task异常
public <U> CompletableFuture<U>     handle(BiFunction<? super T,Throwable,? extends U> fn)
public <U> CompletableFuture<U>     handleAsync(BiFunction<? super T,Throwable,? extends U> fn)
public <U> CompletableFuture<U>     handleAsync(BiFunction<? super T,Throwable,? extends U> fn, Executor executor)

与whenComplete()不同的是这个函数返回CompletableFuture并不是原始的CompletableFuture返回的值，而是BiFunction返回的值。

CompletableFuture的组合

1、thenApply

当计算结算完成之后，后面可以继续一系列的thenApply，来完成值的转化。

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)

它们与handle方法的区别在于handle方法会处理正常计算值和异常，因此它可以屏蔽异常，避免异常继续抛出。而thenApply方法只是用来处理正常值，因此一旦有异常就会抛出。

2、thenAccept

只对CompletableFuture的结果进行消费，不返回值。

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

3、thenAcceptBoth

这个方法用来等待另一个CompletableFuture的结果。

CompletableFuture<String> future = CompletableFuture.supplyAsync(()->{
            return "hello world";
        });
CompletableFuture future5 =  future.thenAcceptBoth(CompletableFuture.completedFuture("compose"),
                (x, y) -> System.out.println(x+y));//hello world compose

4、either和all

（1）applyToEither是当任意一个CompletableFuture计算完成的时候就会执行。

Random rand = new Random();
        CompletableFuture<Integer> future9 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000 + rand.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 100;
        });
        CompletableFuture<Integer> future10 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000 + rand.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 200;
        });
        //两个中任意一个计算完成,那么触发Runnable的执行
        CompletableFuture<String> f =  future10.applyToEither(future9,i -> i.toString());
        //两个都计算完成,那么触发Runnable的执行
        CompletableFuture f1 = future10.acceptEither(future9,(e)->{
            System.out.println(e);
        });
        System.out.println(f.get());

（2）allOf方法是当所有的CompletableFuture都执行完成后执行计算

（3）anyOf是当任意一个CompletableFuture执行完成后会执行计算

秒吧学习