83

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

如果其中任何一个失败，那么最终的未来就会失败。这就是我实施的方式：

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);

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

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

我做错了什么？

注意：当任何未来失败时，上面返回的未来会失败。接受的答案也应该采取这一点。

4

9 回答 9

100

使用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;

另外，如果您想在第一次失败时取消剩余操作，请exec.shutdownNow();在result.completeExceptionally(ex);. 当然，这假设exec仅存在于这一计算中。如果没有，您将不得不循环并单独取消每个剩余的Future。

于 2015-05-04T09:26:28.363 回答
11

正如Misha 所指出的，您正在过度使用…Async操作。此外，您正在编写一个复杂的操作链，对不反映您的程序逻辑的依赖项进行建模：

  • 您创建了一个工作 x，这取决于您列表中的第一个和第二个工作
  • 您创建了一个工作 x+1，这取决于工作 x 和列表中的第三个工作
  • 您创建了一个工作 x+2，这取决于工作 x+1 和列表中的第 4 个工作
  • …</li>
  • 您创建了一个工作 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);

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

在这种特定情况下，一项作业简单地迭代大量的先决作业并在必要时等待可能比对大量依赖项建模并让每个作业通知依赖作业完成情况更有效。

于 2015-05-04T11:08:42.927 回答
10

您可以获取 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())
    );
}
于 2016-05-28T08:47:38.153 回答
6

要添加@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());
于 2017-06-14T13:47:57.250 回答
5

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

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);  

   }
}

如果您不介意使用 3rd 方库cyclops-react（我是作者）有一组用于 CompletableFutures（以及 Optionals、Streams 等）的实用方法

  List<CompletableFuture<String>> listOfFutures;

  CompletableFuture<ListX<String>> sequence =CompletableFutures.sequence(listOfFutures);
于 2015-09-19T10:29:01.247 回答
1

Disclaimer: This will not completely answer the initial question. It will lack the "fail all if one fails" part. However, I can't answer the actual, more generic question, because it was closed as a duplicate of this one: Java 8 CompletableFuture.allOf(...) with Collection or List. So I will answer here:

How to convert List<CompletableFuture<V>> to CompletableFuture<List<V>> using Java 8's stream API?

Summary: Use the following:

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);
}

Example usage:

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

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

Complete Example:

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);
    }

}
于 2018-05-17T11:57:00.993 回答
0

Javaslang有一个非常方便的FutureAPI。它还允许从期货集合中创建集合的未来。

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-

于 2016-11-22T09:46:49.920 回答
0

您的任务可以很容易地完成，如下所示，

final List<CompletableFuture<Module> futures =...
CompletableFuture.allOf(futures.stream().toArray(CompletableFuture[]::new)).join();
于 2020-03-26T06:36:24.907 回答
0

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

它实现了一个逻辑来返回“M 中至少有 N 个”CompletionStage-s，其中包含允许允许多少错误的策略。所有/任何情况都有方便的方法，加上剩余期货的取消政策，加上代码处理 CompletionStage-s（接口）而不是 CompletableFuture（具体类）。

于 2016-09-10T20:40:05.197 回答