Руководство по Java ExecutorService

Автор оригинала: baeldung.

1. Обзор

ExecutorService – это API JDK, который упрощает выполнение задач в асинхронном режиме. Вообще говоря, ExecutorService автоматически предоставляет пул потоков и API для назначения ему задач.

Дальнейшее чтение:

Руководство по фреймворку Fork/Join в Java

Обзор java.util.concurrent

Руководство по java.util.concurrent.Блокировки

2. Создание экземпляра ExecutorService

2.1. Заводские методы класса Исполнителей

Самый простой способ создать ExecutorService – это использовать один из заводских методов класса Executors .

Например, в следующей строке кода будет создан пул потоков с 10 потоками:

ExecutorService executor = Executors.newFixedThreadPool(10);

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

2.2. Непосредственно создайте службу ExecutorService

Поскольку ExecutorService является интерфейсом, можно использовать экземпляр любой его реализации. В пакете java.util.concurrent есть несколько реализаций на выбор, или вы можете создать свою собственную.

Например, класс ThreadPoolExecutor имеет несколько конструкторов, которые мы можем использовать для настройки executorservice и его внутреннего пула:

ExecutorService executorService = 
  new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,   
  new LinkedBlockingQueue());

Вы можете заметить, что приведенный выше код очень похож на исходный код заводского метода newSingleThreadExecutor(). В большинстве случаев подробная ручная настройка не требуется.

3. Назначение задач исполнителю

ExecutorService может выполнять Выполняемые и вызываемые задачи. Чтобы все было просто в этой статье, будут использоваться две примитивные задачи. Обратите внимание, что мы используем здесь лямбда-выражения вместо анонимных внутренних классов:

Runnable runnableTask = () -> {
    try {
        TimeUnit.MILLISECONDS.sleep(300);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
};

Callable callableTask = () -> {
    TimeUnit.MILLISECONDS.sleep(300);
    return "Task's execution";
};

List> callableTasks = new ArrayList<>();
callableTasks.add(callableTask);
callableTasks.add(callableTask);
callableTasks.add(callableTask);

Мы можем назначить задачи ExecutorService , используя несколько методов, включая execute () , который унаследован от интерфейса Executor , а также submit () , invokeAny() и invokeAll() .

Метод execute() является void и не дает никакой возможности получить результат выполнения задачи или проверить состояние задачи (выполняется ли она):

executorService.execute(runnableTask);

submit() отправляет Вызываемую или выполняемую задачу в ExecutorService и возвращает результат типа Future :

Future future = 
  executorService.submit(callableTask);

вызов Any() назначает набор задач ExecutorService , вызывая запуск каждой из них, и возвращает результат успешного выполнения одной задачи (если было успешное выполнение):

String result = executorService.invokeAny(callableTasks);

invokeAll() назначает коллекцию задач ExecutorService , вызывая запуск каждой из них, и возвращает результат выполнения всех задач в виде списка объектов типа Future :

List> futures = executorService.invokeAll(callableTasks);

Прежде чем идти дальше, нам нужно обсудить еще два вопроса: завершение работы ExecutorService и работа с типами Future return.

4. Завершение работы службы ExecutorService

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

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

С другой стороны, приложение может дойти до конца, но не будет остановлено, потому что ожидание ExecutorService приведет к продолжению работы JVM.

Чтобы правильно завершить работу ExecutorService , у нас есть API shutdown() и shutdownNow () .

Метод shutdown() не вызывает немедленного уничтожения ExecutorService . Это приведет к тому, что ExecutorService перестанет принимать новые задачи и завершит работу после того, как все запущенные потоки завершат свою текущую работу:

executorService.shutdown();

Метод shutdownNow() пытается немедленно уничтожить ExecutorService , но это не гарантирует, что все запущенные потоки будут остановлены одновременно:

List notExecutedTasks = executorService.shutDownNow();

Этот метод возвращает список задач, ожидающих обработки. Разработчик должен решить, что делать с этими задачами.

Один хороший способ закрыть ExecutorService (который также рекомендуется Oracle ) – использовать оба этих метода в сочетании с методом awaitTermination() :

executorService.shutdown();
try {
    if (!executorService.awaitTermination(800, TimeUnit.MILLISECONDS)) {
        executorService.shutdownNow();
    } 
} catch (InterruptedException e) {
    executorService.shutdownNow();
}

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

5. Будущий Интерфейс

Методы submit() и invoke All() возвращают объект или коллекцию объектов типа Feature , что позволяет нам получить результат выполнения задачи или проверить состояние задачи (выполняется ли она).

Интерфейс Future предоставляет специальный метод блокировки get() , который возвращает фактический результат выполнения Вызываемой задачи или null в случае выполняемой задачи:

Future future = executorService.submit(callableTask);
String result = null;
try {
    result = future.get();
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

Вызов метода get() во время выполнения задачи приведет к блокировке выполнения до тех пор, пока задача не будет выполнена должным образом и результат не будет доступен.

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

String result = future.get(200, TimeUnit.MILLISECONDS);

Если период выполнения больше указанного (в данном случае 200 миллисекунд), a TimeoutException будет выброшено.

Мы можем использовать метод isDone () , чтобы проверить, обработана ли уже назначенная задача или нет.

Интерфейс Future также предусматривает отмену выполнения задачи с помощью метода cancel() и проверку отмены с помощью метода isCancelled() :

boolean canceled = future.cancel(true);
boolean isCancelled = future.isCancelled();

6. Интерфейс ScheduledExecutorService

ScheduledExecutorService выполняет задачи с некоторой предопределенной задержкой и/или периодически.

Еще раз, лучший способ создать экземпляр ScheduledExecutorService – это использовать заводские методы класса Executors .

Для этого раздела мы используем ScheduledExecutorService с одним потоком:

ScheduledExecutorService executorService = Executors
  .newSingleThreadScheduledExecutor();

Чтобы запланировать выполнение одной задачи после фиксированной задержки, используйте метод scheduled() службы ScheduledExecutorService .

Два метода scheduled() позволяют выполнять Выполняемые или Вызываемые задачи:

Future resultFuture = 
  executorService.schedule(callableTask, 1, TimeUnit.SECONDS);

Метод scheduleAtFixedRate() позволяет периодически запускать задачу после фиксированной задержки. Приведенный выше код задерживается на одну секунду перед выполнением вызываемой задачи .

Следующий блок кода запустит задачу после начальной задержки в 100 миллисекунд. И после этого он будет выполнять одну и ту же задачу каждые 450 миллисекунд:

Future resultFuture = service
  .scheduleAtFixedRate(runnableTask, 100, 450, TimeUnit.MILLISECONDS);

Если процессору требуется больше времени для выполнения назначенной задачи, чем параметру period метода scheduleAtFixedRate () , то ScheduledExecutorService будет ждать завершения текущей задачи перед запуском следующей.

Если необходимо иметь фиксированную задержку между итерациями задачи, следует использовать scheduleWithFixedDelay () .

Например, следующий код гарантирует 150-миллисекундную паузу между окончанием текущего выполнения и началом другого:

service.scheduleWithFixedDelay(task, 100, 150, TimeUnit.MILLISECONDS);

В соответствии с контрактами методов scheduleAtFixedRate() и scheduleWithFixedDelay () , период выполнения задачи завершится при завершении службы ExecutorService или если во время выполнения задачи возникнет исключение .

7. ExecutorService vs Fork/Join

После выпуска Java 7 многие разработчики решили заменить фреймворк ExecutorService фреймворком fork/join.

Однако это не всегда правильное решение. Несмотря на простоту и частое повышение производительности, связанное с fork/join, это снижает контроль разработчика над параллельным выполнением.

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

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

8. Заключение

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

Давайте подведем их итоги:

Сохранение неиспользуемого ExecutorService в активном состоянии : См .Подробное объяснение в разделе 4 о том, как закрыть ExecutorService .

Неправильная емкость пула потоков при использовании пула потоков фиксированной длины : Очень важно определить, сколько потоков потребуется приложению для эффективного выполнения задач. Слишком большой пул потоков приведет к ненужным накладным расходам только для создания потоков, которые в основном будут находиться в режиме ожидания. Слишком немногие могут заставить приложение казаться невосприимчивым из-за длительных периодов ожидания задач в очереди.

Вызов метода Future ‘s get() после отмены задачи : Попытка получить результат уже отмененной задачи вызывает исключение CancellationException .

Неожиданно длительная блокировка с помощью Future ‘s get() method : Мы должны использовать тайм-ауты, чтобы избежать неожиданных ожиданий.

Как всегда, код для этой статьи доступен в репозитории GitHub .