ExecutorCompletionService? Why do need one if we have invokeAll?

by

Using a ExecutorCompletionService.poll/take, you are receiving the Futures as they finish, in completion order (more or less). Using ExecutorService.invokeAll, you do not have this power; you either block until are all completed, or you specify a timeout after which the incomplete are cancelled.

static class SleepingCallable implements Callable<String> {

  final String name;
  final long period;

  SleepingCallable(final String name, final long period) {
    this.name = name;
    this.period = period;
  }

  public String call() {
    try {
      Thread.sleep(period);
    } catch (InterruptedException ex) { }
    return name;
  }
}

Now, below I will demonstrate how invokeAll works:

final ExecutorService pool = Executors.newFixedThreadPool(2);
final List<? extends Callable<String>> callables = Arrays.asList(
    new SleepingCallable("quick", 500),
    new SleepingCallable("slow", 5000));
try {
  for (final Future<String> future : pool.invokeAll(callables)) {
    System.out.println(future.get());
  }
} catch (ExecutionException | InterruptedException ex) { }
pool.shutdown();

This produces the following output:

C:\dev\scrap>java CompletionExample
... after 5 s ...
quick
slow

Using CompletionService, we see a different output:

final ExecutorService pool = Executors.newFixedThreadPool(2);
final CompletionService<String> service = new ExecutorCompletionService<String>(pool);
final List<? extends Callable<String>> callables = Arrays.asList(
    new SleepingCallable("slow", 5000),
    new SleepingCallable("quick", 500));
for (final Callable<String> callable : callables) {
  service.submit(callable);
}
pool.shutdown();
try {
  Future<String> future;
  do {
    future = pool.isTerminated() ? service.poll() : service.take();
    if (future != null) {
      System.out.println(future.get());
    }
  } while (future != null);
} catch (ExecutionException | InterruptedException ex) { }

This produces the following output:

C:\dev\scrap>java CompletionExample
... after 500 ms ...
quick
... after 5 s ...
slow

Note the times are relative to program start, not the previous message.

Test out a working example on Replit.

More Related Contents:

  1. How to wait for all threads to finish, using ExecutorService?
  2. ExecutorService that interrupts tasks after a timeout
  3. Synchronization vs Lock
  4. Executors.newCachedThreadPool() versus Executors.newFixedThreadPool()
  5. How to properly use Java Executor?
  6. What are the advantages of using an ExecutorService?
  7. Java’s Fork/Join vs ExecutorService – when to use which?
  8. Impossible to make a cached thread pool with a size limit?
  9. ThreadPoolExecutor Block When Queue Is Full?
  10. Java ExecutorService: awaitTermination of all recursively created tasks
  11. What is adaptive spinning w.r.t lock acquisition?
  12. Run Java Threads sequentially
  13. FixedThreadPool vs CachedThreadPool: the lesser of two evils
  14. WAITING at sun.misc.Unsafe.park(Native Method)
  15. ThreadPoolExecutor Block When its Queue Is Full?
  16. How to pause/resume all threads in an ExecutorService in Java?
  17. Whether to use invokeAll or submit – java Executor service
  18. Is there an advantage to use a Synchronized Method instead of a Synchronized Block?
  19. When and how should I use a ThreadLocal variable?
  20. What is the difference between Thread.start() and Thread.run()?
  21. How do synchronized static methods work in Java and can I use it for loading Hibernate entities?
  22. Why use a ReentrantLock if one can use synchronized(this)?
  23. How do I implement task prioritization using an ExecutorService in Java 5?
  24. More efficient way for pausing loop wanted
  25. Instruction reordering & happens-before relationship [duplicate]
  26. Why should I use concurrent characteristic in parallel stream with collect?
  27. Java: ExecutorService that blocks on submission after a certain queue size [duplicate]
  28. Why doesn’t volatile in java 5+ ensure visibility from another thread?
  29. Future task of ExecutorService not truly cancelling
  30. Multithreading Socket communication Client/Server

Leave a Comment