从 List<CompletableFuture> 转换为 CompletableFuture<List>

Question

我正在尝试将List<CompletableFuture<X>>转换为CompletableFuture<List<T>>。这非常有用，当您有许多异步任务并且需要获取所有任务的结果时。

如果其中任何一个失败，则最终的future会失败。这是我实现的方式：

public static <T> CompletableFuture<List<T>> sequence2(List<CompletableFuture<T>> com, ExecutorService exec) {
    if(com.isEmpty()){
        throw new IllegalArgumentException();
    }
    Stream<? extends CompletableFuture<T>> stream = com.stream();
    CompletableFuture<List<T>> init = CompletableFuture.completedFuture(new ArrayList<T>());
    return stream.reduce(init, (ls, fut) -> ls.thenComposeAsync(x -> fut.thenApplyAsync(y -> {
        x.add(y);
        return x;
    },exec),exec), (a, b) -> a.thenCombineAsync(b,(ls1,ls2)-> {
        ls1.addAll(ls2);
        return ls1;
    },exec));
}

运行它：

ExecutorService executorService = Executors.newCachedThreadPool();
Stream<CompletableFuture<Integer>> que = IntStream.range(0,100000).boxed().map(x -> CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep((long) (Math.random() * 10));
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return x;
}, executorService));
CompletableFuture<List<Integer>> sequence = sequence2(que.collect(Collectors.toList()), executorService);

如果其中任何一个失败，它就会失败。即使有一百万个 futures，也能按预期输出。我的问题是：假设有超过 5000 个 futures，如果其中任何一个失败，我会得到 StackOverflowError。

线程“pool-1-thread-2611”中的异常 java.lang.StackOverflowError at java.util.concurrent.CompletableFuture.internalComplete(CompletableFuture.java:210) at java.util.concurrent.CompletableFuture$ThenCompose.run(CompletableFuture.java:1487) at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:193) at java.util.concurrent.CompletableFuture.internalComplete(CompletableFuture.java:210) at java.util.concurrent.CompletableFuture$ThenCompose.run(CompletableFuture.java:1487)

我做错了什么？

注意：上面返回的 future 在任何一个 future 失败时都会失败。接受的答案也应该考虑到这一点。

- Jatin

---

如果我是你，我会实现一个“Collector”而不是... - fge

@fge，这实际上是一个非常好的建议。我来自Scala世界，在那里我们有类似的东西。在这里Collector可能更适合。但我想实现方式可能会相似。 - Jatin

Answer 1

使用CompletableFuture.allOf(...)：

static<T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> com) {
    return CompletableFuture.allOf(com.toArray(new CompletableFuture<?>[0]))
            .thenApply(v -> com.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList())
            );
}

关于你的实现，我有一些评论：

你使用的.thenComposeAsync、.thenApplyAsync和.thenCombineAsync可能不是你预期的行为。这些...Async方法会在一个单独的线程中运行提供给它们的函数。因此，在你的情况下，你会导致将新项目添加到列表中在提供的执行器中运行。没有必要将轻量级操作塞进缓存的线程执行器中。不要没有充分理由就使用thenXXXXAsync方法。

此外，reduce不应该用于累加可变容器中的元素。即使在流是顺序的时候它可能能够正确工作，但如果流被并行处理，它将失败。要执行可变减少操作，请改用.collect。

如果你想在第一个失败后立即完成整个计算，请在你的sequence方法中执行以下操作：

CompletableFuture<List<T>> result = CompletableFuture.allOf(com.toArray(new CompletableFuture<?>[0]))
        .thenApply(v -> com.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList())
        );

com.forEach(f -> f.whenComplete((t, ex) -> {
    if (ex != null) {
        result.completeExceptionally(ex);
    }
}));

return result;

如果您希望在第一次失败后取消剩余的操作，可以在result.completeExceptionally(ex);之后立即添加exec.shutdownNow();。当然，这假定exec仅存在于此一次计算中。如果不是这样，您将需要循环遍历并逐个取消每个剩余的Future。

- Misha

---

有一件事我不太明白，就是allof的返回类型是CompletableFuture<Void>，但我们却返回了CompletableFuture<List<T>>，而且没有任何编译器警告。我之前并不知道void的这种性质。 - Jatin

@Jatin 我想你可能是对的。明天早上我精神更好时会重新考虑并相应地修改我的答案。 - Misha

@Jatin 你说得对，在当前reduce的实现中，只要保持sequence2方法中的流是顺序的，ArrayList就是安全的。然而，编写打破流并行化后导致错误的流构造非常不可取。至少，如果你依赖于流的顺序，就应该将reduce的第三个参数设置为(a, b) -> {throw new IllegalStateException("Parallel not allowed");}。 - Misha

这正是您原始解决方案（使用thenCombine）的行为。如果您想要短路计算并立即触发异常完成，这很容易做到。请参见更新的答案。 - Misha

@AbhijitSarkar иҝҷдәӣд»»еҠЎе№¶дёҚжҳҜз”ұjoinи°ғз”Ёзҡ„гҖӮдҪҝз”ЁallOfзҡ„еҘҪеӨ„еңЁдәҺеҪ“allOfи§ҰеҸ‘ж—¶пјҢжүҖжңүд»»еҠЎйғҪе·Іе®ҢжҲҗпјҢиҖҢjoinеҸӘиҺ·еҸ–з»“жһңгҖӮ - Misha

Answer 2

您可以获取 Spotify 的 CompletableFutures 库并使用 allAsList 方法。我认为它受到了 Guava 的 Futures.allAsList 方法的启发。

public static <T> CompletableFuture<List<T>> allAsList(
    List<? extends CompletionStage<? extends T>> stages) {

---

如果你不想使用库，这里是一个简单的实现：

public <T> CompletableFuture<List<T>> allAsList(final List<CompletableFuture<T>> futures) {
    return CompletableFuture.allOf(
        futures.toArray(new CompletableFuture[futures.size()])
    ).thenApply(ignored ->
        futures.stream().map(CompletableFuture::join).collect(Collectors.toList())
    );
}

- oskansavli

Answer 3

正如Misha指出的，您正在过度使用...Async操作。此外，您正在组合一系列操作，模拟一种不反映程序逻辑的依赖关系：

• 您创建了一个作业x，该作业依赖于列表中的第一个和第二个作业
• 您创建了一个作业x + 1，该作业依赖于作业x和列表中的第三个作业
• 您创建了一个作业x + 2，该作业依赖于作业x + 1和列表中的第4个作业
• …
• 您创建了一个作业x + 5000，该作业依赖于作业x + 4999和列表中的最后一个作业

然后，取消（显式或由于异常）这个递归组合的作业可能会进行递归执行，并且可能会因为StackOverflowError而失败。这取决于具体实现。

正如Misha已经展示的那样，有一个方法allOf，可以使您实现原始意图，定义一个作业依赖于列表中的所有作业。

然而，值得注意的是，甚至那也不是必要的。由于您正在使用无界线程池执行器，因此可以简单地发布一个异步作业，将结果收集到列表中，然后您就完成了。等待完成被暗示了，因为无论如何都会请求每个作业的结果。

ExecutorService executorService = Executors.newCachedThreadPool();
List<CompletableFuture<Integer>> que = IntStream.range(0, 100000)
  .mapToObj(x -> CompletableFuture.supplyAsync(() -> {
    LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos((long)(Math.random()*10)));
    return x;
}, executorService)).collect(Collectors.toList());
CompletableFuture<List<Integer>> sequence = CompletableFuture.supplyAsync(
    () -> que.stream().map(CompletableFuture::join).collect(Collectors.toList()),
    executorService);

当线程数量有限而作业可能会产生额外的异步作业时，使用组合依赖操作的方法很重要，以避免等待作业从必须先完成的作业中窃取线程。但这里两种情况都不存在。

在这种特定情况下，一个作业只需迭代这个大量的前提作业，并在必要时等待，可能比建模这个大量的依赖关系并让每个作业通知有关完成情况的依赖作业更有效。

- Holger

---

一个需要注意的问题是，如果使用supplyAsync而不是allOf，将会消耗线程池中的一个线程来等待所有任务完成。如果我没记错的话，allOf会在分配给各自任务的线程内操作。对于大多数用例来说不是什么大问题，但值得注意。 - Misha

@Misha：我确实提到了如果线程数量有限，它将会占用一个线程，并且它之所以在这里工作是因为使用了无限制的线程池执行器（并且没有异步子任务被生成）。 - Holger

@Holger 这个答案的问题在于：如果稍后的任何一个未来失败了，它仍然会等待其中一个完成。相反，一旦出现问题，返回的 future 就应该立即失败。 - Jatin

实际上，我对这个事实感到满意。但不包括线程窃取。 - Jatin

Answer 4

补充一下 @Misha 所提供的答案，它还可以进一步扩展为一个收集器：

 public static <T> Collector<CompletableFuture<T>, ?, CompletableFuture<List<T>>> sequenceCollector() {
    return Collectors.collectingAndThen(Collectors.toList(), com -> sequence(com));
}

现在您可以：

Stream<CompletableFuture<Integer>> stream = Stream.of(
    CompletableFuture.completedFuture(1),
    CompletableFuture.completedFuture(2),
    CompletableFuture.completedFuture(3)
);
CompletableFuture<List<Integer>> ans = stream.collect(sequenceCollector());

- Jatin

Answer 5

使用thenCombine在CompletableFuture上进行序列操作的示例

public<T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> com){

    CompletableFuture<List<T>> identity = CompletableFuture.completedFuture(new ArrayList<T>());

    BiFunction<CompletableFuture<List<T>>,CompletableFuture<T>,CompletableFuture<List<T>>> combineToList = 
            (acc,next) -> acc.thenCombine(next,(a,b) -> { a.add(b); return a;});

    BinaryOperator<CompletableFuture<List<T>>> combineLists = (a,b)-> a.thenCombine(b,(l1,l2)-> { l1.addAll(l2); return l1;}) ;  

    return com.stream()
              .reduce(identity,
                      combineToList,
                      combineLists);  

   }
} 

如果您不介意使用第三方库，cyclops-react（我是作者）提供了一组针对CompletableFuture（以及Optional、Stream等）的实用程序方法。

  List<CompletableFuture<String>> listOfFutures;

  CompletableFuture<ListX<String>> sequence =CompletableFutures.sequence(listOfFutures);

- John McClean

Answer 6

免责声明: 这篇文章并不能完全回答最初的问题。它将缺少“如果一个失败就全部失败”的部分。然而，我无法回答实际更通用的问题，因为它被关闭为这个问题的重复: Java 8 CompletableFuture.allOf(...) with Collection or List。所以我会在这里回答:

如何使用Java 8的流API将 List<CompletableFuture<V>> 转换为 CompletableFuture<List<V>>？

总结: 使用以下代码：

private <V> CompletableFuture<List<V>> sequence(List<CompletableFuture<V>> listOfFutures) {
    CompletableFuture<List<V>> identity = CompletableFuture.completedFuture(new ArrayList<>());

    BiFunction<CompletableFuture<List<V>>, CompletableFuture<V>, CompletableFuture<List<V>>> accumulator = (futureList, futureValue) ->
        futureValue.thenCombine(futureList, (value, list) -> {
                List<V> newList = new ArrayList<>(list.size() + 1);
                newList.addAll(list);
                newList.add(value);
                return newList;
            });

    BinaryOperator<CompletableFuture<List<V>>> combiner = (futureList1, futureList2) -> futureList1.thenCombine(futureList2, (list1, list2) -> {
        List<V> newList = new ArrayList<>(list1.size() + list2.size());
        newList.addAll(list1);
        newList.addAll(list2);
        return newList;
    });

    return listOfFutures.stream().reduce(identity, accumulator, combiner);
}

例子用法：

List<CompletableFuture<String>> listOfFutures = IntStream.range(0, numThreads)
    .mapToObj(i -> loadData(i, executor)).collect(toList());

CompletableFuture<List<String>> futureList = sequence(listOfFutures);

完整示例：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.BiFunction;
import java.util.function.BinaryOperator;
import java.util.stream.IntStream;

import static java.util.stream.Collectors.toList;

public class ListOfFuturesToFutureOfList {

    public static void main(String[] args) {
        ListOfFuturesToFutureOfList test = new ListOfFuturesToFutureOfList();
        test.load(10);
    }

    public void load(int numThreads) {
        final ExecutorService executor = Executors.newFixedThreadPool(numThreads);

        List<CompletableFuture<String>> listOfFutures = IntStream.range(0, numThreads)
            .mapToObj(i -> loadData(i, executor)).collect(toList());

        CompletableFuture<List<String>> futureList = sequence(listOfFutures);

        System.out.println("Future complete before blocking? " + futureList.isDone());

        // this will block until all futures are completed
        List<String> data = futureList.join();
        System.out.println("Loaded data: " + data);

        System.out.println("Future complete after blocking? " + futureList.isDone());

        executor.shutdown();
    }

    public CompletableFuture<String> loadData(int dataPoint, Executor executor) {
        return CompletableFuture.supplyAsync(() -> {
            ThreadLocalRandom rnd = ThreadLocalRandom.current();

            System.out.println("Starting to load test data " + dataPoint);

            try {
                Thread.sleep(500 + rnd.nextInt(1500));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("Successfully loaded test data " + dataPoint);

            return "data " + dataPoint;
        }, executor);
    }

    private <V> CompletableFuture<List<V>> sequence(List<CompletableFuture<V>> listOfFutures) {
        CompletableFuture<List<V>> identity = CompletableFuture.completedFuture(new ArrayList<>());

        BiFunction<CompletableFuture<List<V>>, CompletableFuture<V>, CompletableFuture<List<V>>> accumulator = (futureList, futureValue) ->
            futureValue.thenCombine(futureList, (value, list) -> {
                    List<V> newList = new ArrayList<>(list.size() + 1);
                    newList.addAll(list);
                    newList.add(value);
                    return newList;
                });

        BinaryOperator<CompletableFuture<List<V>>> combiner = (futureList1, futureList2) -> futureList1.thenCombine(futureList2, (list1, list2) -> {
            List<V> newList = new ArrayList<>(list1.size() + list2.size());
            newList.addAll(list1);
            newList.addAll(list2);
            return newList;
        });

        return listOfFutures.stream().reduce(identity, accumulator, combiner);
    }

}

- Kai Stapel

---

在累加器中，您应该使用thenCombine()而不是thenApply()，以避免调用join()。否则，调用线程实际上会执行该操作，因此集合只有在所有操作完成后才会返回。您可以在futureList.join()之前添加一个打印语句来检查这一点：只有在所有未来的“_成功加载测试数据_”都已打印出来后，它才会被打印出来。 - Didier L

@DidierL 如果我将thenApply()更改为thenCombine()，那么对CompletableFuture<List<V>>的最终join()调用将不再阻塞，而是立即返回一个空结果。因此，列表的未来将不会等待所有单个未来完成。但这是整个事情的最初想法。 - Kai Stapel

是的，我确实忘记了Collector依赖于变异。你代码的问题在于它等同于CompletableFuture.completedFuture(listOfFutures.stream().map(CompletableFuture::join).collect(toList()));。集合实际上返回的是一个已经完成的future，所以不再需要返回future了。 - Didier L

你可能是正确的，这与我的“完整示例”在功能上是等效的。然而，该示例仅用于说明如何使用toFutureList()收集器。不等效的是listOfFutures.stream().map(CompletableFuture::join).collect(toList())和listOfFutures.stream().collect(toFutureList())。前者为您提供了所有未来完成的完整结果，而后者为您提供了一个值列表的未来，您可以将其传递或映射到其他值而不会阻塞。 - Kai Stapel

你的理解是错误的：后者实际上做了同样的事情。你的收集器只是在调用线程上对所有 future 调用 join()，并将结果包装在一个已经完成的 CompletableFuture 中。这是阻塞的。正如我之前所说，只需在流收集后添加一个打印语句，你就会发现这个打印语句只会在所有 future 完成后才会出现。 - Didier L

是的，你说得对。我已经相应地更新了我的答案。感谢你指出这一点。 - Kai Stapel

Answer 7

您的任务可以像以下这样轻松完成：

final List<CompletableFuture<Module> futures =...
CompletableFuture.allOf(futures.stream().toArray(CompletableFuture[]::new)).join();

- Janitha Madushan

Answer 8

Javaslang拥有非常方便的Future API。它还允许将一个future集合转换成一个future集合。

List<Future<String>> listOfFutures = ... 
Future<Seq<String>> futureOfList = Future.sequence(listOfFutures);

请查看http://static.javadoc.io/io.javaslang/javaslang/2.0.5/javaslang/concurrent/Future.html#sequence-java.lang.Iterable-

- Mathias Dpunkt

---

我喜欢这个答案。但它依赖于 javaslang.concurrent.Future :( - Jatin

这是真的 - 但是如果你使用过javaslang Future，你就不想回到java Future或CompletableFuture了。 - Mathias Dpunkt

Answer 9

除了 Spotify Futures 库之外，您还可以尝试我的代码，位于此处：https://github.com/vsilaev/java-async-await/blob/master/net.tascalate.async.examples/src/main/java/net/tascalate/concurrent/CompletionStages.java（具有与同一包中其他类的依赖关系）。

它实现了一个逻辑，以返回“至少 N 个完成阶段中的 M 个”CompletionStage-s，并规定了允许容忍多少错误。对于所有/任何情况，都有方便的方法，剩余 futures 的取消策略，以及代码处理 CompletionStage-s（接口）而不是 CompletableFuture（具体类）。

- Valery Silaev

---

更新：建议的代码已移动到单独的库中，https://github.com/vsilaev/tascalate-concurrent - Valery Silaev

如果链接失效，这个答案就毫无用处了。请将代码嵌入到答案中。 - Simon Forsberg