ExecutorService, как ждать завершения всех задач

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

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

ComputeDTaskреализует работоспособный. Это кажется для правильного выполнения задач, но код падает 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, которые будут возвращать нули (см. Определение Executors.callable(). Вероятно, вы хотите сделать небольшой рефакторинг, чтобы вы могли получить полезный ответ или ссылку на базовый код ComputeDTask, но я могу не скажи из твоего примера.

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

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

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

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

• invokeAll () не желает принимать коллекцию <Callable <t >>

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

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

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

Также вы можете использовать Runtime.availableProcessorsколичество аппаратных потоков, чтобы правильно инициализировать пул потоков.

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

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

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

Есть несколько примеров, показывающих, как использоватьCompletionService в книге Java Concurrency in Practice .

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

Вы можете подождать, пока работа завершится через определенный интервал:

int maxSecondsPerComputeDTask = 20;
try {
    while (!es.awaitTermination(uniquePhrases.size() * maxSecondsPerComputeDTask, TimeUnit.SECONDS)) {
        // consider giving up with a 'break' statement under certain conditions
    }
} catch (InterruptedException e) {
    throw new RuntimeException(e);    
}

Или вы могли бы использовать ExecutorService . submit ( Runnable ) и собирать объекты Future, которые он возвращает, и вызывать get () для каждого по очереди, чтобы дождаться их завершения.

ExecutorService es = Executors.newFixedThreadPool(2);
Collection<Future<?>> futures = new LinkedList<<Future<?>>();
for (DataTable singleTable : uniquePhrases) {
    futures.add(es.submit(new ComputeDTask(singleTable)));
}
for (Future<?> future : futures) {
   try {
       future.get();
   } catch (InterruptedException e) {
       throw new RuntimeException(e);
   } catch (ExecutionException e) {
       throw new RuntimeException(e);
   }
}

InterruptedException чрезвычайно важен для правильной обработки. Это то, что позволяет вам или пользователям вашей библиотеки безопасно завершить долгий процесс.

Просто используйте

latch = new CountDownLatch(noThreads)

В каждой теме

latch.countDown();

и как барьер

latch.await();

Основная причина для IllegalMonitorStateException :

Брошенный, чтобы указать, что поток попытался ждать на мониторе объекта или уведомить другие потоки, ожидающие на мониторе объекта, не имея указанного монитора.

Из вашего кода вы только что вызвали wait () на ExecutorService без владения блокировкой.

Ниже код исправит IllegalMonitorStateException

try 
{
    synchronized(es){
        es.wait(); // Add some condition before you call wait()
    }
} 

Следуйте одному из следующих подходов, чтобы дождаться завершения всех задач, которые были представлены ExecutorService.

1. Перебирайте все Futureзадачи с момента submitвключения ExecutorServiceи проверяйте статус с блокировкой вызова get()на Futureобъекте

2. Использование invokeAll наExecutorService

3. Использование CountDownLatch

4. Используя ForkJoinPool или newWorkStealingPool из Executors(начиная с Java 8)

5. Выключите пул, как рекомендовано на странице документации оракула.

   void shutdownAndAwaitTermination(ExecutorService pool) {
      pool.shutdown(); // Disable new tasks from being submitted
      try {
      // Wait a while for existing tasks to terminate
      if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
          pool.shutdownNow(); // Cancel currently executing tasks
          // Wait a while for tasks to respond to being cancelled
          if (!pool.awaitTermination(60, TimeUnit.SECONDS))
          System.err.println("Pool did not terminate");
      }
   } catch (InterruptedException ie) {
        // (Re-)Cancel if current thread also interrupted
        pool.shutdownNow();
        // Preserve interrupt status
        Thread.currentThread().interrupt();
   }

   Если вы хотите изящно дождаться завершения всех задач, когда вы используете опцию 5 вместо опций с 1 по 4, измените

   if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {

   в

   а, while(condition)который проверяет каждую 1 минуту.

Ты можешь использовать ExecutorService.invokeAll метод, он выполнит все задачи и подождет, пока все потоки не завершат свою задачу.

Здесь завершено Javadoc

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

Вот пример кода с ExecutorService.invokeAll

public class Test {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService service = Executors.newFixedThreadPool(3);
        List<Callable<String>> taskList = new ArrayList<>();
        taskList.add(new Task1());
        taskList.add(new Task2());
        List<Future<String>> results = service.invokeAll(taskList);
        for (Future<String> f : results) {
            System.out.println(f.get());
        }
    }

}

class Task1 implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            Thread.sleep(2000);
            return "Task 1 done";
        } catch (Exception e) {
            e.printStackTrace();
            return " error in task1";
        }
    }
}

class Task2 implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            Thread.sleep(3000);
            return "Task 2 done";
        } catch (Exception e) {
            e.printStackTrace();
            return " error in task2";
        }
    }
}

У меня также есть ситуация, когда у меня есть набор документов для сканирования. Я начну с исходного «начального» документа, который должен быть обработан, этот документ содержит ссылки на другие документы, которые также должны быть обработаны, и так далее.

В моей основной программе я просто хочу написать что-то вроде следующего, где Crawlerконтролирует кучу потоков.

Crawler c = new Crawler();
c.schedule(seedDocument); 
c.waitUntilCompletion()

Та же самая ситуация произошла бы, если бы я хотел перемещаться по дереву; я бы вставил в корневой узел, процессор для каждого узла добавлял бы детей в очередь по мере необходимости, и группа потоков обрабатывала бы все узлы в дереве, пока их больше не было.

Я не смог найти ничего в JVM, что мне показалось немного удивительным. Поэтому я написал класс, ThreadPoolкоторый можно использовать напрямую или подкласс для добавления методов, подходящих для домена, например schedule(Document). Надеюсь, поможет!

ThreadPool Javadoc | специалист

Добавьте все темы в коллекцию и отправьте, используя invokeAll. Если вы можете использовать invokeAllметод ExecutorService, JVM не перейдет к следующей строке, пока все потоки не будут завершены.

Вот хороший пример: invokeAll через ExecutorService

Отправьте свои задачи в Runner, а затем дождитесь вызова метода waitTillDone () следующим образом:

Runner runner = Runner.runner(2);

for (DataTable singleTable : uniquePhrases) {

    runner.run(new ComputeDTask(singleTable));
}

// blocks until all tasks are finished (or failed)
runner.waitTillDone();

runner.shutdown();

Чтобы использовать его, добавьте эту зависимость gradle / maven: 'com.github.matejtymes:javafixes:1.0'

Для получения более подробной информации смотрите здесь: https://github.com/MatejTymes/JavaFixes или здесь: http://matejtymes.blogspot.com/2016/04/executor-that-notifies-you-when-task.html

Простая альтернатива этому - использовать потоки вместе с join. См .: Присоединение Темы

Я просто подожду, пока исполнитель завершит работу с указанным тайм-аутом, который, по вашему мнению, подходит для выполнения задач.

 try {  
         //do stuff here 
         exe.execute(thread);
    } finally {
        exe.shutdown();
    }
    boolean result = exe.awaitTermination(4, TimeUnit.HOURS);
    if (!result)

    {
        LOGGER.error("It took more than 4 hour for the executor to stop, this shouldn't be the normal behaviour.");
    }

Похоже, вам нужно ForkJoinPoolи использовать глобальный пул для выполнения задач.

public static void main(String[] args) {
    // the default `commonPool` should be sufficient for many cases.
    ForkJoinPool pool = ForkJoinPool.commonPool(); 
    // The root of your task that may spawn other tasks. 
    // Make sure it submits the additional tasks to the same executor that it is in.
    Runnable rootTask = new YourTask(pool); 
    pool.execute(rootTask);
    pool.awaitQuiescence(...);
    // that's it.
}

Прелесть в pool.awaitQuiescenceтом, что метод блокирует использование потока вызывающей стороны для выполнения своих задач, а затем возвращает его, когда он действительно пуст.