CompletableFuture原理与用法

CompletableFuture 是由Java8引入的，这让我们编写清晰可读的异步代码变得更加容易，该类功能比Future 更加强大。

什么是 CompletableFuture

在Java中CompletableFuture用于异步编程，异步通常意味着非阻塞，运行任务单独的线程，与主线程隔离。并且通过回调可以在主线程中得到异步任务的执行状态，是否完成和异常等信息。

通过这种方式，主线程不会被阻塞，不需要一直等到子线程完成。主线程可以并行的执行其他任务。使用这种并行方式，可以极大的提高程序的性能。

为什么要引入 CompletableFuture

一些业务场景我们需要使用多线程异步执行任务，加快任务执行速度，所以 JDK5 新增了 Future 接口，用于描述一个异步计算的结果。

虽然 Future 以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，我们必须使用 Future.get 的方式阻塞调用线程，或者使用轮询方式判断 Future.isDone 任务是否结束，再获取结果。

Future<String> future = executor.submit(()->{
       Thread.sleep(2000);
       return "hello world";
});

// 获取结果
System.out.println(future.get());

从上面的形式看来不能及时地得到计算结果，所以要实现真正的异步，上述这样是完全不够的。

若需要更强大的异步处理能力，单纯使用 Future 接口或者 FutureTask 类并不能很好地完成以下业务场景：

将两个异步计算合并为一个，这两个异步计算之间相互独立，同时第二个又依赖于第一个的结果；
等待Future集合种的所有任务都完成；
仅等待Future集合种最快结束的任务完成（有可能因为他们试图通过不同的方式计算同一个值），并返回它的结果；
通过编程方式完成一个Future任务的执行（即以手工设定异步操作结果的方式）；
应对Future的完成时间（即当Future的完成时间完成时会收到通知，并能使用Future的计算结果进行下一步的的操作，不只是简单地阻塞等待操作的结果）。

所以JDK 8.0新增了CompletableFuture 来解决上述这些痛点。

Future vs CompletableFuture

CompletableFuture 是 Future API的扩展。

Future 被用于作为一个异步计算结果的引用。提供一个 isDone() 方法来检查计算任务是否完成。当任务完成时，get() 方法用来接收计算任务的结果。

Future 的局限性

不能手动完成

当你写了一个函数，用于通过一个远程API获取一个电子商务产品最新价格。因为这个 API 太耗时，你把它允许在一个独立的线程中，并且从你的函数中返回一个 Future。现在假设这个API服务宕机了，这时你想通过该产品的最新缓存价格手工完成这个Future 。你会发现无法这样做。

Future 的结果在非阻塞的情况下，不能执行更进一步的操作

Future 不会通知你它已经完成了，它提供了一个阻塞的 get() 方法通知你结果。你无法给 Future 植入一个回调函数，当 Future 结果可用的时候，用该回调函数自动的调用 Future 的结果。

多个 Future 不能串联在一起组成链式调用

有时候你需要执行一个长时间运行的计算任务，并且当计算任务完成的时候，你需要把它的计算结果发送给另外一个长时间运行的计算任务等等。你会发现你无法使用 Future 创建这样的一个工作流。

不能组合多个 Future 的结果

假设你有10个不同的Future，你想并行的运行，然后在它们运行未完成后运行一些函数。你会发现你也无法使用 Future 这样做。

没有异常处理

Future API 没有任务的异常处理结构居然有如此多的限制，幸好我们有CompletableFuture，你可以使用 CompletableFuture 达到以上所有目的。

CompletableFuture 实现了 Future 和 CompletionStage 接口，并且提供了许多关于创建，链式调用和组合多个 Future 的便利方法集，而且有广泛的异常处理支持。

CompletableFuture的应用场景

执行比较耗时的操作时，尤其是那些依赖一个或多个远程服务的操作，使用异步任务可以改善程序的性能，加快程序的响应速度。
使用CompletableFuture类，它提供了异常管理的机制，让你有机会抛出、管理异步任务执行种发生的异常。
如果这些异步任务之间相互独立，或者他们之间的的某一些的结果是另一些的输入，你可以讲这些异步任务构造或合并成一个。

CompletableFuture设计思想

CompletableFuture 按照类似“观察者模式”的设计思想，原理分析可以从“观察者”和“被观察者”两个方面着手。

由于回调种类多，但结构差异不大，所以这里单以一元依赖中的thenApply为例，不再枚举全部回调类型，如下图所示：

被观察者

每个CompletableFuture都可以被看作一个被观察者，其内部有一个Completion类型的链表成员变量stack，用来存储注册到其中的所有观察者。当被观察者执行完成后会弹栈stack属性，依次通知注册到其中的观察者。上面例子中步骤fn2就是作为观察者被封装在UniApply中。
被观察者CF中的result属性，用来存储返回结果数据。这里可能是一次RPC调用的返回值，也可能是任意对象，在上面的例子中对应步骤fn1的执行结果。

观察者

CompletableFuture支持很多回调方法，例如thenAccept、thenApply、exceptionally等，这些方法接收一个函数类型的参数f，生成一个Completion类型的对象（即观察者），并将入参函数f赋值给Completion的成员变量fn，然后检查当前CF是否已处于完成状态（即result!=null），如果已完成直接触发fn，否则将观察者Completion加入到CF的观察者链stack中，再次尝试触发，如果被观察者未执行完则其执行完毕之后通知触发。

观察者中的dep属性：指向其对应的CompletableFuture，在上面的例子中dep指向CF2。
观察者中的src属性：指向其依赖的CompletableFuture，在上面的例子中src指向CF1。
观察者Completion中的fn属性：用来存储具体的等待被回调的函数。这里需要注意的是不同的回调方法（thenAccept、thenApply、exceptionally等）接收的函数类型也不同，即fn的类型有很多种，在上面的例子中fn指向fn2。

CompletableFuture核心功能

CompletableFuture的功能主要体现在它的CompletionStage,如下图所示：

CompletionStage接口定义了任务编排的方法，执行某一阶段，可以向下执行后续阶段，可以实现如下功能：

转换（thenCompose）
组合（thenCombine）
消费（thenAccept）
运行（thenRun）
带返回的消费（thenApply）

具体其他功能大家可以根据需求自行查看。

CompletableFuture创建使用

创建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)

Asynsc 表示异步，而 supplyAsync 与 runAsync 不同在与前者异步返回一个结果，后者是 void。第二个函数第二个参数表示是用我们自己创建的线程池，否则采用默认的 ForkJoinPool.commonPool() 作为它的线程池.其中Supplier是一个函数式接口，代表是一个生成者的意思，传入0个参数，返回一个结果。

CompletableFuture<String> future = CompletableFuture.supplyAsync(()->{
    return "hello world";
});

//阻塞的获取结果  ''helllo world"
System.out.println(future.get());

CompletableFuture用法详解

没有返回值的异步任务

如果你想异步的运行一个后台任务并且不需要任务返回结果，就可以使用 runAsync()。

runAsync() 返回一个新的 CompletableFuture，它在运行给定操作后由在 ForkJoinPool.commonPool() 运行的任务异步完成。

/**
 * 没有返回值的异步任务
 */
@Test
public void runAsync() throws Exception {
    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
        log.info("Current thread name: {}", Thread.currentThread().getName());
        // 模拟长时间运行的作业
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        };
    });

    // 主线程阻塞
    future.get();
    System.out.println("主线程结束");
}

有返回值的异步任务

当运行一个异步任务并且需要有返回结果时，就可以使用 supplyAsync()。

CompletableFuture.supplyAsync() 它持有 supplier<T> 并且返回 CompletableFuture<T>，T 是通过调用传入的 supplier 取得的值的类型。

CompletableFuture<String> future = CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public String get() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return "Success";
    }
});

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

Supplier<T> 是一个简单的函数式接口，表示 supplier 的结果。它有一个 get()，该方法可以写入你的后台任务中，并且返回结果。

还可以使用 lambda 表达式使得上面的示例更加简明：

/**
 * 有返回值的异步任务
 */
@Test
public void supplyAsync() throws Exception {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        log.info("Current thread name: {}", Thread.currentThread().getName());
        // 模拟长时间运行的作业
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        };
        return "Success";
    });

    // 主线程阻塞
    System.out.println(future.get());
}

上述 runAsync() 和 supplyAsync() 都是在单独的线程中执行他们的任务，但在实际业务中我们不会只创建一个线程。

自定义线程池执行方法

CompletableFuture 可以从全局的 ForkJoinPool.commonPool() 获得一个线程中执行这些任务。但也可以创建一个线程池并传给 runAsync() 和 supplyAsync() 来让他们从线程池中获取一个线程执行它们的任务。

CompletableFuture API 的所有方法都有两个变体-一个接受Executor作为参数，另一个不这样：

static CompletableFuture<Void>  runAsync(Runnable runnable)
static CompletableFuture<Void>  runAsync(Runnable runnable, Executor executor)
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

创建一个线程池，并传递给其中一个方法：

Executor executor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    }
    return "Success";
}, executor);

线程串行化

由于 CompletableFuture.get() 方法是阻塞的，它会一直等到 Future 完成，并且在完成后返回结果。但是，这是我们想要的吗？对于构建异步系统，我们应该附上一个回调给 CompletableFuture，当 Future 完成的时候，自动的获取结果。

如果不想等待结果返回，就可以把需要等待 Future 完成执行的逻辑写入到回调函数中。

可以使用 thenApply()、thenAccept() 、thenRun() 回调给 CompletableFuture。

thenApply()

当一个线程依赖另一个线程时，可以使用 thenApply() 来把这两个线程串行化。

thenApply：可以使用 thenApply() 处理和改变 CompletableFuture 的结果。

/**
 * 使用 thenApply() 处理和改变CompletableFuture的结果
 *
 * @throws Exception
 */
@Test
public void thenApply1() throws Exception {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        return "Java";
    }).thenApply(o -> {
        return "Hello " + o;
    });

    System.out.println(future.get());
}

thenAccept()

如果你不想从你的回调函数中返回任何东西，仅仅想在 Future 完成后运行一些代码片段，你可以使用 thenAccept() 和 thenRun()，这些方法经常在调用链的最末端的最后一个回调函数中使用。

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

@Test
public void thenAccept() throws Exception {
    CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
        // 模拟长时间运行的作业
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return "Success";
    }).thenAccept(o -> {
        if ("Success".equals(o)) {
            // 模拟长时间运行的作业
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
        }
    });
    future.get();
    log.info("======================");

    System.out.println("结束.");
}

thenRun()

thenRun() 不能访 Future 的结果，它持有一个 Runnable 返回 CompletableFuture：

@Test
public void thenRun() throws Exception {
    CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
        // 模拟长时间运行的作业
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return "Success";
    }).thenRun(() -> {
        // 作业完成后执行一些代码片段
        System.out.println("thenRun 执行一些代码片段");
    });

    future.get();
}

结果合并

thenCompose（）

使用 thenCompose() 合并两个有依赖关系的 CompletableFutures 的执行结果。

private static Integer num = 10;

@Test
public void thenCompose() throws Exception {
    //第一步加 10
    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("让num+10;任务开始");
        num += 10;
        return num;
    });

    //合并
    CompletableFuture<Integer> future1 = future.thenCompose(i ->
            //再来一个 CompletableFuture
            CompletableFuture.supplyAsync(() -> i + 1)
    );

    System.out.println(future.get());
    System.out.println(future1.get());
}

thenCombine()

使用 thenCombine() 组合两个独立的 future。当两个独立的 Future 都完成的时候使用 thenCombine() 用来做一些事情。

@Test
public void thenCompose() throws Exception {
    // 长方形:S=长*宽
    //第一步加 10
    CompletableFuture<Integer> lengthFuture = CompletableFuture.supplyAsync(() -> {
        return 50;
    });

    CompletableFuture<Integer> widthFuture = CompletableFuture.supplyAsync(() -> {
        return 30;
    });

    CompletableFuture<Integer> combinedFuture = lengthFuture.thenCombine(widthFuture, (t1, t2) -> {
        System.out.println(t1);
        System.out.println(t2);
        return t1 * t2;
    });

    System.out.println(combinedFuture.get());
}

合并多个任务的结果

使用 thenCompose() 和 thenCombine() 可以把两个 CompletableFuture 组合在一起。如果要是想组合任意数量的 CompletableFuture，应该怎么做？

可以使用 allOf 和 anyOf 组合任意多个 CompletableFuture。这两个函数都是静态函数，参数是变长的 CompletableFuture 的集合。

allOf 和 anyOf 的区别，前者是「与」，后者是「或」。

allOf()

allOf 的返回值是 CompletableFuture<Void> 类型，这是因为每个传入的 CompletableFuture 的返回值都可能不同，所以组合的结果是无法用某种类型来表示的，索性返回 Void 类型。那么，如何获取每个 CompletableFuture 的执行结果呢？

例子：并行地下载 N 个资源，待下载完成之后，并资源额外处理。

@Test
public void allOf() throws Exception {
    // URL 列表集合
    List<String> webLinks = Arrays.asList("https://www.baidu.com/", "https://www.bing.com/", "https://www.so.com/");

    // 并行下载多个网页
    List<CompletableFuture<String>> contentFutureList = webLinks.stream().map(webLink -> downloadWebLink(webLink)).collect(Collectors.toList());

    // 通过allof,等待所有网页下载完毕，收集返回结果
    CompletableFuture<Void> allFutures = CompletableFuture.allOf(contentFutureList.toArray(new CompletableFuture[contentFutureList.size()]));


    // 附上回调函数，获取结果集
    // 方式一
    CompletableFuture<List<String>> result = allFutures.thenApply(v -> contentFutureList.stream().map(o -> {
        try {
            return o.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }).collect(Collectors.toList()));
    System.out.println(result.get());

    // 方式二
    result = allFutures.thenApply(v -> contentFutureList.stream().map(CompletableFuture<String>::join).collect(Collectors.toList()));
    System.out.println(result.get());
}

private CompletableFuture<String> downloadWebLink(String webLink) {

    return CompletableFuture.supplyAsync(() -> {
        // 模拟下载过程
        System.out.println("开始下载网页：" + webLink);
        return "这是一个网页内容";
    });

}

这里有个关键问题，因为 allof 没有返回值，所以通过 theApply，给 allFutures 附上一个回调函数。在回调函数里面，以此调用么一个 Future 的 Get() 函数，获取结果后存入 List<String> 中。

anyOf()

anyOf 的含义是只要有任意一个 CompletableFuture 结束，就可以做接下来的事情，而无须像 allOf 那样，等待所有的 CompletableFuture 结束。

但由于每个 CompletableFuture 的返回值类型可能不同，意味着无法判断是什么类型，所以 anyOf 的返回值是 CompletableFuture<Object> 类型。

@Test
public void anyOf() throws Exception {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "future1 结果";
    });
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.SECONDS.sleep(4);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "future2 结果";
    });
    CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "future3 结果";
    });

    CompletableFuture<Object> future = CompletableFuture.anyOf(future1, future2, future3);
    System.out.println(future.get());
}

在该例子中，因为 future1、future2、future3 的返回值都是 CompletableFuture<String>，所以anyOf 的返回的 Object 一定也是 String 类型。

并且在 3 个 future 中，future3 睡眠时间最短，会最先执行完成， anyOfFuture.get() 获取的也就是 future3 的内容。future1、future2 的返回结果会被丢弃。

异常处理

在调用 supplyAsync() 任务中发生一个错误，这时候没有任何 thenApply 会被调用并且 future 将以一个异常结束。如果在第一个 thenApply() 发生错误，这时候第二个和第三个将不会被调用，同样的，future 将以异常结束。

exceptionally()

回调处理异常，从原始Future中生成的错误恢复的机会。

@Test
public void exceptionally() throws Exception {
    Integer age = -1;
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        if (age < 0) {
            throw new IllegalArgumentException("年龄不能为负数");
        }

        return "张三";
    }).exceptionally(ex -> {
        System.out.println(ex.getMessage());
        return "Unknown!";
    });

    System.out.println(future.get());
}

handle()

从异常恢复，无论一个异常是否发生它都会被调用。

@Test
public void handle() throws Exception {
    Integer age = -1;
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        if (age < 0) {
            throw new IllegalArgumentException("年龄不能为负数");
        }

        return "张三";
    }).handle((res, ex) -> {
        if (ex != null) {
            System.out.println(ex.getMessage());
            return "Unknown!";
        }
        return res;
    });

    System.out.println(future.get());
}

如果异常发生 res 参数将是 null，否则 ex 将是 null。

参考：CompletableFuture原理与用法详解