java - CompletableFuture Java 8 异常行为

Question

我注意到 Java 8 中带有流式传输的 CompleteableFutures 出现了一些异常行为。

String [] arr = new String[]{"abc", "def", "cde", "ghj"};
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        List<String> lst =
                Arrays.stream(arr)
                        .map(r ->
                                CompletableFuture.supplyAsync(() -> {
                                    try {
                                        Thread.sleep(5000);
                                        return "e";
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                    return null;
                                }, executorService)
                        )
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList());

上面的代码需要 4*5000 = 20 秒来执行，因此这意味着 future 正在相互等待。

 String [] arr = new String[]{"abc", "def", "cde", "ghj"};
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        List<CompletableFuture<String>> lst =
                Arrays.stream(arr)
                        .map(r ->
                                CompletableFuture.supplyAsync(() -> {
                                    try {
                                        Thread.sleep(5000);
                                        return "d";
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                    return null;
                                }, executorService)
                        )
                        .collect(Collectors.toList());

        List<String> s =
                lst
                        .stream()
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList());

        System.out.println(s);

但是，此代码运行时间为 5 秒，这意味着 future 是并行运行的。

我不明白的地方:在第二个示例中，我明确地获取了 futures 列表，然后进行连接，这需要 5 秒，第一个示例我让它流过，看起来等待。

这背后的原因是什么？

Answer 1

流不一定执行一个阶段，然后执行下一阶段。他们可以按照自己选择的任何顺序组成操作。

例如，

Arrays.stream(array).map(e -> f(e)).map(e -> g(e)).collect(toList());

最终的运行方式与

相同

Arrays.stream(array).map(e -> g(f(e))).collect(toList());

...这将产生您所看到的结果: future 一次生成一个并立即加入，而不是全部预先生成然后加入。

事实上，如果您不执行异步操作，通常采用第二种方式会更高效。这样，流框架就不必存储 f 的所有结果，然后存储 g 的所有结果:它只能存储 g(f(e)) 的结果。流框架无法知道您正在执行异步代码，因此它会执行正常的高效操作。