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

Ответ 1

Используя ExecutorCompletionService.poll/take, вы получаете Future по мере их завершения, в порядке завершения (более или менее). Используя 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

+1 до @veer. Я почувствовал необходимость добавить некоторые примеры, которые он сейчас сделал...

Используя ExecutorCompletionService, вы можете получить уведомление сразу после завершения каждого из ваших заданий. Для сравнения, ExecutorService.invokeAll(...) ждет завершения всех ваших заданий перед возвратом коллекции Future s:

// 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);
}
threadPool.shutdown();
while (!threadPool.isTerminated()) {
    // the take() blocks until any of the jobs complete
    // this joins with the jobs in the order they _finish_
    Future<Result> 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));
}

Ответ 3

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

Ответ 4

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

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

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