Остановка ExecutorService

Рассмотрим простейшую программу:

public static void main(String[] args) throws Exception {
	// создаем ExecutorService с фиксированным числом нитей – три
	ExecutorService service = Executors.newFixedThreadPool(3);
 
	// передаем в ExecutorService простое задание типа Runnable
	service.submit(() -> System.out.println("done"));
}

Запустив программу, получаем в консоли ожидаемый вывод:

Но далее не видим привычного для IntellijIDEA вывода:

Его обычно можно увидеть в конце выполнения программы.

Почему так происходит?

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

Остановка ExecutorService

Итак, ExecutorService нужно за собой “закрывать” (останавливать). Сделать это можно двумя способами:

1. void shutdown() — после вызова этого метода ExecutorService больше не принимает новые задания. Все задания, которые раннее были переданы в ExecutorService, продолжат свое выполнение.

   
   public static void main(String[] args) throws Exception {
   ExecutorService service = Executors.newFixedThreadPool(3);
       	service.submit(() -> System.out.println("task 1"));
       	service.submit(() -> System.out.println("task 2"));
       	service.shutdown();
       	// здесь произойдет RejectedExecutionException
       	service.submit(() -> System.out.println("task 3"));
   }
   

2. List<Runnable> shutdownNow() — метод пытается остановить текущие активные задания. Задачи, которые ждали своей очереди, отбрасываются и возвращаются в виде списка Runnable.

   
   public static void main(String[] args) throws Exception {
       ExecutorService service = Executors.newFixedThreadPool(5);
       List.of(1, 2, 3, 4, 5, 6, 7, 8).forEach(i -> service.submit(() -> System.out.println(i)));
       List<Runnable> runnables = service.shutdownNow();
       runnables.forEach(System.out::println);
   }
   

Вывод программы:

Вывод будет отличаться от запуска к запуску. В выводе есть 2 вида строк:

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

• объект типа FutureTask, после вызова у него метода toString(). Это задачи, которые были переданы ExecutorService на выполнение, но не были обработаны.

В этом выводе есть еще один интересный нюанс. Если бы мы жили в идеальном мире, то сначала вывелись бы все числа, а потом объекты типа FutureTask. Но из-за проблем синхронизации строки в выводе перемешаны.

Другие методы

Кроме этого у ExecutorService есть еще несколько методов, связанных с его остановкой:

1. boolean awaitTermination(long timeout, TimeUnit unit) — метод блокирует нить, которая его вызвала. Блокировка прерывается, как только наступает любое из трех событий:

   • после вызова метода shutdown() все активные задания и все задания из очереди были выполнены;
   • закончился таймаут, длительность которого определяется параметрами метода;
   • нить, вызвавшая метод awaitTermination(), была прервана.

   Метод возвращает true, если ExecutorService был остановлен до истечения таймаута, и false, если таймаут истек раньше.

   
   public static void main(String[] args) throws Exception {
   	ExecutorService service = Executors.newFixedThreadPool(2);
   	service.submit(() -> System.out.println("task 1"));
   	service.submit(() -> System.out.println("task 2"));
   	service.submit(() -> System.out.println("task 3"));
   	service.shutdown();
   	System.out.println(service.awaitTermination(1, TimeUnit.MICROSECONDS));
   }
   

2. boolean isShutdown() — возвращает true, если у ExecutorService был вызван метод shutdown() или shutdownNow().

   
   public static void main(String[] args) throws Exception {
   	ExecutorService service = Executors.newFixedThreadPool(2);
   	service.submit(() -> System.out.println("task 1"));
   	service.submit(() -> System.out.println("task 2"));
   	service.submit(() -> System.out.println("task 3"));
   	System.out.println(service.isShutdown());
   	service.shutdown();
   	System.out.println(service.isShutdown());
   }
   

3. boolean isTerminated() — возвращает true, если у ExecutorService был вызван метод shutdown() или shutdownNow() и завершено выполнение всех заданий.

   
   public static void main(String[] args) throws Exception {
       ExecutorService service = Executors.newFixedThreadPool(5);
       List.of(1, 2, 3, 4, 5, 6, 7, 8).forEach(i -> service.submit(() -> System.out.println(i)));
       service.shutdownNow();
       System.out.println(service.isTerminated());
   }
   

Пример кода с использованием рассмотренных методов:

public static void main(String[] args) throws Exception {
   ExecutorService service = Executors.newFixedThreadPool(16);
   Callable<String> task = () -> {
       Thread.sleep(1);
       return "Done";
   };
 
   // добавляем в очередь на выполнение 10 тыс. заданий
   List<Future<String>> futures = IntStream.range(0, 10_000)
           .mapToObj(i -> service.submit(task))
           .collect(Collectors.toList());
   System.out.printf("На выполнение отправлено %d заданий.%n", futures.size());
 
   // пробуем закрыть
   service.shutdown();
   // ждем окончания работы 100 миллисекунд
   if (service.awaitTermination(100, TimeUnit.MILLISECONDS)) {
       System.out.println("Все задания выполнены!");
   } else {
       // принудительно останавливаем
       List<Runnable> notExecuted = service.shutdownNow();
       System.out.printf("Так и не запустилось %d заданий.%n", notExecuted.size());
   }
 
   System.out.printf("Всего выполнено %d заданий.%n", futures.stream().filter(Future::isDone).count());
}

Вывод программы (отличается от запуска к запуску):