异步编程之CompleTableFuture

所谓异步调用其实就是实现一个可无需等待被调用函数的返回值而让操作继续运行的方法。在 Java 语言中，简单的讲就是另启一个线程来完成调用中的部分计算，使调用继续运行或返回，而不需要等待计算结果。但调用者仍需要取线程的计算结果。

在并发时代，我们常用多线程来进行异步编程，但是多线程有一个缺点，就是不容易获取线程执行后的值，尽管用一些奇招可以获取，但是这已经远离了我们的初衷。java 1.5提供了一个future接口,可以对线程中的执行状态进行cancle，判断isCancelled，isDone。利用多线程来获取异步执行后的值，但是也有着诸多的缺点，比如

future没有异常处理程序，通常在future中异常后线程会一直阻塞在内部，造成异常抛不出，线程也退出不了

future中的get超时方法可以解决这个问题，但是若是方法还没有执行完，超时时间也过，又会出现超时异常的问题

1	V get(long timeout, TimeUnit unit)

无法进行链式调用，比如我完成了这个异步任务，接着我还想用这个调用的结果完成下一个任务的调用，future无法直接做到。
无法对多个调用结果进行并行计算，比如我只在其中一个执行成功后就退出，不再执行其他方法，又或者我想全部完成后在统一计算。
不能在无阻塞的情况下进行其他计算，只能在阻塞方法 get之后进行其他步骤的操作

想要用好future确实需要考虑特别多的情况。在java 8中，引入了CompletableFuture，它实现了future接口和CompletionStage接口，规定了异步计算契约的规范，并提供了他们的实现。

大概的归类一下：

以async结尾的方法都会以传入的线程池或者默认的forkjoin线程池执行任务，否则以刚才执行任务的线程继续执行
以then开头的方法会继续刚才的CompletableFuture执行，实现链式调用
中间带有accept的函数，传入类型是一个消费型接口Consumer，该接口接受上一个函数的值并消费,且不返回任何结果
中间带有apply的函数，传入一个函数型funtion接口，该接口接受上一个结果的值并且返回计算后的值，可以用get()接收
中间带有run的函数，传入一个Runnable类型，该接口不会接收值并且进行消费，不会返回值。
中间带有combine或者both由两个阶段触发，可以计算他们两个结果产生的影响,不同的是combine有返回值，而both没有返回值
中间带有Compose和带有apply的意思相同，但是不一样的是Compose方法会生成一个新的CompletableFuture对象继续执行，apply会继续连接刚才传入的CompletableFuture对象执行完该函数

以上的函数api相互兼容，比如thenApplyAsync 执行上一个函数，函数型接口，新线程异步执行

需要注意的是forkjoin的线程会以守护线程的方式执行，在虚拟机中main线程会在非守护线程执行完之后退出，守护线程没有执行完在控制台上看不到结果

我们先简单的生成一个CompletableFuture

1	CompletableFuture<String> completableFuture = new CompletableFuture<String>();

然后用get方法获取结果

1	String result = completableFuture.get();

当然我们的completableFuture什么都没有做，所以get方法会一直阻塞，我们需要在get前面执行complete方法

1	completableFuture.complete("Future's shuaizx");

或者我们也可以在get方法之前抛出一个异常

1	completableFuture.completeExceptionally(new RuntimeException());

一切都在我们的预料当中，但是别忘了我们需要的是异步执行，所以我们开启一个线程来执行这些方法

   CompletableFuture<String> completableFuture=new CompletableFuture<>();
   new Thread(new Runnable() {
    @Override
    public void run() {
        completableFuture.complete("shuaizx");
    }
}).start();


   System.out.println("start");
   System.out.println(completableFuture.get());
   System.out.println("end");

如果只是这样做的话和future没有任何的区别，在新的线程中抛出异常，在外部get还是获取不到结果,也不能中断执行。

public static String get() {
    if(true){
        throw new  RuntimeException();
    }
    return "shuaizx";
}

public static void main(String[] args) throws Exception {
    CompletableFuture<String> completableFuture = new CompletableFuture<>();
    new Thread(new Runnable() {
        @Override
        public void run() {
            String value = get();
            completableFuture.complete(value);
        }
    }).start();


    System.out.println("start");
    System.out.println(completableFuture.get());
    System.out.println("end");

我们需要对线程内的异常进行捕获，重新组装一个completeExceptionally抛出去，这至少可以让我们外部正确处理，不再让get阻塞

public static String get() {
    if(true){
        throw new  RuntimeException();
    }
    return "shuaizx";
}

public static void main(String[] args) throws Exception {
    CompletableFuture<String> completableFuture = new CompletableFuture<>();
    new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                String value = get();
                completableFuture.complete(value);
            }catch (Exception e){
                completableFuture.completeExceptionally(e);
            }
        }
    }).start();


    System.out.println("start");
    System.out.println(completableFuture.get());
    System.out.println("end");

是不是体验很不错了，至少我们不会因为异常不能抛出而感到苦恼了，但是在CompletableFuture内部已经给我们提供了静态方法，可以不用每次都写这种面条式的代码，下面介绍一下这几个方法：

CompletableFuture.runAsync(Runnable runnable);
CompletableFuture.runAsync(Runnable runnable, Executor executor);

CompletableFuture.supplyAsync(Supplier<U> supplier);
CompletableFuture.supplyAsync(Supplier<U> supplier, Executor executor)

在这里介绍一下runAsync函数参数为Runnable，函数无返回值。
Supplier为java 8新增的函数式接口，供给型接口，说白就是一个容器，可以存储执行后的值，在最后调用get方法获取值，如果没有传入Executor线程池参数，系统会给我们安排自定义的ForkJoinPool线程，帮我们生成线程，执行任务。

@FunctionalInterface
public interface Supplier<T> {

    /**
     * Gets a result.
     *
     * @return a result
     */
    T get();
}

举一个简单的例子

CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(()->"shuaizx");

System.out.println("start");
System.out.println(completableFuture.get());
System.out.println("end");

CompletableFuture.runAsync(Runnable runnable)方法接受Runnable参数
举例：

1
2
3

CompletableFuture.runAsync(() -> {
    System.out.println("start");
});

如此简单就可以实现一个自定义的异步执行计划

基本api讲解CompletableFuture

消费型接口

thenAccept传入Consumer型的接口，只消费不返回值，thenAcceptAsync传入自定义的线程池或者forkjoin线程池,以下消费型api遵循此规则，不过多介绍

thenAcceptBoth是用来合并结果当两个CompletionStage都正常执行的时候就会执行提供的action


CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    return 10;
});
System.out.println(future.thenAcceptBoth(CompletableFuture.supplyAsync(() -> {
    return 20;
}),(x,y) -> System.out.println(x+y)).get());

whenComplete消费型接口，传入数据和异常值，不管是否异常都会继续执行,只消费不返回

CompletableFuture<String> futureSatrt = CompletableFuture.supplyAsync(() -> {
    return "start";
}).whenComplete((t,e)->{
    System.out.println(t);
});

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

allOf方法会传入多个CompletableFuture，待全部执行后才返回结果，但是返回的却是一个void，需手动获取，所以不适合作为获取全部结果类型的api


CompletableFuture<String> futureSatrt = CompletableFuture.supplyAsync(() -> {
    System.out.println("start is run");
    return "start";
});
CompletableFuture<String> futureEnd = CompletableFuture.supplyAsync(() -> {
    System.out.println("end is run");
    return "end";
});

System.out.println( CompletableFuture.allOf(futureSatrt, futureEnd).get());

幸运的是java8 stream流的产生，可以让我们很简单的获取全部CompletableFuture的运行结果

但是get必须抛出一个受检查异常，我们是否可以不抛出受检异常而获取结果呢，CompletableFuture也提供了一个非受检异常让我们简单的获取结果
那就是join

CompletableFuture<String> futureSatrt = CompletableFuture.supplyAsync(() -> {
    System.out.println("start is run");
    return "start";
});
CompletableFuture<String> futureEnd = CompletableFuture.supplyAsync(() -> {
    System.out.println("end is run");
    return "end";
});

List<String> collect = Stream.of(futureSatrt, futureEnd).map(f -> {
        return f.join();
}).collect(Collectors.toList());

collect.forEach(s-> System.out.println(s));

函数型接口

thenApply传入一个Function型接口，返回处理后的值，thenApplyAsync需要传入自定义的线程池或者forkjoin线程池，其余不变，以下消费型api遵循此规则，不过多介绍

thenCompose接收一个新的CompletionStage型的函数型接口，返回处理后的值

CompletableFuture<String> futureSatrt = CompletableFuture.supplyAsync(() -> {
    return "start";
}).thenCompose(r->CompletableFuture.supplyAsync(()->{
    return r+"end";}));

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

thenCompose和thenApply的区别，thenApply处理的CompletableFuture和处理后的CompletableFuture是同一个，而thenCompose需要传入新的CompletableFuture，并且处理后的CompletableFuture是一个全新的CompletableFuture

thenCombine,你可以用thenCombine连接两个不同的CompletableFuture让他们连接起来,返回一个新的CompletableFuture，用get获取结果，可以看到组合两个CompletableFuture返回的新的CompletableFuture不是同一个，和thenAcceptBoth的区别是这个接口会返回值，而thenAcceptBoth只消费

CompletableFuture<String> futureSatrt = CompletableFuture.supplyAsync(() -> {
            return "start";
        });
        CompletableFuture<String> futureEnd = CompletableFuture.supplyAsync(() -> {
            return "end";
        });
        CompletableFuture<String> thenCombine = futureSatrt.thenCombine(futureEnd, (start, end) -> start + end);
        System.out.println(thenCombine.get());
        System.out.println(futureSatrt);
        System.out.println(futureEnd);
        System.out.println(thenCombine);

handle接口类型是函数型接口，传入数据和异常值，不管执行是否出现异常都会执行，和whenComplete相对应，返回处理后的值

CompletableFuture<String> futureSatrt = CompletableFuture.supplyAsync(() -> {
    return "start";
}).handle((t,e)->{
    System.out.println(e);
    return t+"end";
});

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

另一个同allof类型的api是anyof，当有一个CompletableFuture运行完时便退出

CompletableFuture<String> futureSatrt = CompletableFuture.supplyAsync(() -> {
    System.out.println("start is run");
    return "start";
});
CompletableFuture<String> futureEnd = CompletableFuture.supplyAsync(() -> {
    System.out.println("end is run");
    return "end";
});

Object o = CompletableFuture.anyOf(futureSatrt, futureEnd).get();

System.out.println(o);

运行的是anyof可以让我们获取到执行的值。

exceptionally可以让我们获取执行的异常并返回自定义的值

CompletableFuture<Object> end = CompletableFuture.supplyAsync(() -> {
    System.out.println("end is run");
    throw new RuntimeException();
}).exceptionally((e) -> "error");

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

以上只是java8版本的CompletableFuture基本api，在java9时又增加部分api，有兴趣的可以下去了解

CompletableFuture原理请移步

http://youngitman.tech/2019/02/13/completablefuture%E6%BA%90%E7%A0%81/

参考资料：