Список ‹Future› в будущее ‹List› последовательность

Используйте 1_:

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 индивидуально.

Как указал Миша, вы злоупотребляете …Async операциями. Кроме того, вы составляете сложную цепочку операций, моделирующую зависимость, не отражающую логику вашей программы:

• вы создаете задание x, которое зависит от первого и второго задания в вашем списке
• вы создаете задание x + 1, которое зависит от задания x и третьего задания в вашем списке
• вы создаете задание x + 2, которое зависит от задания x + 1 и 4-го задания в вашем списке
• …
• вы создаете вакансию x + 5000, которая зависит от вакансии x + 4999 и последней вакансии в вашем списке

Затем отмена (явно или из-за исключения) этого рекурсивно составленного задания может выполняться рекурсивно и может завершиться ошибкой с StackOverflowError. Это зависит от реализации.

Как уже было показано Мишей, существует метод _ 3_, который позволяет моделировать ваш исходное намерение определить одну работу, которая зависит от всех заданий из вашего списка.

Однако стоит отметить, что даже в этом нет необходимости. Поскольку вы используете неограниченный исполнитель пула потоков, вы можете просто опубликовать асинхронное задание, собирающее результаты в список, и все готово. Ожидание завершения подразумевается, в любом случае запрашивая результат каждого задания.

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

Использование методов для составления зависимых операций важно, когда количество потоков ограничено и задания могут порождать дополнительные асинхронные задания, чтобы избежать того, чтобы ожидающие задания крадут потоки из заданий, которые должны завершиться первыми, но это не тот случай.

В этом конкретном случае одно задание, просто повторяющее это большое количество предварительных заданий и ожидание, если необходимо, может быть более эффективным, чем моделирование этого большого количества зависимостей и получение каждым заданием уведомления зависимого задания о завершении.

Вы можете получить библиотеку Spotify CompletableFutures и использовать _ 2_ метод. Думаю, он вдохновлен _ 3_.

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

Пример операции последовательности с использованием 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 (я автор) имеет набор служебных методов для CompletableFutures (и Optionals, Streams и т. д.)

  List<CompletableFuture<String>> listOfFutures;

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

Чтобы добавить к принятому ответу @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());

Заявление об ограничении ответственности: это не дает полного ответа на первоначальный вопрос. В нем не будет части «все выйдет из строя, если один выйдет из строя». Однако я не могу ответить на настоящий, более общий вопрос, потому что он был закрыт как дубликат этого: Java 8 CompletableFuture.allOf (...) с коллекцией или списком. Так что отвечу здесь:

Как преобразовать List<CompletableFuture<V>> в CompletableFuture<List<V>> с помощью потокового API Java 8?

Резюме: используйте следующее:

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

}

В дополнение к библиотеке 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 с политикой допустимого количества ошибок. Существуют удобные методы для всех / любых случаев, плюс политика отмены для оставшихся фьючерсов, плюс код имеет дело с CompletionStage-s (интерфейс), а не с CompletableFuture (конкретный класс).

Javaslang имеет очень удобный _ 1_ 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-

Ваша задача может быть легко выполнена, например,

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