Java8 CompletableFuture指北

一、JAVA 异步处理的演变

1.1 Future

JDK 5引入了Future模式。Future模式是多线程设计常用的一种设计模式。Future模式可以理解成：我有一个任务，提交给了Future，Future替我完成这个任务。期间可以去做其他事情，一段时间之后，我便可以从Future那儿取出结果。Future接口是Java多线程Future模式的实现，在java.util.concurrent包中，可以来进行异步计算。

一般情况下，我们会结合Callable接口和Future接口一起使用，通过ExecutorService的submit方法执行Callable，并返回Future。

以下代码模拟了电商加载商品详情页面的过程。

public void renderPage(CharSequence source) {
List<ImageInfo> info = scanForImageInfo(source);
// create Callable representing download of all images
final Callable<List<ImageData>> task = () ->
  info.stream()
        .map(ImageInfo::downloadImage)
        .collect(Collectors.toList());
    // submit download task to the executor
    Future<List<ImageData>> images = executor.submit(task);
    renderText(source);
try {
   // get all downloaded images (blocking until all are available)
   final List<ImageData> imageDatas = images.get();
   // render images
   imageDatas.forEach(this::renderImage);
} catch (InterruptedException e) {
   // Re-assert the thread’s interrupted status
   Thread.currentThread().interrupt();
   // We don’t need the result, so cancel the task too
   images.cancel(true);
} catch (ExecutionException e) {
  throw launderThrowable(e.getCause()); }
}

Callable表示下载页面所有图像的任务集合提交给Executor执行，该Executor返回Future，通过这些Future，可以查询下载任务的状态。当主线程完成渲染页面的文本时，它会调用Future.get()方法，直到所有下载的结果List<ImageData>完全可用为止。

以上代码明显的缺点是Future.get()方法是阻塞的，在所有下载完成之前，页面没有图像可用于渲染。

1.2 CompletionService

使用CompletionService的一大改进就是把多个图片的加载分发给多个工作单元进行处理

public void renderPage(CharSequence source) { 
  List<ImageInfo> info = scanForImageInfo(source); 
  CompletionService<ImageData> completionService = 
    new ExecutorCompletionService<>(executor); 
  // submit each download task to the completion service 
  info.forEach(imageInfo -> 
    completionService.submit(imageInfo::downloadImage)); 
  renderText(source); 
  // retrieve each RunnableFuture as it becomes 
  // available (and when we are ready to process it). 
  for (int t = 0; t < info.size(); t++) { 
    Future<ImageData> imageFuture = completionService.take(); 
    renderImage(imageFuture.get()); 
  } 
}

1.3 CompletableFuture

CompletableFuture能够将回调放到与任务不同的线程中执行，也能将回调作为继续执行的同步函数，在与任务相同的线程中执行。它避免了传统回调最大的问题，那就是能够将控制流分离到不同的事件处理器中。

public void renderPage(CharSequence source) { 
        List<ImageInfo> info = scanForImageInfo(source); 
        info.forEach(imageInfo -> 
               CompletableFuture 
            .supplyAsync(imageInfo::downloadImage) 
            .thenAccept(this::renderImage)); 
        renderText(source); 
 }

二、CompletableFuture相关API

2.1 声明任务 - runAsync()、supplyAsync()

方法名	描述
CompletableFuture runAsync(Runnable runnable)	使用默认的线程池执行无返回值的异步任务
<U> CompletableFuture<U> supplyAsync(Supplier supplier)	使用默认的线程池执行有返回值的异步任务

2.2 获取结果 - get()、join()

方法名	描述
T get() throws InterruptedException, ExecutionException	阻塞等待异步任务完成
T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException	阻塞等待异步任务完成，并指定任务超时时间，如果超时，则抛出TimeoutException
T join()	阻塞等待异步任务完成

2.3 手动结束 - complete()、completeExceptionally()

方法名	描述
boolean complete(T t)	手动完成异步任务，并指定任务的返回结果
boolean completeExceptionally(Throwable ex)	手动完成异步任务，并抛出异常

2.4 转换结果 - thenApply()、thenCompose()

方法名	描述
<U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)	将执行结果交由指定Function转换
<U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)	将执行结果交由指定Function转换

/** 
 * thenApply() 相当于Stream.map()、thenCompose()相当于Stream.flatMap()
 * thenCompose() 是为了避免 CompletableFuture<CompletableFuture<U>> 式的返回结果
 */
// thenApply()
CompletableFuture<CompletableFuture<String>> future = 
    CompletableFuture.supplyAsync(() -> "Hello")
            .thenApply(result -> CompletableFuture.supplyAsync(() -> result + "World"));
// thenCompose()
CompletableFuture<String> future = 
    CompletableFuture.supplyAsync(() -> "Hello")
            .thenCompose(result -> CompletableFuture.supplyAsync(() -> result + "World"));

2.5 组合结果 - thenCombine()、thenAcceptBoth()

方法名	描述
<U,V> CompletableFuture thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)	并行执行任务other，并用fn组合并转换两个任务的结果
<U> CompletableFuture thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)	并行执行任务other，并用fn组合两个任务的结果，仅消费，无返回值

/**
 * thenCombine() 与 thenAcceptBoth() 都是并行执行两个任务，并对两个任务的结果进行组合
 * thenAcceptBoth()对于结果只是纯消费
 */
// thenCombine()
CompletableFuture<String> task1 = 
    CompletableFuture.supplyAsync(() -> "Hello")
            .thenCombine(CompletableFuture.supplyAsync(() -> "World"), (r1, r2) -> r1 + r2);
// thenAcceptBoth()
CompletableFuture<Void> task2 = 
    CompletableFuture.supplyAsync(() -> "Hello")
            .thenAcceptBoth(CompletableFuture.supplyAsync(() -> "World"), (r1, r2) -> System.out.println(r1 + r2));

2.6 处理结果 - whenComplete()、handle()、thenAccept()、thenRun()

方法名	描述
CompletableFuture whenComplete(BiConsumer<? super T, ? super Throwable> action)	消费结果，可处理异常
<U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)	转换结果，可处理异常
CompletionStage thenAccept(Consumer<? super T> action)	消费结果
CompletableFuture thenRun(Runnable action)	不消费结果，任务结果后执行action

/**
 * whenComplete() 和 handle() 都能捕获异步任务的异常，并针对异常进行结果处理
 * whenComplete() 只消费结果，handle() 可以转换结果
 * thenAccept() 只消费结果
 */
CompletableFuture.supplyAsync(() -> "Hello")
            .whenComplete((result, ex) -> System.out.println(result))
            .handle((result, ex) -> result.concat(" World"))
            .thenAccept(System.out::println);

2.7 时间优先度执行 - acceptEither()、applyToEither()、runAfterEither()

方法名	描述
CompletableFuture acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)	对首先返回的任务结果进行消费
<U> CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)	对首先返回的任务结果进行转换
CompletableFuture runAfterEither(CompletionStage<?> other, Runnable action)	不消费结果，任意一个任务执行完毕即执行action

2.8 工厂方法 - allOf()、anyOf()

方法名	描述
static CompletableFuture allOf(CompletableFuture<?>… cfs)	当所有任务完成后，返回一个新的completed状态的CompletableFuture
static CompletableFuture anyOf(CompletableFuture<?>… cfs)	当任意一个任务完成后，返回一个新的completed状态的CompletableFuture

三、示例

3.1 指定异步任务超时时间

// 某任务需要时间 但是最大的可接受时间为2s 2s后如果任务未完成则抛出TimeoutException并使用默认值结果
CompletableFuture.supplyAsync(() -> getSomething())
                 .applyToEither(timeoutAfter(2, TimeUnit.SECONDS), Function.identity())
                 .exceptionally(ex -> defaultValue())
                 .thenAccept(something -> doAfterGet(something));
// timeoutAfter() 实现
public <T> CompletableFuture<T> timeoutAfter(long timeout, TimeUnit unit) {
    CompletableFuture<T> result = new CompletableFuture<T>();
    delayer.schedule(() -> result.completeExceptionally(new TimeoutException()), timeout, unit);
    return result;
}

3.2 小示例

// 声明任务集合
List<CompletableFuture> tasks = new ArrayList<>(5);
// 根据条件创建任务
Optional<CompletableFuture<U>> optionalTask = Optional.ofNullable(taskService.createTask(params));
// 如果任务创建成功 则处理任务结果：转换、消费 ..
optionalTask.ifPresent(optionalTask ->
    tasks.add(optionalTask
        .thenApplyAsync(firstResult -> {
            convert(firstResult)
        })
        .thenAcceptAsync(finalResult -> {
            dosomething(finalResult)
        })
    )
);
// 声明其他任务
...
// 获取结果
if(!CollectionUtils.isEmpty(tasks)) {
   CompletableFuture[] taskArray = tasks.toArray(new CompletableFuture[tasks.size()]);
   CompletableFuture.allOf(taskArray).join();
}

四、参考文章

• Java8新的异步编程方式 CompletableFuture(一)

• Java8新的异步编程方式 CompletableFuture(二)

• Java8新的异步编程方式 CompletableFuture(三)

• Functional-Style Callbacks Using Java 8’s CompletableFuture

• Asynchronous timeouts with CompletableFutures in Java 8 and Java 9

• Java 8 Concurrency - CompletableFuture in practice