CODEWITHSUNDEEP
javamultithreadingconcurrencyexecutorservicejava.util.concurrent
Cratylus
Cratylus

Reputation: 54074

ExecutorCompletionService? Why do need one if we have invokeAll?

If we use an ExecutorCompletionService we can submit a series of tasks as Callables and get the result interacting with the CompletionService as a queue.

But there is also the invokeAll of ExecutorService that accepts a Collection of tasks and we get a list of Future to retrieve the results.

As far as I can tell, there is no benefit in using one or over the other (except that we avoid a for loop using an invokeAll that we would have to submit the tasks to the CompletionService) and essentially they are the same idea with a slight difference.

So why are there 2 different ways to submit a series of tasks? Am I correct that performance wise they are equivalent? Is there a case that one is more suitable than the other? I can't think of one.

Upvotes: 51

Views: 35902

Answers (4)

Gray
Gray

Reputation: 116878

So why are there 2 different ways to submit a series of tasks? Am I correct that performance wise they are equivalent? Is there a case that one is more suitable than the other? I can't think of one.

By using an ExecutorCompletionService, you can get immediately notified when each of your jobs completes. In comparison, ExecutorService.invokeAll(...) waits for all of your jobs to complete before returning the collection of Futures. This means that (for example), if all but one job completes in 10 minutes but 1 job takes 30 minutes, you will get no results for 30 minutes.

// this waits until _all_ of the jobs complete
List<Future<Object>> futures = threadPool.invokeAll(...);

Instead, when you use a ExecutorCompletionService, you will be able to get the jobs as soon as each of them completes which allows you to (for example) send them on for processing into another thread pool, log results immediately, etc..

ExecutorService threadPool = Executors.newFixedThreadPool(2);
ExecutorCompletionService<Result> compService
      = new ExecutorCompletionService<Result>(threadPool);
for (MyJob job : jobs) {
    compService.submit(job);
}
// shutdown the pool but the jobs submitted continue to run
threadPool.shutdown();
while (true) {
    Future<Result> future;
    // if pool done (all jobs finish after shutdown) then poll() else take()
    if (threadPool.isTerminated()) {
        future = compService.poll();
        if (future == null) {
            break;
        }
    } else {
        // the take() blocks until any of the jobs complete
        // this joins with the jobs in the order they _finish_
        future = compService.take();
    }
    // this get() won't block
    Result result = future.get();
    // you can then put the result in some other thread pool or something
    // to immediately start processing it
    someOtherThreadPool.submit(new SomeNewJob(result));
}

Upvotes: 21

obataku
obataku

Reputation: 29646

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.

Upvotes: 96

Dungeon Hunter
Dungeon Hunter

Reputation: 20603

Comparing by Considering only the Order of Results:

When we use CompletionService whenever a job submitted is finished the result will be pushed to the queue (Completion Order). Then the order of the submitted jobs and the returned results is no more same. So if you are concerned about the order which tasks got executed use CompletionService

Where As invokeAll returns a list of Futures representing the tasks, in the same sequential order as produced by the iterator for the given task list, each of which has completed.

Upvotes: 1

esaj
esaj

Reputation: 16035

I haven't ever actually used ExecutorCompletionService, but I think the case where this could be more useful than "normal" ExecutorService would be when you want to receive the Futures of completed tasks in completion order. With invokeAll, you just get a list that can contain a mix of incomplete and completed tasks at any given time.

Upvotes: 3

Related Questions