How to combine 3 or more CompletionStages?

by

The only way to combine multiple stages that scales well with a growing number of stages, is to use CompletableFuture. If your CompletionStages aren’t CompletableFutures you may still convert them using .toCompletableFuture():

CompletableFuture<A> aCompletionStage = getA().toCompletableFuture();
CompletableFuture<B> bCompletionStage = getB().toCompletableFuture();
CompletableFuture<C> cCompletionStage = getC().toCompletableFuture();
CompletableFuture<D> dCompletionStage = getD().toCompletableFuture();

CompletionStage<Combined> combinedDataCompletionStage = CompletableFuture.allOf(
    aCompletionStage, bCompletionStage, cCompletionStage, dCompletionStage)
    .thenApply(ignoredVoid -> combine(
        aCompletionStage.join(), bCompletionStage.join(),
        cCompletionStage.join(), dCompletionStage.join()) );

This contains more boilerplate than combining two stages via thenCombine but the boilerplate doesn’t grow when adding more stages to it.

Note that even with your original thenCombine approach, you don’t need a Triple, a Pair is sufficient:

CompletionStage<Combined> combinedDataCompletionStage =
    aCompletionStage.thenCombine(bCompletionStage, (Pair::of)).thenCombine(
        cCompletionStage.thenCombine(dCompletionStage, Pair::of),
        (ab, cd) -> combine(ab.getLeft(), ab.getRight(), cd.getLeft(), cd.getRight()));

Still, it doesn’t scale well if you want to combine more stages.

An in-between solution (regarding complexity) might be:

CompletionStage<Combined> combinedDataCompletionStage = aCompletionStage.thenCompose(
    a -> bCompletionStage.thenCompose(b -> cCompletionStage.thenCompose(
        c -> dCompletionStage.thenApply(d -> combine(a, b, c, d)))));

That’s simpler in its structure but still doesn’t scale well with more more stages.

More Related Contents:

  1. Custom thread pool in Java 8 parallel stream
  2. Convert from List to CompletableFuture
  3. Parallel streams, collectors and thread safety
  4. Why should I use concurrent characteristic in parallel stream with collect?
  5. Nested Java 8 parallel forEach loop perform poor. Is this behavior expected?
  6. convert String list in List of object [duplicate]
  7. Random errors when changing series using JFreeChart
  8. Is there an advantage to use a Synchronized Method instead of a Synchronized Block?
  9. What is the difference between Thread.start() and Thread.run()?
  10. org.apache.tomcat.util.bcel.classfile.ClassFormatException: Invalid byte tag in constant pool: 15
  11. java.lang.NullPointerException is thrown using a method-reference but not a lambda expression
  12. Java 8 Streams FlatMap method example
  13. Different types of thread-safe Sets in Java
  14. Java 8 lambdas group list into map
  15. Fetch first element of stream matching the criteria
  16. How do I implement task prioritization using an ExecutorService in Java 5?
  17. Java 8 Stream with batch processing
  18. Java Concurrency: CAS vs Locking [closed]
  19. Why can’t I assign lambda to Object?
  20. Java 8 Collectors.toMap SortedMap
  21. Get value from one Optional or another
  22. Java: ExecutorService that blocks on submission after a certain queue size [duplicate]
  23. How to add elements of a Java8 stream into an existing List
  24. Future task of ExecutorService not truly cancelling
  25. Multithreading Socket communication Client/Server
  26. Why am I getting a parse exception when I try to parse the current LocalDateTime [duplicate]
  27. What is the reason why “synchronized” is not allowed in Java 8 interface methods?
  28. Java 8: Parallel FOR loop
  29. Setting Ideal size of Thread Pool [duplicate]
  30. How to serialize and deserialize Java 8’s java.time types with Gson? [closed]

Leave a Comment