CompletableFuture 详解

JDK1.8中的CompletableFuture就为我们提供了异步函数式编程,CompletableFuture提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合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)

入参Runnable

@FunctionalInterface
public interface Runnable {
public abstract void run();
}

入参Supplier

@FunctionalInterface
public interface Supplier<T> {
T get();
}

executor参数指定使用的线程池，没有指定会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。

runAsync不支持返回值。

supplyAsync支持返回值。

示例

CompletableFuture<Void> runAsync = CompletableFuture.runAsync(
() -> System.out.println("[1]"+Thread.currentThread().getId() + "--" + Thread.currentThread().getName()));

CompletableFuture<String> supplyAsync = CompletableFuture
.supplyAsync(() -> {
System.out.println("[2]"+Thread.currentThread().getId() + "--" + Thread.currentThread().getName());
return "hello";
});

System.out.println("[3]"+Thread.currentThread().getId() + "--" + Thread.currentThread().getName());

//获取返回值
System.out.println(runAsync.get());
System.out.println(supplyAsync.get());

结果

[1]50--ForkJoinPool.commonPool-worker-1

[3]30--http-nio-8080-exec-1

[2]50--ForkJoinPool.commonPool-worker-1

null

hello

CompletableFuture 回调方法

用于CompletableFuture的计算完成，或者抛出异常的时候，进行下一步操作，主要方法如下：

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)

1.exceptionally

当原始的CompletableFuture抛出异常的时候，会进入exceptionally方法，用来处理异常的情况

入参Function

@FunctionalInterface
public interface Function<T, R> {

R apply(T t);

...

}

2.whenComplete、whenCompleteAsync

whenComplete入参BiConsumer<? super T,? super Throwable>它可以处理正常的计算结果，或者异常情况

第一个参数T:  CompletableFuture计算结果

第二个参数Throwable：CompletableFuture抛出的异常

@FunctionalInterface
public interface BiConsumer<T, U> {

void accept(T t, U u);

...

}

带async后缀函数，由线程池里面拿一个空的线程或者新启一个线程来执行回调，与执行回调的runAsync或suppleAsync线程以及执行whenCompleteAsync事件注册的线程无关。

示例

    CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(() -> {
System.out.println("[1]"+Thread.currentThread().getId() + "--" + Thread.currentThread().getName());
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
});

supplyAsync.whenComplete((data, err) -> {
System.out.println("[2]"+Thread.currentThread().getId() + "--" + Thread.currentThread().getName());
if (err != null) {
System.out.println(err.getMessage());
}
data++;
System.out.println("[2]"+data);
});
System.out.println("[main]"+supplyAsync.get());
System.out.println("[main]"+Thread.currentThread().getId() + "--" + Thread.currentThread().getName());
}

结果

[1]51--ForkJoinPool.commonPool-worker-1

[2]51--ForkJoinPool.commonPool-worker-1

[main]1

[main]31--http-nio-8080-exec-1

[2]2

runAsync无返回值，whenComplete入参BiConsumer<? super T,? super Throwable>的第一个参数T的类型为Void

注意

执行whenComplete内容的线程不一定是执行runAsync或suppleAsync的线程，也可能是执行到whenComplete事件注册的线程。

以示例中线程为例：

当线程【id：51】执行完毕的时候，主线程还没有执行到whenComplete（即whenComplete事件还没有注册），那么线程【id：51】就不会去同步执行whenComplete的回调。这个时候哪个线程执行到了whenComplete的事件注册，就由哪个线程自己来同步执行whenComplete的事件内容，示例为主线程执行。

若线程主线程在线程【id：51】执行完毕前，执行到whenComplete的事件注册，则由线程【id：51】执行whenComplete内容

3.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)

入参Function

@FunctionalInterface
public interface Function<T, R> {

R apply(T t);

...

}

与whenComplete的区别

whenComplete返回的CompletableFuture类型与上一个任务类型一致，且whenComplete执行完毕时两个任务返回值一致。

thenApply返回的CompletableFuture类型、返回值可不与上一个任务类型一致

whenComplete可处理异常情况，而thenApply异常时不执行

示例

CompletableFuture<List<String>> supplyAsync1 = CompletableFuture.supplyAsync(() -> Lists.newArrayList("1"));
CompletableFuture<String> stringCompletableFuture = supplyAsync1.thenApply((a) -> a.get(0));
CompletableFuture<List<String>> integerCompletableFuture = supplyAsync1.whenComplete((data, err) -> data.add(1,"2"));
Thread.sleep(1000);
System.out.println("supplyAsync1:"+supplyAsync1.get());
System.out.println("stringCompletableFuture:"+stringCompletableFuture.get());
System.out.println("integerCompletableFuture:"+integerCompletableFuture.get());

结果

supplyAsync1:[1, 2]

stringCompletableFuture:1

integerCompletableFuture:[1, 2]

4.thenAccept

接收任务的处理结果，并消费处理，无返回结果。

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);

入参Consumer

@FunctionalInterface
public interface Consumer<T> {

void accept(T t);

...

｝

示例

CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(() -> 1);
CompletableFuture<Void> voidCompletableFuture = supplyAsync.thenAccept(data -> System.out.println(data));

5.handle

对上一个任务的结果进行处理。
handle 方法和 thenApply 方法处理方式基本一样。不同的是 handle 是在任务完成后再执行，还可以处理异常的任务。thenApply 只可以执行正常的任务，任务出现异常则不执行 thenApply 方法。

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

入参BiFunction

@FunctionalInterface
public interface BiFunction<T, U, R> {

R apply(T t, U u);

...

}

handle与whenComplete的区别

1.都是对结果进行处理,handle有返回值,whenComplete没有返回值

2.上一个任务异常时whenComplete可判断异常但不处理异常（仍然会抛异常）,而handle可实现exceptionally的功能：对异常进行"美化"（不再抛异常）

6、thenRun

跟 thenAccept 方法不一样的是，不关心任务的处理结果。只要上面的任务执行完成，就开始执行 thenAccept 。

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

入参Runnable

@FunctionalInterface
public interface Runnable {
public abstract void run();
}

该方法同 thenAccept 方法类似。不同的是上个任务处理完成后，并不会把计算的结果传给 thenRun 方法。只是处理完任务后，执行 thenAccept 的后续操作。

7、thenCombine 合并任务

thenCombine 会把两个 CompletionStage 的任务都执行完成后，把两个任务的结果一块交给 thenCombine 来处理（类似thenApply转化操作）。

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);

入参

参数一：CompletionStage<? extends U> other

Public  class  CompletableFuture<T>  implements Future<T>, CompletionStage<T>

参数二：BiFunction

@FunctionalInterface
public interface BiFunction<T, U, R> {
R apply(T t, U u);

...

｝

8、thenAcceptBoth

当两个CompletionStage都执行完成后，接收两个任务的处理结果，并消费处理，无返回结果（类似thenAccept）。

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<? extends U> other

Public  class  CompletableFuture<T>  implements Future<T>, CompletionStage<T>

参数二：BiConsumer

@FunctionalInterface
public interface BiConsumer<T, U> {

void accept(T t, U u);

...

}

9、applyToEither 方法

两个CompletionStage，谁执行返回的结果快，就用哪个CompletionStage的结果进行下一步的转化操作（类似thenApply）。

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);

入参二：Function

@FunctionalInterface
public interface Function<T, R> {

R apply(T t);

...

}

10、acceptEither 方法

两个CompletionStage，谁执行返回的结果快，就用哪个CompletionStage的结果进行下一步的消耗操作（类似thenAccept），无返回值。

public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other,Consumer<? super T> action);

public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action);

public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action,Executor executor);

入参二：Consumer

@FunctionalInterface
public interface Consumer<T> {

void accept(T t);

...

｝

11、runAfterEither 方法

两个CompletionStage，任何一个完成了都会执行下一步的操作（类似thenRun ）

public CompletionStage<Void> runAfterEither(CompletionStage<?> other,Runnable action);

public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action);

public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action,Executor executor);

入参二：Runnable

@FunctionalInterface
public interface Runnable {
public abstract void run();
}

12、runAfterBoth

两个CompletionStage，都完成了计算才会执行下一步的操作（类似thenRun ）

public CompletionStage<Void> runAfterBoth(CompletionStage<?> other,Runnable action);

public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action);

public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action,Executor executor);

入参二：Runnable

@FunctionalInterface
public interface Runnable {
public abstract void run();
}

13、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) ;

入参Function

@FunctionalInterface
public interface Function<T, R> {

R apply(T t);

...

}

Function<? super T, ? extends CompletionStage<U>> fn

返回值为CompletionStage类型

示例

CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(() -> 1);
CompletableFuture<String> future = supplyAsync
.thenCompose(integer -> CompletableFuture.supplyAsync(() -> "结果为：" + integer * 10));

thenApply()和thenCompose（）的区别：

thenapply（）是返回的是非CompletableFuture类型：它的功能相当于将CompletableFuture<T>转换成CompletableFuture<U>。

thenCompose（）用来连接两个CompletableFuture，返回值是新的CompletableFuture：

总结：thenApply（）转换的是泛型中的类型，是同一个CompletableFuture；
thenCompose（）用来连接两个CompletableFuture，是生成一个新的CompletableFuture。