Многопоточное программирование в Java - Тимур Машнин Страница 9

Вы можете использовать метод isCancelled, чтобы проверить задача отменена или нет.

Кроме того, после отмены задачи, метод isDone всегда будет возвращать true.

Метод future.get блокирует и ждет завершения задачи.

Если, вы вызываете удаленную службу в задаче, а удаленная служба отключена, то future.get заблокирует приложение навсегда.

Чтобы это предотвратить, вы можете добавить в метод get время ожидания.

Также, как и с объектом Runnable, с помощью объекта Callable, вы можете одновременно создать сразу несколько задач и отправить их на выполнение.

Обратите внимание, что, во-первых, здесь мы создаем фиксированный пул из двух потоков.

Во-вторых, мы используем метод invokeAll, чтобы выполнить несколько задач, передавая коллекцию объектов Callable.

Метод invokeAll возвращает список объектов Future.

Метод invokeAll ожидает, пока все результаты не будут вычислены до их возврата. То есть метод invokeAll возвращает список объектов Future, для которых isDone true.

В отличие от метода submit, который возвращает сразу, метод invokeAll ожидает завершения задач.

Так как метод invokeAll возвращает коллекцию, ее можно обработать с помощью Stream API.

В отличие от метода invokeAll, метод invokeAny передает на выполнение список задач и ожидает выполнение любой из них, то есть наиболее быстрой.

Для выполнения задачи либо периодически, либо после указанной задержки, можно использовать ScheduledExecutorService.

Здесь метод schedule принимает объект Runnable, задержку и единицу задержки, и определяет выполнение задачи через 5 секунд с момента отправки.

Метод scheduleAtFixedRate принимает объект Runnable, начальную задержку, период выполнения и единицу времени, и запускает выполнение заданной задачи после указанной задержки и затем периодически выполняет ее с указанным интервалом.

Имейте в виду, что scheduleAtFixedRate не учитывает фактическую продолжительность задачи. Поэтому, если вы укажете период в одну секунду, но для выполнения задачи потребуется 2 секунды, тогда пул потоков начнет быстро перегружаться.

Поэтому для периодического запуска задачи лучше использовать метод scheduleWithFixedDelay.

Разница заключается в том, что период ожидания применяется от конца задачи до начала следующей задачи.

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

Чтобы удалить задачу без остановки ScheduledExecutorService, можно использовать результат ScheduledFuture, возвращаемый методами интерфейса ScheduledExecutorService, и применить к этому результату метод cancel.

Для возврата результата, в метод schedule можно передать не объект Runnable, а объект Callable, и применить к результату ScheduledFuture метод get, так как ScheduledFuture расширяет уже знакомый нам интерфейс Future.

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

Создание ExecutorService с помощью newFixedThreadPool эквивалентно использованию ThreadPoolExecutor с очередью LinkedBlockingQueue.

Эта очередь в данном случае неограниченна и упорядочивает элементы FIFO (первый пришел-первый ушел).

Новые элементы вставляются в хвост очереди, а элементы в начале очереди обрабатываются.

Однако использование ThreadPoolExecutor позволяет, например, увеличить максимальный размер пула потоков и ограничить очередь.

При этом если пул потоков не достиг еще своего основного размера, он создает новые потоки.

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

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

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

Здесь мы с помощью ThreadPoolExecutor создаем пул из двух потоков с очередью SynchronousQueue.

Эта очередь с нулевой емкостью.

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

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

Такой режим может быть полезен для незамедлительной обработки задач в фоне.

Фреймворк Fork-Join

Фреймворк Executor предоставляет несколько интерфейсов, таких как ExecutorService, для создания различных типов пулов потоков и выполнения ими задач.

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

В Java 7 добавлен класс ForkJoinPool, реализующий интерфейс ExecutorService и специально предназначенный для выполнения ForkJoinTask.

ForkJoinPool отличается от других видов ExecutorService главным образом благодаря реализации шаблона кража работы work-stealing.

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

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

ForkJoinPool имеет отдельные параллельные очереди, в отличие от Executor пула, который имеет только одну очередь.

Причем эти очереди являются очередями deque или двойными очередями (double ended queue), представляющими собой линейные коллекции, которые поддерживают вставку и удаление элементов на обоих концах.

Фреймворк Fork-Join разделяет задачу на большие подзадачи и обрабатывает каждую такую задачу в отдельном потоке.

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

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

Далее решения подзадач объединяются для получения окончательного результата.

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

Тогда он захватывает с конца другой очереди большую подзадачу и начинает с ней работать.

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

Таким образом, резюмируя, фреймворк fork/join помогает ускорить параллельную обработку, пытаясь использовать все доступные ядра процессора с помощью подхода «разделяй и властвуй».

На практике это означает, что фреймворк сначала создает форки или «вилки», рекурсивно разбивая задачу на более мелкие независимые подзадачи, пока они не будут достаточно простыми, чтобы выполняться асинхронно.

После этого начинается этап «join», в котором результаты всех подзадач рекурсивно объединяются в один результат, или в случае задачи, которая ничего не возвращает, программа просто ждет, пока не будет выполнена каждая подзадача.

Чтобы обеспечить эффективное параллельное выполнение, фреймворк fork/join использует пул