list< future>到future< list>序列

ggazkfy8  于 2021-07-09  发布在  Java
关注(0)|答案(9)|浏览(531)

我正在努力改变 List<CompletableFuture<X>> 至 CompletableFuture<List<T>> . 当您有许多异步任务并且需要获得所有这些任务的结果时，这非常有用。
如果他们中的任何一个失败了，那么最后的未来就失败了。我就是这样实现的：

  1. public static <T> CompletableFuture<List<T>> sequence2(List<CompletableFuture<T>> com, ExecutorService exec) {
  2.         if(com.isEmpty()){
  3.             throw new IllegalArgumentException();
  4.         }
  5.         Stream<? extends CompletableFuture<T>> stream = com.stream();
  6.         CompletableFuture<List<T>> init = CompletableFuture.completedFuture(new ArrayList<T>());
  7.         return stream.reduce(init, (ls, fut) -> ls.thenComposeAsync(x -> fut.thenApplyAsync(y -> {
  8.             x.add(y);
  9.             return x;
  10.         },exec),exec), (a, b) -> a.thenCombineAsync(b,(ls1,ls2)-> {
  11.             ls1.addAll(ls2);
  12.             return ls1;
  13.         },exec));
  14.     }

要运行它：

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

如果其中任何一个失败了，那么它就失败了。即使有一百万个期货，它也能如期产出。我的问题是：假设有超过5000个期货，如果其中任何一个失败了，我会得到一个 StackOverflowError :
java.util.concurrent.completablefuture.internalcomplete（completablefuture）处的线程“pool-1-thread-2611”java.lang.StackOverflower出现异常。java:210)在java.util.concurrent.completablefuture$thencose.run（completablefuture。java:1487)在java.util.concurrent.completablefuture.postcomplete（completablefuture。java:193)在java.util.concurrent.completablefuture.internalcomplete（completablefuture。java:210)在java.util.concurrent.completablefuture$thencose.run（completablefuture。java:1487)
我做错什么了？
注：以上返回的future在任意future失效时立即失效。公认的答案也应该是这一点。

Javaconcurrencyjava-8completable-future

9条答案

按热度按时间
4ngedf3f

4ngedf3f1#

javaslang有一个非常方便的 Future 应用程序编程接口。它还允许从未来的集合中创造未来的集合。

  1. List<Future<String>> listOfFutures = ... 
  2. 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语言-

赞(0）分享  回复(0）举报  2021-07-09
fcy6dtqo

fcy6dtqo2#

你的任务可以很容易地完成，比如，

  1. final List<CompletableFuture<Module> futures =...
  2. CompletableFuture.allOf(futures.stream().toArray(CompletableFuture[]::new)).join();
赞(0）分享  回复(0）举报  2021-07-09
tpgth1q7

tpgth1q73#

免责声明：这不会完全回答最初的问题。它将缺少“如果一个人失败了，那就全失败”的部分。但是，我无法回答实际的、更一般的问题，因为它是作为这个问题的副本关闭的：java8 completablefuture.allof（…）with collection or list。所以我在这里回答：
如何转换 List<CompletableFuture<V>> 至 CompletableFuture<List<V>> 使用Java8的流api？
摘要：请使用以下命令：

  1. private <V> CompletableFuture<List<V>> sequence(List<CompletableFuture<V>> listOfFutures) {
  2.     CompletableFuture<List<V>> identity = CompletableFuture.completedFuture(new ArrayList<>());
  4.     BiFunction<CompletableFuture<List<V>>, CompletableFuture<V>, CompletableFuture<List<V>>> accumulator = (futureList, futureValue) ->
  5.         futureValue.thenCombine(futureList, (value, list) -> {
  6.                 List<V> newList = new ArrayList<>(list.size() + 1);
  7.                 newList.addAll(list);
  8.                 newList.add(value);
  9.                 return newList;
  10.             });
  12.     BinaryOperator<CompletableFuture<List<V>>> combiner = (futureList1, futureList2) -> futureList1.thenCombine(futureList2, (list1, list2) -> {
  13.         List<V> newList = new ArrayList<>(list1.size() + list2.size());
  14.         newList.addAll(list1);
  15.         newList.addAll(list2);
  16.         return newList;
  17.     });
  19.     return listOfFutures.stream().reduce(identity, accumulator, combiner);
  20. }

用法示例：

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

完整示例：

  1. import java.util.ArrayList;
  2. import java.util.List;
  3. import java.util.concurrent.CompletableFuture;
  4. import java.util.concurrent.Executor;
  5. import java.util.concurrent.ExecutorService;
  6. import java.util.concurrent.Executors;
  7. import java.util.concurrent.ThreadLocalRandom;
  8. import java.util.function.BiFunction;
  9. import java.util.function.BinaryOperator;
  10. import java.util.stream.IntStream;
  12. import static java.util.stream.Collectors.toList;
  14. public class ListOfFuturesToFutureOfList {
  16.     public static void main(String[] args) {
  17.         ListOfFuturesToFutureOfList test = new ListOfFuturesToFutureOfList();
  18.         test.load(10);
  19.     }
  21.     public void load(int numThreads) {
  22.         final ExecutorService executor = Executors.newFixedThreadPool(numThreads);
  24.         List<CompletableFuture<String>> listOfFutures = IntStream.range(0, numThreads)
  25.             .mapToObj(i -> loadData(i, executor)).collect(toList());
  27.         CompletableFuture<List<String>> futureList = sequence(listOfFutures);
  29.         System.out.println("Future complete before blocking? " + futureList.isDone());
  31.         // this will block until all futures are completed
  32.         List<String> data = futureList.join();
  33.         System.out.println("Loaded data: " + data);
  35.         System.out.println("Future complete after blocking? " + futureList.isDone());
  37.         executor.shutdown();
  38.     }
  40.     public CompletableFuture<String> loadData(int dataPoint, Executor executor) {
  41.         return CompletableFuture.supplyAsync(() -> {
  42.             ThreadLocalRandom rnd = ThreadLocalRandom.current();
  44.             System.out.println("Starting to load test data " + dataPoint);
  46.             try {
  47.                 Thread.sleep(500 + rnd.nextInt(1500));
  48.             } catch (InterruptedException e) {
  49.                 e.printStackTrace();
  50.             }
  52.             System.out.println("Successfully loaded test data " + dataPoint);
  54.             return "data " + dataPoint;
  55.         }, executor);
  56.     }
  58.     private <V> CompletableFuture<List<V>> sequence(List<CompletableFuture<V>> listOfFutures) {
  59.         CompletableFuture<List<V>> identity = CompletableFuture.completedFuture(new ArrayList<>());
  61.         BiFunction<CompletableFuture<List<V>>, CompletableFuture<V>, CompletableFuture<List<V>>> accumulator = (futureList, futureValue) ->
  62.             futureValue.thenCombine(futureList, (value, list) -> {
  63.                     List<V> newList = new ArrayList<>(list.size() + 1);
  64.                     newList.addAll(list);
  65.                     newList.add(value);
  66.                     return newList;
  67.                 });
  69.         BinaryOperator<CompletableFuture<List<V>>> combiner = (futureList1, futureList2) -> futureList1.thenCombine(futureList2, (list1, list2) -> {
  70.             List<V> newList = new ArrayList<>(list1.size() + list2.size());
  71.             newList.addAll(list1);
  72.             newList.addAll(list2);
  73.             return newList;
  74.         });
  76.         return listOfFutures.stream().reduce(identity, accumulator, combiner);
  77.     }
  79. }
展开查看全部
赞(0）分享  回复(0）举报  2021-07-09
iyfamqjs

iyfamqjs4#

在completablefuture上使用thencombine的一个序列操作示例

  1. public<T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> com){
  3.     CompletableFuture<List<T>> identity = CompletableFuture.completedFuture(new ArrayList<T>());
  5.     BiFunction<CompletableFuture<List<T>>,CompletableFuture<T>,CompletableFuture<List<T>>> combineToList = 
  6.             (acc,next) -> acc.thenCombine(next,(a,b) -> { a.add(b); return a;});
  8.     BinaryOperator<CompletableFuture<List<T>>> combineLists = (a,b)-> a.thenCombine(b,(l1,l2)-> { l1.addAll(l2); return l1;}) ;  
  10.     return com.stream()
  11.               .reduce(identity,
  12.                       combineToList,
  13.                       combineLists);  
  15.    }
  16. }

如果你不介意使用第三方库的话，cyclops react（我是作者）有一套用于completablefutures（以及optionals、streams等）的实用方法

  1. List<CompletableFuture<String>> listOfFutures;
  3.   CompletableFuture<ListX<String>> sequence =CompletableFutures.sequence(listOfFutures);
展开查看全部
赞(0）分享  回复(0）举报  2021-07-09
pdsfdshx

pdsfdshx5#

你可以得到spotify的 CompletableFutures 图书馆与使用 allAsList 方法。我想它的灵感来自Guava Futures.allAsList 方法。

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

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

  1. public <T> CompletableFuture<List<T>> allAsList(final List<CompletableFuture<T>> futures) {
  2.     return CompletableFuture.allOf(
  3.         futures.toArray(new CompletableFuture[futures.size()])
  4.     ).thenApply(ignored ->
  5.         futures.stream().map(CompletableFuture::join).collect(Collectors.toList())
  6.     );
  7. }
赞(0）分享  回复(0）举报  2021-07-09
xriantvc

xriantvc6#

使用 CompletableFuture.allOf(...) :

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

关于您的实现的一些评论：
你使用 .thenComposeAsync , .thenApplyAsync 以及 .thenCombineAsync 很可能做不到你所期望的。这些 ...Async 方法在单独的线程中运行提供给它们的函数。因此，在您的示例中，您正在使添加到列表中的新项在提供的执行器中运行。无需将轻量级操作填充到缓存的线程执行器中。不要使用 thenXXXXAsync 没有充分理由的方法。
另外， reduce 不应用于积累到可变容器中。即使当流是连续的时它可以正常工作，但是如果流是并行的，它也会失败。要执行可变缩减，请使用 .collect 相反。
如果要在第一次失败后立即完成整个计算，请在 sequence 方法：

  1. CompletableFuture<List<T>> result = CompletableFuture.allOf(com.toArray(new CompletableFuture<?>[0]))
  2.         .thenApply(v -> com.stream()
  3.                 .map(CompletableFuture::join)
  4.                 .collect(Collectors.toList())
  5.         );
  7. com.forEach(f -> f.whenComplete((t, ex) -> {
  8.     if (ex != null) {
  9.         result.completeExceptionally(ex);
  10.     }
  11. }));
  13. return result;

此外，如果要在第一次失败时取消其余操作，请添加 exec.shutdownNow(); 刚好在…之后 result.completeExceptionally(ex); . 当然，这是假设 exec 只存在于这一个计算中。如果没有，你将不得不循环并取消每一个剩余的 Future 个别地。

展开查看全部
赞(0）分享  回复(0）举报  2021-07-09
ezykj2lf

ezykj2lf7#

除了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（具体类）的代码。

赞(0）分享  回复(0）举报  2021-07-09
4ktjp1zp

4ktjp1zp8#

加上@misha接受的答案，可以进一步扩展为收集器：

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

现在您可以：

  1. Stream<CompletableFuture<Integer>> stream = Stream.of(
  2.     CompletableFuture.completedFuture(1),
  3.     CompletableFuture.completedFuture(2),
  4.     CompletableFuture.completedFuture(3)
  5. );
  6. CompletableFuture<List<Integer>> ans = stream.collect(sequenceCollector());
赞(0）分享  回复(0）举报  2021-07-09
ddhy6vgd

ddhy6vgd9#

正如米莎所指出的，你过度使用了 …Async 操作。此外，您正在构建一个复杂的操作链，对一个不反映您的程序逻辑的依赖项进行建模：
创建一个作业x，它取决于列表中的第一个和第二个作业
创建一个job x+1，它取决于job x和列表中的第三个job
创建一个job x+2，它取决于job x+1和列表中的第4个job
…
创建一个作业x+5000，它取决于作业x+4999和列表中的最后一个作业
然后，取消（显式地或由于异常）这个递归组合的作业可能会递归地执行，并且可能会失败 StackOverflowError . 这取决于实现。
正如米莎所说，有一种方法， allOf 它允许你对你的初衷进行建模，定义一个依赖于你列表中所有作业的作业。
然而，值得注意的是，即使这样也没有必要。因为您使用的是一个无限线程池执行器，所以您只需发布一个异步作业，将结果收集到一个列表中即可。不管怎样，要求每个作业的结果意味着等待完成。

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

当线程数量有限且作业可能产生额外的异步作业时，使用组合依赖操作的方法非常重要，以避免等待作业从必须首先完成的作业中窃取线程，但这里的情况都不是这样。
在这种特定情况下，一个作业简单地迭代大量的先决条件作业并在必要时等待，可能比建模大量依赖项并让每个作业通知依赖作业完成情况更有效。

展开查看全部
赞(0）分享  回复(0）举报  2021-07-09
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com