从列表转换为完全未来

Question

我正在尝试转换列表

如果其中任何一个失败了，那么最后的未来就失败了。我就是这样实施的：

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

如果其中任何一个失败了，那么它就失败了。即使有100万个期货，它也能如期产出。我的问题是：如果有超过5000个期货，如果其中任何一个失败，我会得到一个stackoverflowerr：

线程“pool-1-thread-2611”java中出现异常。java上的lang.StackOverflowerr。util。同时发生的完全的未来。java上的internalComplete（CompletableFuture.java:210）。util。同时发生的可完成的未来$然后合成。在java上运行（CompletableFuture.java:1487）。util。同时发生的完全的未来。在java完成后（CompletableFuture.java:193）。util。同时发生的完全的未来。java上的internalComplete（CompletableFuture.java:210）。util。同时发生的可完成的未来$然后合成。run（CompletableFuture.java:1487）

我做错了什么？

注：上述返回的future在任何future出现故障时立即失效。公认的答案也应该考虑这一点。




Answer 1

匿名用户

您可以获取Spotify的CompletableFutures库，并使用allAsList方法。我认为它的灵感来自番石榴的未来。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())
    );
}

Answer 2

正如米沙所指出的，你过度使用了...异步操作。此外，您正在构建一个复杂的操作链，模拟一个不反映您的程序逻辑的依赖关系：

• 您创建了一个作业x，它取决于列表中的第一个和第二个作业
• 根据作业x和列表中的第三个作业创建作业x 1
• 创建一个作业x2，它取决于作业x1和列表中的第四个作业
• …
• 您创建了一个作业x 5000，它取决于作业x 4999和列表中的最后一个作业

然后，取消（显式地或由于异常）这个递归组合的作业可能会递归地执行，并且可能会失败。这依赖于实现。

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

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

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

Answer 3

使用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())
            );
}

对您的实施有几点意见：

您对的使用。然后是ComposeAsync，。然后应用同步和。thenCombineAsync可能没有达到您的预期。这些。。。Async方法在单独的线程中运行提供给它们的函数。因此，在您的情况下，您正在使添加到列表中的新项在提供的执行器中运行。无需将轻量级操作塞进缓存的线程执行器中。没有充分的理由，不要使用thenxxasync方法。

此外，减少不应用于累积到可变容器中。即使当流是顺序的时，它可能会正常工作，但如果流是并行的，它也会失败。若要执行可变还原，请使用。

如果要在第一次失败后立即完成整个计算，请在顺序中执行以下操作方法：

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。关闭现在（） 紧接着结果。完全例外（ex） 。当然，这假设exec只存在于这一次计算中。如果没有，你将不得不循环并逐个取消剩余的Future。