Java编程的逻辑 (94) - 组合式异步编程

Posted 2021-03-03 Ivyvivid

前面两节讨论了Java 8中的函数式数据处理，那是对38节到55节介绍的容器类的增强，它可以将对集合数据的多个操作以流水线的方式组合在一起。本节继续讨论Java 8的新功能，主要是一个新的类CompletableFuture，它是对65节到83节介绍的并发编程的增强，它可以方便地将多个有一定依赖关系的异步任务以流水线的方式组合在一起，大大简化多异步任务的开发。

之前介绍了那么多并发编程的内容，还有什么问题不能解决？CompletableFuture到底能解决什么问题？与之前介绍的内容有什么关系？具体如何使用？基本原理是什么？本节进行详细讨论，我们先来看它要解决的问题。

异步任务管理

在现代软件开发中，系统功能越来越复杂，管理复杂度的方法就是分而治之，系统的很多功能可能会被切分为小的服务，对外提供Web API，单独开发、部署和维护。比如，在一个电商系统中，可能有专门的产品服务、订单服务、用户服务、推荐服务、优惠服务、搜索服务等，在对外具体展示一个页面时，可能要调用多个服务，而多个调用之间可能还有一定的依赖，比如，显示一个产品页面，需要调用产品服务，也可能需要调用推荐服务获取与该产品有关的其他推荐，还可能需要调用优惠服务获取该产品相关的促销优惠，而为了调用优惠服务，可能需要先调用用户服务以获取用户的会员级别。

另外，现代软件经常依赖很多第三方的服务，比如地图服务、短信服务、天气服务、汇率服务等，在实现一个具体功能时，可能要访问多个这样的服务，这些访问之间可能存在着一定的依赖关系。

为了提高性能，充分利用系统资源，这些对外部服务的调用一般都应该是异步的、尽量并发的。我们在77节介绍过异步任务执行服务，使用ExecutorService可以方便地提交单个独立的异步任务，可以方便地在需要的时候通过Future接口获取异步任务的结果，但对于多个尤其是有一定依赖关系的异步任务，这种支持就不够了。

于是，就有了CompletableFuture，它是一个具体的类，实现了两个接口，一个是Future，另一个是CompletionStage，Future表示异步任务的结果，而CompletionStage字面意思是完成阶段，多个CompletionStage可以以流水线的方式组合起来，对于其中一个CompletionStage，它有一个计算任务，但可能需要等待其他一个或多个阶段完成才能开始，它完成后，可能会触发其他阶段开始运行。CompletionStage提供了大量方法，使用它们，可以方便地响应任务事件，构建任务流水线，实现组合式异步编程。

具体怎么使用呢？下面我们会逐步说明，CompletableFuture也是一个Future，我们先来看与Future类似的地方。

与Future/FutureTask对比

基本的任务执行服务

我们先通过示例来简要回顾下异步任务执行服务和Future，在异步任务执行服务中，用Callable或Runnable表示任务，以Callable为例，一个模拟的外部任务为：

private static Random rnd = new Random();

static int delayRandom(int min, int max) {
    int milli = max > min ? rnd.nextInt(max - min) : 0;
    try {
        Thread.sleep(min + milli);
    } catch (InterruptedException e) {
    }
    return milli;
}

static Callable<Integer> externalTask = () -> {
    int time = delayRandom(20, 2000);
    return time;
};

externalTask表示外部任务，我们使用了Lambda表达式，不熟悉可以参看91节，delayRandom用于模拟延时。

假定有一个异步任务执行服务，其代码为：

private static ExecutorService executor =
        Executors.newFixedThreadPool(10);

通过任务执行服务调用外部服务，一般返回Future，表示异步结果，示例代码为：

public static Future<Integer> callExternalService(){
    return executor.submit(externalTask);
}

在主程序中，结合异步任务和本地调用的示例代码为：

public static void master() {
    // 执行异步任务
    Future<Integer> asyncRet = callExternalService();

    // 执行其他任务 ...

    // 获取异步任务的结果，处理可能的异常
    try {
        Integer ret = asyncRet.get();
        System.out.println(ret);
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (ExecutionException e) {
        e.printStackTrace();
    }
}

基本的CompletableFuture

使用CompletableFuture可以实现类似功能，不过，它不支持使用Callable表示异步任务，而支持Runnable和Supplier，Supplier替代Callable表示有返回结果的异步任务，与Callale的区别是，它不能抛出受检异常，如果会发生异常，可以抛出运行时异常。

使用Supplier表示异步任务，代码与Callable类似，替换变量类型即可，即：

static Supplier<Integer> externalTask = () -> {
    int time = delayRandom(20, 2000);
    return time;
};

使用CompletableFuture调用外部服务的代码可以为：

public static Future<Integer> callExternalService(){
    return CompletableFuture.supplyAsync(externalTask, executor);
}

supplyAsync是一个静态方法，其定义为：

public static <U> CompletableFuture<U> supplyAsync(
    Supplier<U> supplier, Executor executor)

它接受两个参数supplier和executor，内部，它使用executor执行supplier表示的任务，返回一个CompletableFuture，调用后，任务被异步执行，这个方法立即返回。

supplyAsync还有一个不带executor参数的方法：

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)

没有executor，任务被谁执行呢？与系统环境和配置有关，一般来说，如果可用的CPU核数大于2，会使用Java 7引入的Fork/Join任务执行服务，即ForkJoinPool.commonPool()，该任务执行服务背后的工作线程数一般为CPU核数减1，即Runtime.getRuntime().availableProcessors()-1，否则，会使用ThreadPerTaskExecutor，它会为每个任务创建一个线程。

对于CPU密集型的运算任务，使用Fork/Join任务执行服务是合适的，但对于一般的调用外部服务的异步任务，Fork/Join可能是不合适的，因为它的并行度比较低，可能会让本可以并发的多任务串行运行，这时，应该提供Executor参数。

后面我们还会看到很多以Async结尾命名的方法，一般都有两个版本，一个带Executor参数，另一个不带，其含义是相同的，就不再重复介绍了。

对于类型为Runnable的任务，构建CompletableFuture的方法为：

public static CompletableFuture<Void> runAsync(
    Runnable runnable)
public static CompletableFuture<Void> runAsync(
    Runnable runnable, Executor executor)

它与supplyAsync是类似的，具体就不赘述了。

CompletableFuture对Future的基本增强

Future有的接口，CompletableFuture都是支持的，不过，CompletableFuture还有一些额外的相关方法，比如：

public T join()
public boolean isCompletedExceptionally()
public T getNow(T valueIfAbsent)

join与get方法类似，也会等待任务结束，但它不会抛出受检异常，如果任务异常结束了，join会将异常包装为运行时异常CompletionException抛出。

Future有isDone方法检查任务是否结束了，但不知道任务是正常结束还是异常结束，isCompletedExceptionally方法可以判断任务是否是异常结束了。

getNow与join类似，区别是，如果任务还没有结束，它不会等待，而是会返回传入的参数valueIfAbsent。

进一步理解Future/CompletableFuture

前面例子都使用了任务执行服务，其实，任务执行服务与异步结果Future不是绑在一起的，可以自己创建线程返回异步结果，为进一步理解，我们看些示例。

使用FutureTask调用外部服务，代码可以为：

public static Future<Integer> callExternalService() {
    FutureTask<Integer> future = new FutureTask<>(externalTask);
    new Thread() {
        public void run() {
            future.run();
        }
    }.start();
    return future;
}

内部自己创建了一个线程，线程调用FutureTask的run方法，我们在77节分析过FutureTask的代码，run方法会调用externalTask的call方法，并保存结果或碰到的异常，唤醒等待结果的线程。

使用CompletableFuture，也可以直接创建线程，并返回异步结果，代码可以为：

public static Future<Integer> callExternalService() {
    CompletableFuture<Integer> future = new CompletableFuture<>();
    new Thread() {
        public void run() {
            try {
                future.complete(externalTask.get());
            } catch (Exception e) {
                future.completeExceptionally(e);
            }
        }
    }.start();
    return future;
}

这里使用了CompletableFuture的两个方法：

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

这两个方法显式设置任务的状态和结果，complete设置任务成功完成，结果为value，completeExceptionally设置任务异常结束，异常为ex。Future接口没有对应的方法，FutureTask有相关方法但不是public的(是protected)。设置完后，它们都会触发其他依赖它们的CompletionStage。具体会触发什么呢？我们接下来再看。

响应结果或异常

使用Future，我们只能通过get获取结果，而get可能会需要阻塞等待，而通过CompletionStage，可以注册回调函数，当任务完成或异常结束时自动触发执行，有两类注册方法，whenComplete和handle，我们分别来看下。

whenComplete

whenComplete的声明为：

public CompletableFuture<T> whenComplete(
    BiConsumer<? super T, ? super Throwable> action)

参数action表示回调函数，不管前一个阶段是正常结束还是异常结束，它都会被调用，函数类型是BiConsumer，接受两个参数，第一个参数是正常结束时的结果值，第二个参数是异常结束时的异常，BiConsumer没有返回值。whenComplete的返回值还是CompletableFuture，它不会改变原阶段的结果，还可以在其上继续调用其他函数。看个简单的示例：

CompletableFuture.supplyAsync(externalTask).whenComplete((result, ex) -> {
    if (result != null) {
        System.out.println(result);
    }
    if (ex != null) {
        ex.printStackTrace();
    }
}).join();

result表示前一个阶段的结果，ex表示异常，只可能有一个不为null。

whenComplete注册的函数具体由谁执行呢？一般而言，这要看注册时任务的状态，如果注册时任务还没有结束，则注册的函数会由执行任务的线程执行，在该线程执行完任务后执行注册的函数，如果注册时任务已经结束了，则由当前线程(即调用注册函数的线程)执行。

如果不希望当前线程执行，避免可能的同步阻塞，可以使用其他两个异步注册方法：

public CompletableFuture<T> whenCompleteAsync(
    BiConsumer<? super T, ? super Throwable> action)
public CompletableFuture<T> whenCompleteAsync(
    BiConsumer<? super T, ? super Throwable> action, Executor executor)       

与前面介绍的以Async结尾的方法一样，对第一个方法，注册函数action会由默认的任务执行服务(即ForkJoinPool.commonPool()或ThreadPerTaskExecutor执行)，对第二个方法，会由参数中指定的executor执行。

handle

whenComplete只是注册回调函数，不改变结果，它返回了一个CompletableFuture，但这个CompletableFuture的结果与调用它的CompletableFuture是一样的，还有一个类似的注册方法handle，其声明为：

public <U> CompletableFuture<U> handle(
    BiFunction<? super T, Throwable, ? extends U> fn)

回调函数是一个BiFunction，也是接受两个参数，一个是正常结果，另一个是异常，但BiFunction有返回值，在handle返回的CompletableFuture中，结果会被BiFunction的返回值替代，即使原来有异常，也会被覆盖，比如：

String ret =
    CompletableFuture.supplyAsync(()->{
        throw new RuntimeException("test");
    }).handle((result, ex)->{
        return "hello";
    }).join();
System.out.println(ret);

输出为"hello"。异步任务抛出了异常，但通过handle方法，改变了结果。

与whenComplete类似，handle也有对应的异步注册方法handleAsync，具体我们就不探讨了。

exceptionally

whenComplete和handle都是既响应正常完成也响应异常，如果只对异常感兴趣，可以使用exceptionally，其声明为：

public CompletableFuture<T> exceptionally(
    Function<Throwable, ? extends T> fn)

它注册的回调函数是Function，接受的参数为异常，返回一个值，与handle类似，它也会改变结果，具体就不举例了。

除了响应结果和异常，使用CompletableFuture，可以方便地构建有多种依赖关系的任务流，我们先来看简单的依赖单一阶段的情况。

构建依赖单一阶段的任务流

thenRun

在一个阶段正常完成后，执行下一个任务，看个简单示例：

Runnable taskA = () -> System.out.println("task A");
Runnable taskB = () -> System.out.println("task B");
Runnable taskC = () -> System.out.println("task C");

CompletableFuture.runAsync(taskA)
    .thenRun(taskB)
    .thenRun(taskC)
    .join();

这里，有三个异步任务taskA, taskB和taskC，通过thenRun自然地描述了它们的依赖关系，thenRun是同步版本，有对应的异步版本thenRunAsync：

public CompletableFuture<Void> thenRunAsync(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)

在thenRun构建的任务流中，只有前一个阶段没有异常结束，下一个阶段的任务才会执行，如果前一个阶段发生了异常，所有后续阶段都不会运行，结果会被设为相同的异常，调用join会抛出运行时异常CompletionException。

thenRun指定的下一个任务类型是Runnable，它不需要前一个阶段的结果作为参数，也没有返回值，所以，在thenRun返回的CompletableFuture中，结果类型为Void，即没有结果。

thenAccept/thenApply

如果下一个任务需要前一个阶段的结果作为参数，可以使用thenAccept或thenApply方法：

public CompletableFuture<Void> thenAccept(
    Consumer<? super T> action)
public <U> CompletableFuture<U> thenApply(
    Function<? super T,? extends U> fn)

thenAccept的任务类型是Consumer，它接受前一个阶段的结果作为参数，没有返回值。thenApply的任务类型是Function，接受前一个阶段的结果作为参数，返回一个新的值，这个值会成为thenApply返回的CompletableFuture的结果值。看个简单示例：

Supplier<String> taskA = () -> "hello";
Function<String, String> taskB = (t) -> t.toUpperCase();
Consumer<String> taskC = (t) -> System.out.println("consume: " + t);

CompletableFuture.supplyAsync(taskA)
    .thenApply(taskB)
    .thenAccept(taskC)
    .join();

taskA的结果是"hello"，传递给了taskB，taskB转换结果为"HELLO"，再把结果给taskC，taskC进行了输出，所以输出为：

consume: HELLO

CompletableFuture中有很多名称带有run, accept或apply的方法，它们一般与任务的类型相对应，run与Runnable对应，accept与Consumer对应，apply与Function对应，后续就不赘述了。

thenCompose

与thenApply类似，还有一个方法thenCompose，声明为：

public <U> CompletableFuture<U> thenCompose(
    Function<? super T, ? extends CompletionStage<U>> fn)

这个任务类型也是Function，也是接受前一个阶段的结果，返回一个新的结果，不过，这个转换函数fn的返回值类型是CompletionStage，也就是说，它的返回值也是一个阶段，如果使用thenApply，结果就会变为CompletableFuture<CompletableFuture<U>>，而使用thenCompose，会直接返回fn返回的CompletionStage，thenCompose与thenApply的区别，就如同Stream API中flatMap与map的区别，看个简单的示例：

Supplier<String> taskA = () -> "hello";
Function<String, CompletableFuture<String>> taskB = (t) ->
    CompletableFuture.supplyAsync(() -> t.toUpperCase());
Consumer<String> taskC = (t) -> System.out.println("consume: " + t);

CompletableFuture.supplyAsync(taskA)
    .thenCompose(taskB)
    .thenAccept(taskC)
    .join();

以上代码中，taskB是一个转换函数，但它自己也执行了异步任务，返回类型也是CompletableFuture，所以使用了thenCompose。

构建依赖两个阶段的任务流

依赖两个都完成

thenRun, thenAccept, thenApply和thenCompose用于在一个阶段完成后执行另一个任务，CompletableFuture还有一些方法用于在两个阶段都完成后执行另一个任务，方法是：

public CompletableFuture<Void> runAfterBoth(
    CompletionStage<?> other, Runnable action
public <U,V> CompletableFuture<V> thenCombine(
    CompletionStage<? extends U> other,
    BiFunction<? super T,? super U,? extends V> fn)
public <U> CompletableFuture<Void> thenAcceptBoth(
    CompletionStage<? extends U> other,
    BiConsumer<? super T, ? super U> action) 

runAfterBoth对应的任务类型是Runnable，thenCombine对应的任务类型是BiFunction，接受前两个阶段的结果作为参数，返回一个结果，thenAcceptBoth对应的任务类型是BiConsumer，接受前两个阶段的结果作为参数，但不返回结果。它们都有对应的异步和带Executor参数的版本，用于指定下一个任务由谁执行，具体就不赘述了。当前阶段和参数指定的另一个阶段other没有依赖关系，并发执行，当两个都执行结束后，开始执行指定的另一个任务。

看个简单的示例，任务A和B执行结束后，执行任务C合并结果，代码为：

Supplier<String> taskA = () -> "taskA";
CompletableFuture<String> taskB = CompletableFuture.supplyAsync(() -> "taskB");
BiFunction<String, String, String> taskC = (a, b) -> a + "," + b;

String ret = CompletableFuture.supplyAsync(taskA)
        .thenCombineAsync(taskB, taskC)
        .join();
System.out.println(ret);

输出为：

taskA,taskB

依赖两个阶段中的一个

前面的方法要求两个阶段都完成后才执行下一个任务，如果只需要其中任意一个阶段完成，可以使用下面的方法：

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

public <U> CompletableFuture<U> applyToEither(
    CompletionStage<? extends T> other, Function<? super T, U> fn)

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

它们都有对应的异步和带Executor参数的版本，用于指定下一个任务由谁执行，具体就不赘述了。当前阶段和参数指定的另一个阶段other没有依赖关系，并发执行，只要当其中一个执行完了，就会启动参数指定的另一个任务，具体就不赘述了。

构建依赖多个阶段的任务流

如果依赖的阶段不止两个，可以使用如下方法：

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

它们是静态方法，基于多个CompletableFuture构建了一个新的CompletableFuture。

对于allOf，当所有子CompletableFuture都完成时，它才完成，如果有的CompletableFuture异常结束了，则新的CompletableFuture的结果也是异常，不过，它并不会因为有异常就提前结束，而是会等待所有阶段结束，如果有多个阶段异常结束，新的CompletableFuture中保存的异常是最后一个的。新的CompletableFuture会持有异常结果，但不会保存正常结束的结果，如果需要，可以从每个阶段中获取。看个简单的示例：

CompletableFuture<String> taskA = CompletableFuture.supplyAsync(() -> {
    delayRandom(100, 1000);
    return "helloA";
}, executor);

CompletableFuture<Void> taskB = CompletableFuture.runAsync(() -> {
    delayRandom(2000, 3000);
}, executor);

CompletableFuture<Void> taskC = CompletableFuture.runAsync(() -> {
    delayRandom(30, 100);
    throw new RuntimeException("task C exception");
}, executor);

CompletableFuture.allOf(taskA, taskB, taskC).whenComplete((result, ex) -> {
    if (ex != null) {
        System.out.println(ex.getMessage());
    }
    if (!taskA.isCompletedExceptionally()) {
        System.out.println("task A " + taskA.join());
    }
});

taskC会首先异常结束，但新构建的CompletableFuture会等待其他两个结束，都结束后，可以通过子阶段(如taskA)的方法检查子阶段的状态和结果。

对于anyOf返回的CompletableFuture，当第一个子CompletableFuture完成或异常结束时，它相应地完成或异常结束，结果与第一个结束的子CompletableFuture一样，具体就不举例了。

小结

本节介绍了Java 8中的组合式异步编程CompletableFuture：

• 它是对Future的增强，但可以响应结果或异常事件，有很多方法构建异步任务流
• 根据任务由谁执行，一般有三类对应方法，名称不带Async的方法由当前线程或前一个阶段的线程执行，带Async但没有指定Executor的方法由默认Excecutor执行(ForkJoinPool.commonPool()或ThreadPerTaskExecutor)，带Async且指定Executor参数的方法由指定的Executor执行
• 根据任务类型，一般也有三类对应方法，名称带run的对应Runnable，带accept的对应Consumer，带apply的对应Function

使用CompletableFuture，可以简洁自然地表达多个异步任务之间的依赖关系和执行流程，大大简化代码，提高可读性。

下一节，我们探讨Java 8对日期和时间API的增强。