@Test public void testMerge() throws Exception { ExecutorService es = Executors.newFixedThreadPool(3); try { for (int n : new int[] {1, 2, 3, 997, 1000}) { List<CompletableFuture<Integer>> futures = new ArrayList<>(); for (int i = 1; i <= n; i++) { final int num = i; futures.add(CompletableFuture.supplyAsync(() -> num, es)); } CompletableFuture<Integer> mergedFutures = CompletableFutureUtil.merge(futures, Integer::sum, () -> 0); assertEquals(n * (n + 1) / 2, mergedFutures.get().intValue()); } CompletableFuture<Integer> mergedFutures = CompletableFutureUtil.merge(Collections.emptyList(), Integer::sum, () -> 0); assertEquals(0, mergedFutures.get().intValue()); } finally { es.shutdown(); } }
public static <T> CompletableFuture<T> merge(List<CompletableFuture<T>> futures, BiFunction<T,T,T> mergeFunc, Supplier<T> nothing) { if (futures.isEmpty()) { return CompletableFuture.completedFuture(nothing.get()); } while (futures.size() > 1) { ArrayList<CompletableFuture<T>> mergedFutures = new ArrayList<>(futures.size() / 2); for (int i = 0; i < futures.size(); i += 2) { if (i + 1 == futures.size()) { mergedFutures.add(futures.get(i)); } else { mergedFutures.add(futures.get(i).thenCombine(futures.get(i + 1), mergeFunc)); } } futures = mergedFutures; } return futures.get(0); }
CompletableFutureUtil { public static <T> CompletableFuture<T> merge(List<CompletableFuture<T>> futures, BiFunction<T,T,T> mergeFunc, Supplier<T> nothing) { if (futures.isEmpty()) { return CompletableFuture.completedFuture(nothing.get()); } while (futures.size() > 1) { ArrayList<CompletableFuture<T>> mergedFutures = new ArrayList<>(futures.size() / 2); for (int i = 0; i < futures.size(); i += 2) { if (i + 1 == futures.size()) { mergedFutures.add(futures.get(i)); } else { mergedFutures.add(futures.get(i).thenCombine(futures.get(i + 1), mergeFunc)); } } futures = mergedFutures; } return futures.get(0); } }
CompletableFutureUtil { public static <T> CompletableFuture<T> merge(List<CompletableFuture<T>> futures, BiFunction<T,T,T> mergeFunc, Supplier<T> nothing) { if (futures.isEmpty()) { return CompletableFuture.completedFuture(nothing.get()); } while (futures.size() > 1) { ArrayList<CompletableFuture<T>> mergedFutures = new ArrayList<>(futures.size() / 2); for (int i = 0; i < futures.size(); i += 2) { if (i + 1 == futures.size()) { mergedFutures.add(futures.get(i)); } else { mergedFutures.add(futures.get(i).thenCombine(futures.get(i + 1), mergeFunc)); } } futures = mergedFutures; } return futures.get(0); }  }
CompletableFutureUtil { public static <T> CompletableFuture<T> merge(List<CompletableFuture<T>> futures, BiFunction<T,T,T> mergeFunc, Supplier<T> nothing) { if (futures.isEmpty()) { return CompletableFuture.completedFuture(nothing.get()); } while (futures.size() > 1) { ArrayList<CompletableFuture<T>> mergedFutures = new ArrayList<>(futures.size() / 2); for (int i = 0; i < futures.size(); i += 2) { if (i + 1 == futures.size()) { mergedFutures.add(futures.get(i)); } else { mergedFutures.add(futures.get(i).thenCombine(futures.get(i + 1), mergeFunc)); } } futures = mergedFutures; } return futures.get(0); }  static CompletableFuture<T> merge(List<CompletableFuture<T>> futures,
      BiFunction<T,T,T> mergeFunc, Supplier<T> nothing); }
CompletableFutureUtil { public static <T> CompletableFuture<T> merge(List<CompletableFuture<T>> futures, BiFunction<T,T,T> mergeFunc, Supplier<T> nothing) { if (futures.isEmpty()) { return CompletableFuture.completedFuture(nothing.get()); } while (futures.size() > 1) { ArrayList<CompletableFuture<T>> mergedFutures = new ArrayList<>(futures.size() / 2); for (int i = 0; i < futures.size(); i += 2) { if (i + 1 == futures.size()) { mergedFutures.add(futures.get(i)); } else { mergedFutures.add(futures.get(i).thenCombine(futures.get(i + 1), mergeFunc)); } } futures = mergedFutures; } return futures.get(0); }  static CompletableFuture<T> merge(List<CompletableFuture<T>> futures,
      BiFunction<T,T,T> mergeFunc, Supplier<T> nothing);  }