Waiting on a list of Future

Question

I have a method which returns a List of futures  List lt Future lt O gt  gt  futures   getFutures      Now I want to wait until either all futures are done processing successfully or any of the tasks whose output is returned by a future throws an exception  Even if one task throws an exception  there is no point in waiting for the other futures   Simple approach would be to  wait         For Future f   futures         try          f get           catch Exception e             TODO catch specific exception           this future threw exception   means somone could not do its task        return                  But the problem here is if  for example  the 4th future throws an exception  then I will wait unnecessarily for the first 3 futures to be available   How to solve this  Will count down latch help in any way  I m unable to use Future isDone because the java doc says  boolean isDone   Returns true if this task completed  Completion may be due to normal termination  an exception  or cancellation -- in all of these cases  this method will return true

User · Answer

Use a CompletableFuture in Java 8         Kick of multiple  asynchronous lookups     CompletableFuture lt User gt  page1   gitHubLookupService findUser  Test1        CompletableFuture lt User gt  page2   gitHubLookupService findUser  Test2        CompletableFuture lt User gt  page3   gitHubLookupService findUser  Test3            Wait until they are all done     CompletableFuture allOf page1 page2 page3  join         logger info  -- gt      page1 get

User · Answer

execute suppliers as future tasks then wait   join for getting results         param functors a supplier s  to execute         return a list of results             private List getResultsInFuture Supplier lt   gt     functors            CompletableFuture   futures   stream functors                   map CompletableFuture  supplyAsync                   collect Collectors toList                     toArray new CompletableFuture functors length            CompletableFuture allOf futures  join            return stream futures  map a- gt                try                   return a get                  catch  InterruptedException   ExecutionException e                      logger error  an error occurred during runtime execution a function  e                   return null                           collect Collectors toList

User · Answer

The CompletionService will take your Callables with the  submit   method and you can retrieve the computed futures with the  take   method   One thing you must not forget is to terminate the ExecutorService by calling the  shutdown   method  Also you can only call this method when you have saved a reference to the executor service so make sure to keep one   Example code - For a fixed number of work items to be worked on in parallel   ExecutorService service   Executors newFixedThreadPool Runtime getRuntime   availableProcessors      CompletionService lt YourCallableImplementor gt  completionService    new ExecutorCompletionService lt YourCallableImplementor gt  service    ArrayList lt Future lt YourCallableImplementor gt  gt  futures   new ArrayList lt Future lt YourCallableImplementor gt  gt      for  String computeMe   elementsToCompute        futures add completionService submit new YourCallableImplementor computeMe         now retrieve the futures after computation  auto wait for it  int received   0   while received  lt  elementsToCompute size       Future lt YourCallableImplementor gt  resultFuture   completionService take      YourCallableImplementor result   resultFuture get     received         important  shutdown your ExecutorService service shutdown      Example code - For a dynamic number of work items to be worked on in parallel   public void runIt        ExecutorService service   Executors newFixedThreadPool Runtime getRuntime   availableProcessors         CompletionService lt CallableImplementor gt  completionService   new ExecutorCompletionService lt CallableImplementor gt  service       ArrayList lt Future lt CallableImplementor gt  gt  futures   new ArrayList lt Future lt CallableImplementor gt  gt            Initial workload is 8 threads     for  int i   0  i  lt  9  i              futures add completionService submit write new CallableImplementor                             boolean finished   false      while   finished            try               Future lt CallableImplementor gt  resultFuture              resultFuture   completionService take                CallableImplementor result   resultFuture get                finished   doSomethingWith result getResult                 result setResult null               result   null              resultFuture   null                After work package has been finished create new work package and add it to futures             futures add completionService submit write new CallableImplementor                catch  InterruptedException   ExecutionException e                  handle interrupted and assert correct thread   work packet count                                       important  shutdown your ExecutorService     service shutdown       public class CallableImplementor implements Callable      boolean result        Override     public CallableImplementor call   throws Exception             business logic goes here         return this             public boolean getResult             return result             public void setResult boolean result            this result   result

User · Answer

You can use a CompletionService to receive the futures as soon as they are ready and if one of them throws an exception cancel the processing  Something like this   Executor executor   Executors newFixedThreadPool 4   CompletionService lt SomeResult gt  completionService           new ExecutorCompletionService lt SomeResult gt  executor      4 tasks for int i   0  i  lt  4  i         completionService submit new Callable lt SomeResult gt             public SomeResult call                               return result                     int received   0  boolean errors   false   while received  lt  4  amp  amp   errors          Future lt SomeResult gt  resultFuture   completionService take      blocks if none available       try            SomeResult result   resultFuture get             received                     do something with the result               catch Exception e                   log          errors   true              I think you can further improve to cancel any still executing tasks if one of them throws an error

User · Answer

In case that you want combine a List of CompletableFutures  you can do this    List lt CompletableFuture lt Void gt  gt  futures   new ArrayList lt  gt            Add futures to this ArrayList of CompletableFutures     CompletableFuture allOf   method demand a variadic arguments    You can use this syntax to pass a List instead CompletableFuture lt Void gt  allFutures   CompletableFuture allOf              futures toArray new CompletableFuture futures size           Wait for all individual CompletableFuture to complete    All individual CompletableFutures are executed in parallel allFutures get      For more details on Future  amp  CompletableFuture  useful links   1  Future  https   www baeldung com java-future  2  CompletableFuture  https   www baeldung com java-completablefuture  3  CompletableFuture  https   www callicoder com java-8-completablefuture-tutorial

User · Answer

I ve got a utility class that contains these    FunctionalInterface public interface CheckedSupplier lt X gt      X get   throws Throwable     public static  lt X gt  Supplier lt X gt  uncheckedSupplier final CheckedSupplier lt X gt  supplier        return    - gt            try               return supplier get              catch  final Throwable checkedException                throw new IllegalStateException checkedException                        Once you have that  using a static import  you can simple wait for all futures like this   futures stream   forEach future - gt  uncheckedSupplier future  get  get       you can also collect all their results like this   List lt MyResultType gt  results   futures stream        map future - gt  uncheckedSupplier future  get  get         collect Collectors toList       Just revisiting my old post and noticing that you had another grief      But the problem here is if  for example  the 4th future throws an exception  then I will wait unnecessarily for the first 3 futures to be available    In this case  the simple solution is to do this in parallel   futures stream   parallel     forEach future - gt  uncheckedSupplier future  get  get       This way the first exception  although it will not stop the future  will break the forEach-statement  like in the serial example  but since all wait in parallel  you won t have to wait for the first 3 to complete

User · Answer

maybe this would help  nothing would replaced with raw thread  yeah   I suggest run each Future guy with a separated thread  they goes parallel   then when ever one of the got error  it just signal the manager Handler class    class Handler        private Thread thisThread  private boolean failed false  private Thread   trds  public void waitFor      thisThread Thread currentThread      List lt Future lt Object gt  gt  futures   getFutures      trds new Thread futures size       for  int i   0  i  lt  trds length  i          RunTask rt new RunTask futures get i   this       trds i  new Thread rt         synchronized  this        for Thread tx trds         tx start                  for Thread tx trds       try  tx join          catch  InterruptedException e          System out println  Job failed    break           if  failed  System out println  Job Done        private List lt Future lt Object gt  gt  getFutures       return null     public synchronized void cancelOther   if failed  return     failed true    for Thread tx trds       tx stop     Deprecated but works here like a boss    thisThread interrupt              class RunTask implements Runnable  private Future f private Handler h  public RunTask Future f Handler h  this f f this h h   public void run    try  f get     beware about state of working  the stop   method throws ThreadDeath Error at any thread state  unless it blocked by some operation   catch Exception e  System out println  Error  stopping other guys      h cancelOther     catch Throwable t  System out println  Oops  some other guy has stopped working              I have to say the above code would error didn t check   but I hope I could explain the solution  please have a try

User · Answer

If you are using Java 8 and don t want to manipulate CompletableFutures  I have written a tool to retrieve results for a List lt Future lt T gt  gt  using streaming  The key is that you are forbidden to map Future  get  as it throws   public final class Futures        private Futures               public static  lt E gt  Collector lt Future lt E gt   Collection lt E gt   List lt E gt  gt  present                 return new FutureCollector lt  gt                private static class FutureCollector lt T gt  implements Collector lt Future lt T gt   Collection lt T gt   List lt T gt  gt                private final List lt Throwable gt  exceptions   new LinkedList lt  gt               Override         public Supplier lt Collection lt T gt  gt  supplier                         return LinkedList  new                      Override         public BiConsumer lt Collection lt T gt   Future lt T gt  gt  accumulator                         return  r  f  - gt                    try                                       r add f get                                       catch  InterruptedException e                                     catch  ExecutionException e                                        exceptions add e getCause                                                          Override         public BinaryOperator lt Collection lt T gt  gt  combiner                         return  l1  l2  - gt                    l1 addAll l2                   return l1                                     Override         public Function lt Collection lt T gt   List lt T gt  gt  finisher                         return l - gt                     List lt T gt  ret   new ArrayList lt  gt  l                   if   exceptions isEmpty                        throw new AggregateException exceptions  ret                    return ret                                      Override         public Set lt java util stream Collector Characteristics gt  characteristics                         return java util Collections emptySet                      This needs an AggregateException that works like C  s  public class AggregateException extends RuntimeException                              private static final long serialVersionUID   -4477649337710077094L       private final List lt Throwable gt  causes      private List lt   gt  successfulElements       public AggregateException List lt Throwable gt  causes  List lt   gt  l                this causes   causes          successfulElements   l             public AggregateException List lt Throwable gt  causes                this causes   causes              Override     public synchronized Throwable getCause                 return this             public List lt Throwable gt  getCauses                 return causes             public List lt   gt  getSuccessfulElements                 return successfulElements             public void setSuccessfulElements List lt   gt  successfulElements                this successfulElements   successfulElements             This component acts exactly as C  s Task WaitAll  I am working on a variant that does the same as CompletableFuture allOf  equivalento to Task WhenAll   The reason why I did this is that I am using Spring s ListenableFuture and don t want to port to CompletableFuture despite it is a more standard way

User · Answer

You can use an ExecutorCompletionService  The documentation even has an example for your exact use-case      Suppose instead that you would like to use the first non-null result of the set of tasks  ignoring any that encounter exceptions  and cancelling all other tasks when the first one is ready    void solve Executor e  Collection lt Callable lt Result gt  gt  solvers  throws InterruptedException       CompletionService lt Result gt  ecs   new ExecutorCompletionService lt Result gt  e       int n   solvers size        List lt Future lt Result gt  gt  futures   new ArrayList lt Future lt Result gt  gt  n       Result result   null      try           for  Callable lt Result gt  s   solvers              futures add ecs submit s            for  int i   0  i  lt  n    i                try                   Result r   ecs take   get                    if  r    null                        result   r                      break                                  catch  ExecutionException ignore                                  finally           for  Future lt Result gt  f   futures              f cancel true              if  result    null          use result       The important thing to notice here is that ecs take   will get the first completed task  not just the first submitted one  Thus you should get them in the order of finishing the execution  or throwing an exception

User · Answer

If you are using Java 8 then you can do this easier with CompletableFuture and CompletableFuture allOf  which applies the callback only after all supplied CompletableFutures are done      Waits for  all  futures to complete and returns a list of results     If  any  future completes exceptionally then the resulting future will also complete exceptionally   public static  lt T gt  CompletableFuture lt List lt T gt  gt  all List lt CompletableFuture lt T gt  gt  futures        CompletableFuture   cfs   futures toArray new CompletableFuture futures size           return CompletableFuture allOf cfs               thenApply ignored - gt  futures stream                                        map CompletableFuture  join                                       collect Collectors toList

User · Answer

import java util ArrayList  import java util Iterator  import java util List  import java util concurrent ExecutionException  import java util concurrent ExecutorService  import java util concurrent Executors  import java util concurrent Future  import java util stream Collectors  import java util stream Stream   public class Stack2          public static void waitFor List lt Future lt   gt  gt  futures            List lt Future lt   gt  gt  futureCopies   new ArrayList lt Future lt   gt  gt  futures    contains features for which status has not been completed         while   futureCopies isEmpty       worst case  all task worked without exception  then this method should wait for all tasks             Iterator lt Future lt   gt  gt  futureCopiesIterator   futureCopies iterator                while  futureCopiesIterator hasNext                      Future lt   gt  future   futureCopiesIterator next                    if  future isDone       already done                     futureCopiesIterator remove                        try                           future get      no longer waiting                       catch  InterruptedException e                              ignore                           only happen when current Thread interrupted                       catch  ExecutionException e                            Throwable throwable   e getCause      real cause of exception                         futureCopies forEach f - gt  f cancel true     cancel other tasks that not completed                         return                                                                            public static void main String   args            ExecutorService executorService   Executors newFixedThreadPool 3            Runnable runnable1   new Runnable                 public void run                    try                       Thread sleep 5000                     catch  InterruptedException e                                                       Runnable runnable2   new Runnable                 public void run                    try                       Thread sleep 4000                     catch  InterruptedException e                                                         Runnable fail   new Runnable                 public void run                    try                       Thread sleep 1000                       throw new RuntimeException  bla bla bla                      catch  InterruptedException e                                                        List lt Future lt   gt  gt  futures   Stream of runnable1 fail runnable2                   map executorService  submit                   collect Collectors toList              double start   System nanoTime            waitFor futures           double end    System nanoTime  -start  1e9          System out println end    seconds

[java] Waiting on a list of Future

Examples related to java

Examples related to multithreading

Examples related to future