ExecutorService, how to wait for all tasks to finish

Какой самый простой способ дождаться завершения всех задач ExecutorService? Моя задача в первую очередь вычислительная, поэтому я просто хочу запустить большое количество заданий - по одному на каждом ядре. Прямо сейчас моя настройка выглядит следующим образом:

ExecutorService es = Executors.newFixedThreadPool(2);
for (DataTable singleTable : uniquePhrases) {   
    es.execute(new ComputeDTask(singleTable));
}
try{
    es.wait();
} 
catch (InterruptedException e){
    e.printStackTrace();
}

ComputeDTask реализует runnable . Кажется, что задачи выполняются правильно, но код вылетает при wait() с IllegalMonitorStateException. Это странно, потому что я поиграл с некоторыми игрушечными примерами, и, похоже, это сработало.

uniquePhrases содержит несколько десятков тысяч элементов. Должен ли я использовать другой метод? Я ищу что-то максимально простое.

Самый простой подход - использовать ExecutorService.invokeAll() который делает то, что вы хотите, однострочно. Говоря вашим языком, вам нужно будет изменить или обернуть ComputeDTask для реализации Callable<>, что может дать вам немного больше гибкости. Вероятно, в вашем приложении есть осмысленная реализация Callable.call(), но вот способ обернуть ее, если не используете Executors.callable().

ExecutorService es = Executors.newFixedThreadPool(2);
List<Callable<Object>> todo = new ArrayList<Callable<Object>>(singleTable.size());

for (DataTable singleTable: uniquePhrases) { 
    todo.add(Executors.callable(new ComputeDTask(singleTable))); 
}

List<Future<Object>> answers = es.invokeAll(todo);

Как указывали другие, вы могли бы использовать версию с таймаутом invokeAll(), если это уместно. В этом примере answers будет содержать набор Futures, которые будут возвращать null (см. Определение Executors.callable(). Вероятно, то, что вы хотите сделать, это небольшой рефакторинг, чтобы вы могли получить полезный ответ или ссылку на базовый ComputeDTask, но я не могу сказать из вашего примера.

Если это непонятно, обратите внимание, что invokeAll() не вернется, пока не будут выполнены все задачи. (т. Е. Все Futureы в вашей answers коллекции будут сообщать .isDone(), если их спросят.) Это позволяет избежать ручного завершения работы, ожидания завершения и т.д. и позволяет вам повторно использовать это ExecutorService аккуратно для нескольких циклов, при желании.

Есть несколько связанных вопросов по SO:

• Как дождаться завершения всех потоков




• Возвращает значения из потоков java




• invokeAll() не желает принимать коллекцию<Вызываемый<t>>




• Нужно ли синхронизировать?

Ничто из этого не относится строго к вашему вопросу, но они немного проясняют, как люди думают, что Executor/ExecutorService следует использовать.

Если вы хотите дождаться завершения всех задач, используйте shutdown метод вместо wait. Затем выполните его с помощью awaitTermination.

Кроме того, вы можете использовать Runtime.availableProcessors для получения количества аппаратных потоков, чтобы вы могли правильно инициализировать свой threadpool.

Если ожидание завершения всех задач в ExecutorService не совсем ваша цель, а скорее ожидание завершения определенного пакета задач, вы можете использовать CompletionService — в частности, an ExecutorCompletionService.

Идея состоит в том, чтобы создать ExecutorCompletionService обертку Executor, отправить некоторое известное количество задач через CompletionService, затем извлечь это такое же количество результатов из очереди завершения, используя либо take() (которая блокирует), либо poll() (которая не выполняет). После того, как вы нарисовали все ожидаемые результаты, соответствующие отправленным вами задачам, вы знаете, что все они выполнены.

Позвольте мне заявить об этом еще раз, потому что из интерфейса это не очевидно: вы должны знать, сколько вещей вы помещаете в CompletionService , чтобы знать, сколько вещей нужно попытаться извлечь. Это особенно важно для take() метода: вызовите его слишком часто по одному разу, и он заблокирует ваш вызывающий поток до тех пор, пока какой-нибудь другой поток не отправит другое задание тому же CompletionService.

В книге "Параллелизм Java на практике" CompletionService есть несколько примеров, показывающих, как использовать.

Если вы хотите дождаться завершения выполнения службы-исполнителя, вызовите shutdown(), а затем, awaitTermination(units, UnitType), например awaitTermination(1, MINUTE). ExecutorService не блокируется на своем собственном мониторе, поэтому вы не можете использовать wait и т.д.