ExecutorService shutdown() не ждет, пока все потоки будут завершены

У меня есть код, в котором одновременно выполняются 4 потока. Я хочу дождаться завершения всех этих 4 потоков. И только после этого продолжать аппоток.

Я пробовал два подхода:

  1. Thread#join()< /a>, этот подход работает так, как ожидалось. Код, который идет после join(), выполняется только после завершения всех потоков.
  2. ExecutorService#shutdown(), этот метод позволяет выполнять код, который идет после shutdown(), даже если не все потоки завершены.

Пример кода:

ExecutorService service = Executors.newFixedThreadPool(cpuCoresNum);

for (int i = 0; i < cpuCoresNum; i++) {

    service.submit(() -> {
        try {
            foo(); // some long execution function
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
}

service.shutdown();

System.out.println("We're done! All threads are finished!");

Мой вопрос:

  • Почему submit() и shutdown() не дождутся завершения всех потоков и не напечатают «Готово! Все потоки завершены!» сразу после вызова service.shutdown();?

java multithreading concurrency executorservice java-threads
person Mike B.    schedule 02.04.2016    source источник

Ответы (3)


arrow_upward
10
arrow_downward

Ответ доступен в ExecutorService.shutdown() Javadoc:

Этот метод не ожидает завершения выполнения ранее отправленных задач. Используйте awaitTermination для этого.

Если вы хотите дождаться завершения работы потоков, у вас есть следующие варианты:

  • получить Future экземпляров, возвращенных submit(), и вызвать get() для каждого Future экземпляра
  • после вызова shutdown в service вызовите awaitTermination в service, пока он не вернет true
  • вместо вызова submit в service добавьте свои экземпляры Runnable в java.util.List и передайте этот список методу invokeAll, вызываемому в service
person Adam Siemion    schedule 02.04.2016
comment
Мои потоки не возвращают никакого значения, поэтому второй вариант кажется более актуальным. - person Mike B.; 02.04.2016
comment
@МайкБ. submit возвращает Future независимо от того, передаете ли вы Runnable или Callable, в обоих случаях вызов get() для Future ожидает завершения работы потока. Прежде чем звонить awaitTermination, вы должны позвонить shutdown. - person Adam Siemion; 02.04.2016
comment
awaitTermination() требует некоторого тайм-аута для указания, но в моем случае я не знаю, сколько времени это займет, поэтому какое значение я должен указать в качестве timeout, чтобы гарантировать, что все потоки будут завершены правильно и не будут вынуждены останавливаться из-за в timeout? Из документации: «Блокируется до тех пор, пока все задачи не завершат выполнение после запроса на завершение работы, или не истечет время ожидания, или текущий поток не будет прерван, в зависимости от того, что произойдет раньше». - person Mike B.; 02.04.2016
comment
Еще один вариант, который я бы добавил для ожидания завершения всех выполнений, — дать каждому потоку ссылку на CountDownLatch, который инициализируется количеством потоков. Вызовите latch.await() в своем основном потоке, а рабочие потоки вызовут latch.countDown(), когда они закончат работу. - person augray; 02.04.2016
comment
@МайкБ. Вы можете передать очень большое число в качестве тайм-аута, например. Long.MAX_VALUE однако всегда полезно указать тайм-аут, так как ваш код всегда может заблокироваться или зайти в тупик, поэтому, я думаю, разработчики JDK сделали этот атрибут обязательным. - person Adam Siemion; 02.04.2016
comment
Ожидание одного Future не гарантирует завершения всех заданий. Так что первая пуля не реальный вариант. - person Holger; 05.04.2016
comment
@Holger, конечно, когда вы вызываете get для одного Future, это только гарантирует, что это Future завершено. Я обновил ответ, чтобы избежать двусмысленности. - person Adam Siemion; 05.04.2016

arrow_upward
2
arrow_downward

Благодаря предложениям @Adam Siemion вот окончательный код:

ExecutorService service = Executors.newFixedThreadPool(cpuCoresNum);

int itNum = 1;

for (int i = 0; i < cpuCoresNum; i++) {

    int treadID = itNum++;

    service.submit(() -> {
        Thread.currentThread().setName("Thread_#" + treadID);
        try {
            foo();
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
}

// wait until all threads will be finished
service.shutdown();
try {
    service.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
    e.printStackTrace();
}
person Mike B.    schedule 02.04.2016
comment
Спасибо, что поделились решением - person Ercan; 10.03.2020

arrow_upward
2
arrow_downward

Рекомендуемый путь со страницы документации оракула ExecutorService:

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

shutdown(): Инициирует упорядоченное завершение работы, при котором ранее отправленные задачи выполняются, но новые задачи не принимаются.

shutdownNow():Попытка остановить все активно выполняющиеся задачи, останавливает обработку ожидающих задач и возвращает список задач, ожидающих выполнения.

В приведенном выше примере, если выполнение ваших задач занимает больше времени, вы можете изменить условие if на условие while

Заменять

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

с

 while(!pool.awaitTermination(60, TimeUnit.SECONDS)) {
     Thread.sleep(60000);
 }
person Ravindra babu    schedule 05.04.2016