Зупинка 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());
}

Виведення програми (відрізняється в різних запусках):