Java流:CompletableFuture的加入流

使用 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 单独。

你可以得到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())
    );
}

像 Misha has pointed out ,你过度使用了 …Async 操作。此外,您正在构建一个复杂的操作链,为一个不能反映您的程序逻辑的依赖项建模:

• 您创建的作业x取决于列表中的第一个和第二个作业
• 您创建了一个作业x+1,它取决于作业x和列表中的第三个作业
• 您创建了一个作业x+2,它取决于作业x+1和列表中的第4个作业
• 您创建了一个作业x+5000,它取决于作业x+499和列表中的最后一个作业

然后,取消(显式地或由于异常)这个递归组成的作业可能会递归执行,并且可能会失败 StackOverflowError 。这取决于实施。

像 already shown by 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);

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

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

要将@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());

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

   }
} 

如果您不介意使用第三方库 cyclops-react (我是作者)有一套CompletableFutures(以及Optionals、Streams等)的实用方法

  List<CompletableFuture<String>> listOfFutures;

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

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

如何转换 List<CompletableFuture<V>> 到 CompletableFuture<List<V>> 使用Java 8的流API?

摘要: 使用以下内容:

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

}

你的任务可以很容易地完成,如下所示,

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

除了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(具体类)。

Javaslang 有一个非常方便的 Future API 。它还允许从期货集合中生成集合的未来。

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-