Java 8 - CompletableFuture

Future 是在Java 5引入的，CompletableFuture 是在Java 8引入的，提供了更加强大的功能。两者都是用来实现异步的，避免阻塞主线程.

Future

Future 提供了get 方法来返回异步任务的执行结果，当调用get方法会阻塞直至任务结束返回结果。

public class TestFuture {

    public static void main(String[] args) {
        ExecutorService exec = Executors.newSingleThreadExecutor();
        Future<Integer> f = exec.submit(new MyCallable());

        System.out.println(f.isDone()); // false

        try {
            System.out.println(f.get()); // 1，等待直到Callable完成
            System.out.println(f.isDone()); // true
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        } finally {
            exec.shutdown();
        }
    }
}

class MyCallable implements Callable<Integer> {

    @Override
    public Integer call() throws Exception {
        Thread.sleep(1000);
        return 1;
    }

}

CompletableFuture

CompletableFuture 实现了Future 和CompletionStage 接口，实现异步任务的链式处理，支持多个任务的并发执行、顺序执行，对任务的控制更加精细。

runAsync()

异步执行一个 Runnable 实例，异步任务没有返回值，该方法返回CompletableFuture实例。

CompletableFuture future = CompletableFuture.runAsync(() -> {
    try {
        System.out.println("Running asynchronous task in parallel");
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException ex) {
        throw new IllegalStateException(ex);
    }
});

supplyAsync()

异步执行一个Supplier，异步任务具有返回值，该方法返回CompletableFuture实例。

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    }
    return "This is the result of the asynchronous computation";
});

thenApply()

执行一个 Function，这个函数的执行还是在CompletableFuture.supplyAsync 开启的线程中执行，Function有返回值所以可以链式的连接多个thenApply方法。

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "Feng");
CompletableFuture<String> cf2 = cf.thenApply(name -> "Hello " + name).thenApply(greeting -> greeting + ", Welcome to SH!");
System.out.println(cf.get()); // Feng
System.out.println(cf2.get()); // Hello Feng, Welcome to SH!

thenAccept()

执行一个 Consumer，此时执行异步任务的线程没有返回值。通常该方法应该在操作链的最后。

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "Feng");
CompletableFuture<String> cf2 = cf.thenApply(name -> "Hello " + name).thenApply(greeting -> greeting + ", Welcome to SH!");
cf2.thenAccept(v-> System.out.println(v)); // Hello Feng, Welcome to SH!

thenRun()

执行一个Runnable，和thenAccept一样，此时执行异步任务的线程没有返回值，但不同的是此时也不接收任何参数的传入。

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "Feng");
cf.thenRun(()-> System.out.println("then run.."));

thenCompose()

thenCompose方法接受一个返回CompletableFuture的Function做为参数，和thenApply不同的是：
thenApply接受是一个同步方法，而thenCompose接受的是异步方法。

public class CF2 {

    private static CompletableFuture<String> sendMailAsync(String input) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return input + " | " + "2. Send mail to Administrator.";
        });
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "1. Write mail.");
        CompletableFuture<String> cf2 = cf.thenCompose(CF2::sendMailAsync);
        System.out.println(cf2.get()); // 1. Write mail. | 2. Send mail to Administrator.
    }

}

thenCombine()

thenCombine接受一个 BiFunction，来处理两个CompletableFuture的结果。

public class CF3 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "1. A");
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> "2. B");

        CompletableFuture<String> cf3 = cf1.thenCombine(cf2, (a, b) -> a + " - " + b);
        System.out.println(cf3.get());
    }
}

thenAcceptBoth()

thenAcceptBoth 接受一个 BiConsumer，来处理两个CompletableFuture的结果。

public class CF4 {
    public static void main(String[] args) {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "1. A");
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> "2. B");
        cf1.thenAcceptBoth(cf2, (a, b) -> System.out.println(a + b));
    }
}

方法	描述
<U, R> CompletionStage<R> thenCombine(CompletionStage<U> other, BiFunction<T, U, R> action)	Combines the result of this and other in one, using a BiFunction
<U> CompletionStage<Void> thenAcceptBoth(CompletionStage<U> other, BiConsumer<T, U> action)	Consumes the result of this and other, using a BiConsumer
<U> CompletionStage<Void> runAfterBoth(CompletionStage<U> other, BiConsumer<T, U> action)	Triggers the execution of a Runnable on the completion of this and other

CompletableFuture.allOf()

返回新的CompletableFuture实例，当形参中指定的CompletableFuture都完成了，该方法返回的新CompletableFuture也就完成了，如果要等待形参中指定的CompletableFuture都完成可以使用CompletableFuture.allOf(cf1, cf2, cf3).join()。

CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> 1);
CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> 2);
CompletableFuture cf3 = CompletableFuture.supplyAsync(() -> 3);

CompletableFuture.allOf(cf1, cf2, cf3).join(); // All CompletableFuture success
System.out.println(cf1.get()); // 1
System.out.println(cf2.get()); // 2
System.out.println(cf3.get()); // 3

异常处理

参考如下代码，当cf1执行出现异常，所有下游的CompletableFuture都将出错，可以通过以下两个方法验证：

1. isCompletedExceptionally() 会返回true
2. get() 方法会抛出ExecutionException异常，异常内容是cf1中出现的异常.

public class CF6 {

    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> 4 / 1).thenAccept(v -> System.out.println(v)); // 4
        System.out.println("1 -------");

        // 如果异常不处理，所有下游的CompletableFuture将都会出错
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> 4 / 0);
        CompletableFuture<Integer> cf2 = cf1.thenApply(v -> v++);
        cf2.thenAccept(v -> System.out.println(v)); // nothing
        System.out.println(cf2.isCompletedExceptionally()); // true
        try {
            System.out.println(cf2.get());
        } catch (InterruptedException | ExecutionException e) {
            System.out.println("error"); // error
            
        }

        System.out.println("2 -------");
        // 通过exceptionally处理异常，提供
        CompletableFuture.supplyAsync(() -> 4 / 0).exceptionally(ex -> -1).thenAccept(v -> System.out.println(v)); // -1
    }
}

执行结果：

4
1 -------
true
error
2 -------
-1

异常可以通过exceptionally(Function<Throwable, T> function)方法来处理，如果出现异常，异常会被传入该方法，如果没有异常，直接返回上游的结果。

异常还可以通过handle(BiFunction<T, Throwable, R> bifunction)来处理，如果上游处理出现异常，T值将为null，异常为上游异常。如果上游正常执行，T将是上游的返回值，异常将是null.

CompletableFuture.supplyAsync(() -> 4 / 0)
                .handle((v, ex) -> v != null ? v : -1)
                .thenAccept(v -> System.out.println(v)); // -1

第三种是通过whenComplete(BiConsumer<T, Throwable> biconsumer)来处理异常，同handle, 如果上游处理出现异常，T值将为null，异常为上游异常。如果上游正常执行，T将是上游的返回值，异常将是null。不同的是如果出现异常，whenComplete后续将没有返回值，如果没有异常将向下返回上游的返回值。

CompletableFuture.supplyAsync(() -> 4 / 1)
                .whenComplete((v, ex) -> {
                    if (ex == null) {
                        System.out.println(v); // 4
                    } else {
                        System.out.println(ex.getMessage());
                    }
                }).thenAccept(v -> System.out.println(v)); // 4, 只有当whenComplete之前没有异常才会取到上游的返回值

CompletableFuture.supplyAsync(() -> 4 / 0)
        .whenComplete((v, ex) -> {
            if (ex == null) {
                System.out.println(v);
            } else {
                System.out.println(ex.getMessage()); // java.lang.ArithmeticException: / by zero
            }
        }).thenAccept(v -> System.out.println(v)); // 无

---

参考：

https://stackoverflow.com/questions/35329845/difference-between-completablefuture-future-and-rxjavas-observable
https://blog.knoldus.com/2018/01/20/future-vs-completablefuture-1/
https://blog.knoldus.com/2018/03/30/future-vs-completablefuture-in-java-2/
http://www.deadcoderising.com/java8-writing-asynchronous-code-with-completablefuture/
https://community.oracle.com/docs/DOC-995305