ExecutorCompletionService? Зачем он нужен, если у нас есть invokeAll?

Используя ExecutorCompletionService.poll/take, вы получаете Futures по мере их завершения в порядке завершения (более или менее). Используя ExecutorService.invokeAll, у вас нет этой силы; вы либо блокируете, пока все не будут завершены, либо указываете тайм-аут, по истечении которого незавершенные аннулируются.

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;
  }
}

Ниже я продемонстрирую, как работает invokeAll:

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();

Это дает следующий результат:

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

Используя CompletionService, мы видим другой результат:

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 {
  while (!pool.isTerminated()) {
    final Future<String> future = service.take();
    System.out.println(future.get());
  }
} catch (ExecutionException | InterruptedException ex) { }

Это дает следующий результат:

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

Обратите внимание, что время указано относительно запуска программы, а не предыдущего сообщения.

Полный код можно найти здесь.

Итак, почему существует 2 разных способа отправить серию задач? Правильно ли я, что по производительности они эквивалентны? Есть ли случай, когда один подходит больше, чем другой? Я не могу вспомнить ни одного.

Используя ExecutorCompletionService, вы можете немедленно получать уведомления, когда каждая ваша работа завершена. Для сравнения, ExecutorService.invokeAll(...) ожидает завершения всех ваших заданий перед возвратом коллекции Futures. Это означает, что (например), если все задания, кроме одного, выполняются за 10 минут, а одно задание занимает 30 минут, вы не получите результатов в течение 30 минут.

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

Вместо этого, когда вы используете ExecutorCompletionService, вы сможете получать задания, как только каждое из них завершится, что позволяет вам (например) отправлять их для обработки в другой пул потоков, немедленно регистрировать результаты и т. Д.

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 has terminated (all jobs finished 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));
}

На самом деле я никогда не использовал ExecutorCompletionService, но думаю, что случай, когда это могло бы быть более полезным, чем «нормальный» ExecutorService, был бы тогда, когда вы хотите получить Futures завершенных задач в порядке завершения. С invokeAll вы просто получаете список, который может содержать как незавершенные, так и завершенные задачи в любой момент времени.

Сравнение только с учетом порядка результатов:

Когда мы используем CompletionService всякий раз, когда отправленное задание завершено, результат будет помещен в очередь (Порядок завершения). Тогда порядок отправленных заданий и возвращаемых результатов больше не будет одинаковым. Поэтому, если вас беспокоит порядок выполнения задач, используйте CompletionService

Где As invokeAll возвращает список Futures, представляющих задачи, в том же последовательном порядке, что и итератор для данного списка задач, каждая из которых завершена.