Forkjoinpool java что это
Параллельное программирование в Java8. Создание многопоточных программ с помощью Fork/Join Framework
Статья посвящена такому интересному и полезному механизму (совокупностям механизмов и библиотек), как Fork/Join Framework. Он позволяет многократно ускорить вычисления, добиться максимальных результатов при обработке, используя все доступные возможности системы (процессоры).
В рамках данной статьи будет созданы классы, использующие Fork/Join Framework. В коде показан один из возможных вариантов применения параллельного программирования. Итак, начнем.
Создавая приложения, следует максимально разделять части, отвечающие за запуск, настройку и обработку данных. И данный вариант работы с Fork/Join — не исключение. В примерах будут использованы классы Start, Stream, Calc соответственно.
Часть первая — запуск
Для тестирования создадим класс Start, он будет служить «точкой» запуска. Значение timebetweenStartEnd покажет нам интервал времени между началом и окончанием расчетов. Под расчетами подразумевается возведение в степень чисел от 0 до 1000000 в двух вариантах в однопоточном и многопоточном режиме.
В классе Start определен пул потоков ForkJoinPool(). С помощью метода invoke() был достигнут результат запуска задачи и ожидания ее выполнения. Значение componentValue определено равное 1000000. Во вновь созданном экземпляре класса Stream определены исходные данные. С помощью invoke() мы «переводим» данную задачу на выполнение.
Часть вторая. Настройка. Класс Stream
Вторая часть механизма представляет класс (Stream), отвечающий за настройку многопоточности. Сейчас у нас всего два таких варианта: первый — по количеству обрабатываемых значений в одном потоке (далее — «отсечка)», второй — по количеству процессоров (получаем с помощью метода availableProcessors()). Прошу обратить внимание читателей, что в данной статье не будет прорабатываться механизм динамического создания потоков в зависимости от количества процессоров и/или других условий. Это тема следующей статьи.
В классе использован абстрактный метод compute(), отвечающий за запуск вычислений, в нашем случае это выбор варианта расчета и запуск расчетов в методе go класса Calc. С помощью метода invokeAll() произведем запуск подзадач.
Из алгоритма видно, что в случае, если у нас больше одного процессора, или значение отсечки (500000) больше/равно полученным частям, то происходит расчет. В примере, мы делим forSplit на несколько частей (две) и запускаем две подзадачи. Изменив значение переменной countLimit или выставив значение countProcessors равное единице произойдет запуск только одной задачи по обработке данных.
Часть третья. Выполнение расчета. Класс Calc
Данный класс отвечает за возведение числа в степень. Нижеуказанная часть предназначена для демонстрации и может содержать любые вычисления от перебора коллекций до записи данных в хранилище.
Руководство по структуре Fork / Join в Java
1. Обзор
2. ForkJoinPool
ForkJoinPool сердце рамок. Это реализация ExecutorService, которая управляет рабочими потоками и предоставляет нам инструменты для получения информации о состоянии и производительности пула потоков.
Рабочие потоки могут выполнять только одну задачу за раз, но ForkJoinPool не создает отдельный поток для каждой подзадачи. Вместо этого каждый поток в пуле имеет свою собственную двустороннюю очередь (или deque, произносится как колода ), в которой хранятся задачи.
Эта архитектура жизненно важна для балансировки нагрузки потока с помощью алгоритма кражи работы.
2.1. Алгоритм кражи работы
Проще говоря, свободные потоки пытаются «украсть» работу у занятых потоков.
По умолчанию рабочий поток получает задачи из головы своей собственной двухсторонней очереди. Когда он пуст, поток берет задачу из хвоста двухсторонней очереди другого занятого потока или из глобальной очереди на вход, поскольку именно здесь, вероятно, будут находиться самые большие части работы.
Такой подход сводит к минимуму вероятность того, что потоки будут конкурировать за задачи. Это также уменьшает количество раз, когда поток должен будет искать работу, поскольку он сначала работает с самыми большими доступными фрагментами работы.
2.2. Создание экземпляра ForkJoinPool
Согласно документации Oracle, использование предопределенного общего пула снижает потребление ресурсов, поскольку это препятствует созданию отдельного пула потоков для каждой задачи.
Такого же поведения можно добиться в Java 7, создав ForkJoinPool и назначив его общедоступному статическому полю служебного класса:
Теперь к нему легко получить доступ:
С помощью конструкторов ForkJoinPool можно создать собственный пул потоков с определенным уровнем параллелизма, фабрикой потоков и обработчиком исключений. В приведенном выше примере пул имеет уровень параллелизма 2. Это означает, что пул будет использовать 2 ядра процессора.
3. ForkJoinTask
В результате метод возвращает список.
Список передается в ForkJoinPool с помощью метода invokeAll () :
Этот шаблон можно использовать для разработки ваших собственных классов RecursiveAction . Для этого создайте объект, который представляет общий объем работы, выберите подходящий порог, определите метод разделения работы и определите метод для выполнения работы.
3.2. RecursiveTask
Для задач, возвращающих значение, здесь используется аналогичная логика, за исключением того, что результат для каждой подзадачи объединяется в один результат:
В этом примере это достигается с помощью Java 8 Stream API; метод sum () используется как представление объединения подрезультатов в окончательный результат.
4. Submitting Tasks to the ForkJoinPool
To submit tasks to the thread pool, few approaches can be used.
The submit() or execute()method (their use cases are the same):
The invoke()method forks the task and waits for the result, and doesn’t need any manual joining:
The invokeAll() method is the most convenient way to submit a sequence of ForkJoinTasks to the ForkJoinPool. It takes tasks as parameters (two tasks, var args, or a collection), forks then returns a collection of Future objects in the order in which they were produced.
Alternatively, you can use separate fork() and join() methods. The fork() method submits a task to a pool, but it doesn’t trigger its execution. The join() method must be used for this purpose. In the case of RecursiveAction, the join() returns nothing but null; for RecursiveTask, it returns the result of the task’s execution:
In our RecursiveTask example we used the invokeAll() method to submit a sequence of subtasks to the pool. The same job can be done with fork() and join(), though this has consequences for the ordering of the results.
Чтобы избежать путаницы, обычно рекомендуется использовать метод invokeAll () для отправки более одной задачи в ForkJoinPool.
5. Выводы
Использование инфраструктуры fork / join может ускорить обработку больших задач, но для достижения этого результата следует соблюдать некоторые рекомендации:
Примеры, использованные в этой статье, доступны в связанном репозитории GitHub.
Вилкой в глаз, или ForkJoinPool в Java
Всем привет. Сегодня я хотел бы поговорить о многопоточности. Вернее, не о многопоточности вообще, а о таком её механизме как ForkJoinPool. Нельзя сказать, что данная технология является новой (она появилась ещё в Java 7), или что в сети нельзя найти материалы по данной теме. Информации хватает. Например, для глубокого погружения могу порекомендовать лекцию блистательного Алексея Шипилёва, которую можно без труда найти на YouTube. Но лично мне большинство этих материалов показались либо слишком сложными, либо наоборот – поверхностными. Так же некоторые из них содержат явные ошибки, что вносит ещё большую неразбериху в данную тему. Судя по тому, что в комментариях под одной из этих статей я нашёл вот такую картинку, подобные проблемы были не только у меня.
Если вдруг и вы перелопатили всё, что нашли в сети по поводу ForkJoinPool, а просветления так и не достигли, добро пожаловать под кат. Попробуем на максимально простом примере разобраться в данной теме, с котиками, картинками, всё как вы любите.
Для понимания всего, что будет изложено ниже, крайне желательно быть знакомым с основами многопоточности (Thread, Runnable, Callable, Future и т.д.).
Если вы уже что-то копали по теме ForkJoin, то должны знать, что в основе данной технологии лежит старый как мир принцип «разделяй и властвуй». Легко нагуглить, что если у нас есть какая-то задача, с помощью ForkJoinPool мы сначала делим её на подзадачи, выполняем их, потом объединяем результаты и делаем это всё рекурсивно…
Заполнить массив можно любыми числами (в данном случае это не важно). Скорее всего System.out.println(new Date()) – не самый оптимальный способ измерить скорость выполнения кода, но весьма простой и для нашего примера сгодится. Thread.sleep(1) добавлен для того, чтобы сымитировать задачу, которая при работе в одном потоке вызывает значительную загрузку процессора. У меня на выполнение данного кода ушло 17 секунд. Таким образом, мы имеем некую большую задачу, существенно замедляющую работу нашей программы. Очевидно, что запуск её в параллельном потоке проблемы не решит. Что же делать? Конечно же, разбить эту задачу на подзадачи. Допустим, мы разделим наш массив пополам, суммирование первой части массива запустим в одном потоке, суммирование второй части массива – в другом, а потом сложим получившиеся результаты. Проблема в том, что если задача достаточно большая, то обе её половинки также могут получиться достаточно большого размера, что не лучшим образом скажется на производительности. Следовательно, возможно и их нужно будет поделить на части и продолжать данную операцию до достижения некоего оптимального размера. Когда условие будет достигнуто, каждый из этих кусочков мы отдадим отдельному потоку, а потом соберём получившиеся результаты воедино. Чувствуете? В воздухе отчётливо запахло рекурсией, и мы всё ближе приближаемся к ForkJoinPool.
Допустим, что в деле изучения ForkJoinPool вы уже миновали стадию гнева и находитесь на стадии отрицания, тогда у вас может возникнуть вполне резонный вопрос: «Ну, и зачем нам нужен этот ForkJoin, да ещё и с какой-то рекурсией? Разве нельзя всё сделать проще?» В каком-то смысле можно. Напомню, что у нас есть интерфейс Callable, метод которого call() возвращает некое значение и запускается асинхронно в отдельном потоке. Ничто не мешает нам создать класс, имплементирующий данный интерфейс и содержащий в качестве поля числовой массив. Мы можем поделить наш огромный массив на 100500 маленьких массивов, создать 100500 экземпляров такого класса, создать 100500 отдельных потоков, собрать их в одну коллекцию, запустить их в цикле, потом ещё в одном цикле получить из них значения. Но вы уверены, что хотите построить ещё один велосипед из костылей, а не воспользоваться уже готовым решением, пусть и несколько сложным? Кроме того, описанное решение, обладает ещё одним существенным недостатком. Создание отдельных потоков – операция весьма тяжеловесная и ресурсозатратная. Рассчитывая получить прирост в производительности, и создавая 100500 потоков, мы рискуем получить прямо противоположный результат. Именно по этой причине и был придуман пул потоков, одним из видов которого является ForkJoinPool.
Итак, в основе своей ForkJoinPool – это пул потоков, преимущество которого состоит в том, что он работает на основе принципа WorkStealing, что дословно можно перевести как «кража работы». Когда один из потоков ForkJoinPool заканчивает свою работу, он не идёт пить кофе или чилить в ютубчике, он проявляет «сознательность» и берёт из общей очереди работ новую задачу. Это продолжается до тех пор, пока задачи не кончатся.
Ещё одной особенностью ForkJoinPool является то, что в него нельзя подать Callable или Runnable задачу. У него есть своя иерархия задач, наследуемая от абстрактного класса ForkJoinTask. Основные реализации – RecursiveTask и RecursiveAction. У каждого из них есть абстрактный метод compute(), который и надо реализовывать при наследование. RecursiveTask. compute() возвращает некое значение, RecursiveAction. compute() возвращает void.
Не знаю, как у вас, но у меня при первом знакомстве с данными классами по спине пробежали мурашки. «Раз они recursive, значит в них обязательно надо применить чёрную магию рекурсии…» Как я понял, на самом деле не обязательно (если я не прав, напишите в комментариях). Такой код вполне легален и будет работать.
Если мы передадим экземпляр такого класса на выполнение в ForkJoinPool, то получим обычную строку без всякой рекурсии. Судя по всему, (возможно, я ошибаюсь), создатели данных классов добавили в название слово Recursive в качестве некой рекомендации, а не обязательного требования.
Следующий не вполне очевидный вопрос: как запустить задачу в ForkJoinPool на исполнение? Для этого есть метод T invoke(ForkJoinTask task)
Здесь мы вообще обошлись без всякого ForkJoinPool. Задача выполнила сама себя! Это никуда не годится. Для того, чтобы понять, какой именно метод нужно использовать и в чём разница между ними, обратимся к официальной документации. С методом invoke() всё ясно, он «выполняет данную задачу, возвращая результат по завершении». А вот метод fork() работает немного сложнее, он «организует асинхронное выполнение этой задачи в пуле, в котором выполняется текущая задача». Для полного духовного просветления изменим наш класс.
и видим в консоле: «I am work in thread: ForkJoinPool-1-worker-1» и «I am just innocent simple class».
и видим: «I am work in thread: main» и «I am just innocent simple class». То есть при вызове метода fork() задача не «выполнила сама себя» магическим образом, а была выполнена в том же потоке из которого и был вызван данный метод. Вызов метода ForkJoinPool.invoke() передал задачу на выполнение в один из потоков данного пула. Важно отметить, что метод fork() отправляет задачу в какой-либо поток, но при этом не запускает её выполнения. Для получения результата служит метод join().
Разобравшись с основными классами и методами, вернёмся к нашему массиву и попробуем применить на практике, полученный духовный опыт.
Сначала мы будем рекурсивно делить наш массив на всё более мелкие части, пока не получим массивы, состоящие всего из 2 элементов. Почему именно из 2? Потому что наш условный «слабенький» процессор может условно «быстро» выполнить именно такую условно «маленькую» задачу. Почему рекурсивно? Просто потому, что применение рекурсии в данном случае действительно удобно. Это позволяет сначала выполнить некую работу по подготовке, а потом получить результат. Если вам не нравится рекурсия, то, наверное, можно попробовать найти какой-то другой способ, её применение, судя по всему, не является обязательным.
После того, как мы получим «100500» маленьких массивов, состоящих всего из 2 элементов, мы запустим «100500» маленьких задач на выполнение и суммируем их результаты. И для этого нам не придётся создавать 100500 отдельных нитей выполнения.
При создании экземпляра класса ValueSumCounter мы передаём в него массив. В методе compute() сначала проверяется длинна массива, и если он «слишком большой», то разбивается пополам на 2 части, на основе каждой из которых в свою очередь создаётся своя задача и отправляется на выполнение путём вызова метода fork(). Когда разбивка будет закончена, наступает время «собирать камни», метод join() запускает каждую задачу на выполнение и возвращает полученный результат. Выполнение данной задачи с помощью ForkJoinPool заняло у меня на компьютере 3 секунды. Напомню, что эта же задача, выполненная с помощью цикла в одном потоке, ранее заняла 17 секунд.
Нам осталось убедиться, что при использовании ForkJoinPool не создаётся «100500» отдельных потоков. Для этого добавим в метод compute() всего одну строку
Запустив код на выполнение, мы увидим, что для выполнения большого количества задач используется несколько одних и тех же потоков (в моём случае 4).
Вот, пожалуй, и всё, что хотелось бы рассказать о данной технологии. Если вдруг в коде или описании обнаружатся ошибки, просьба не кидать в автора тапками, а написать в комментариях об этом. Надеюсь, данный материал поможет духовному просветлению юных падаванов, ищущих знаний. Да пребудет с вами сила Java.
Незаслуженно забытый ForkJoinPool
Всем известно о новых функциях, которыми нас порадовал JDK 8, и, вероятно, трудно найти Java-разработчика, который не знает, что такое Java Streams, лямбды или CompletableFutures. Итак, все эти приятности появились несколько лет назад вместе с JDK 8, но что произошло немного раньше, когда состоялся выпуск JDK 7?
Даже если мы заглянем в раздел “Новые функции в релизе JDK 7”, то все еще будет непросто определить одну из самых важных представленных функций. Чтобы увидеть наконец то, что мы ищем, придется перейти в раздел “Утилиты параллелизма”.
Само собой, мы говорим о выпуске в JDK7 фреймворка ForkJoinPool, что, на мой взгляд, не получило заслуживаемой огласки. Я бы сказал, что многие Java-разработчики сейчас вообще незнакомы с ForkJoinPool и где его применять.
В этой статье мы рассмотрим внутренние компоненты фреймворка ForkJoinPool, объясним, почему он так важен, и воздадим должное этой забытой части JDK.
Вступление
Что же из себя представляет фреймворк ForkJoinPool? Это детализированный фреймворк для эффективного распараллеливания выполнения задач. Как уже говорилось, он был представлен как часть релиза JDK 7.
Что же такого интересного в ForkJoinPool? Что он дает нам, чего не могут дать существующие исполнители задач? На этот вопрос можно ответить двумя словами: кража работы (work stealing)!
Дизайн ForkJoinPool основан на фреймворке work-stealing, созданном для Cilk. Если вас интересует оригинальный дизайн, вы можете прочитать об этом здесь.
Как работает ForkJoinPool
Дизайн ForkJoinPool на самом деле прост и в то же время очень эффективен. Он основан на алгоритме “Разделяй и властвуй”: каждая задача разбивается на подзадачи по максимуму, затем они выполняются параллельно, и как только все из них завершаются, происходит объединение результатов.
Звучит знакомо? Да, параллельные потоки Java выполняются очень похожим образом.
Итак, у нас есть фреймворк, который следует вычислительной модели делимых задач, где каждая из них соответствует шаблону, подобному этому фрагменту псевдокода:
Эту задачу можно рекурсивно разделять на несколько подзадач до тех пор, пока не будет достигнуто заранее определенное состояние, когда уже можно счесть, что задача достаточно мала и ее можно выполнять. Примерно это и называется моделью Fork-Join.
Модель Fork-Join
Как мы сказали только что, модель fork-join — это метод, в котором мы разделяем каждую задачу (fork), а затем ждем объединения (join) всех получившихся подзадач и получаем результат.
На рисунке выше видно, как каждая задача разделяется всякий раз, когда вызывается fork. Точно так же, когда все задачи завершаются, они объединяются посредством join для получения конечного результата.
Теперь, когда мы понимаем, как работает эта модель, перейдем к самой важной части: как ForkJoinPool выполняет эти задачи внутри себя?
Внутреннее устройство ForkJoinPool
Вы также можете создать свой собственный ForkJoinPool, указав, сколько потоков вам нужно. Только помните, что пул с числом потоков, превосходящим количество доступных процессоров, для задач с интенсивным использованием процессора полезен не будет. Однако, если у вас задачи интенсивного ввода-вывода (то есть им придется часто ждать завершения операций ввода-вывода), большой пул все-таки может пригодиться.
Каждый рабочий поток имеет собственную двухстороннюю рабочую очередь (deques) типа WorkQueue.
Таким образом, каждый из рабочих потоков продолжает сканировать доступные для выполнения подзадачи. Основная цель в том, чтобы как можно больше нагружать рабочие потоки и максимизировать использование ядер процессора. Поток блокируется только тогда, когда нет доступных для выполнения подзадач.
Что происходит, когда рабочий поток не может найти задачи для запуска в своей собственной очереди? Он будет пытаться “украсть” задачи у тех процессоров, которые загружены сильнее!
Вот тут-то и возникает интересный вопрос: как фреймворк гарантирует, что владелец очереди и “похититель” не будут мешать друг другу, если они попытаются перехватить задачу в одно и то же время?
Чтобы свести к минимуму конкуренцию и сделать ее более эффективной, как владелец очереди, так и “похитители” захватывают задачи из разных частей очереди.
Здесь представлен метод LIFO — Last In, First Out (“последним вошел — первым вышел”). Почему именно так? Разве не разумнее было бы сначала обработать задачи, которые дольше пробыли в очереди?
Нет, не совсем. Основная причина — повышение производительности. Всегда, выбирая самую последнюю задачу, мы увеличиваем шансы на то, что ресурсы задачи все еще будут распределены в кэшах процессора, а это значительно повысит производительность. Это обычно называют локальностью ссылок.
В этом случае мы уже следуем подходу FIFO (“первым вошел — первым вышел”). Это в основном служит для уменьшения конкуренции, необходимой для синхронизации как владельцу очереди, так и “похитителю”.
Еще одна хорошая причина в том, что из-за природы самих делимых задач более старые задачи в очереди, скорее всего, будут больше по объему, поскольку еще не подверглись разделению.
Методы push и pop вызываются только владельцем очереди, а метод poll вызывается только процессором, пытающимся “украсть” работу у другого процессора.
Методы push и pop — это операции CAS (Compare-and-Swap, сравнение с обменом) без ожидания, так что они весьма эффективны. Однако метод poll не всегда свободен от блокировки. Он блокируется в тех случаях, когда очередь почти пуста, поскольку требуется некоторая синхронизация для гарантии, что только владелец или похититель выберет данную задачу, но не оба сразу.
У этого фреймворка действительно любопытный дизайн.
Сводим все воедино
Учитывая то, что мы рассмотрели, становится совершенно очевидно, что для использования этой структуры нам понадобятся задачи, которые легко могут разделяться.
Вот где более чем уместны коллекции Java. Начиная с JDK 8, Java-коллекции можно легко делить, поэтому Java Streams и ForkJoinPool — идеальные партнеры!
Благодаря фреймворку ForkJoinPool параллельные потоки Java могут работать очень эффективно, оптимизируя использование доступных ядер!
Но что если для обработки нужного типа задачи неприменимы потоки Java? В этом случае можно создавать свои собственные “делимые” задачи, просто расширяя класс ForkJoinTask.
Что такое ForkJoinTask? ForkJoinTask — это Java-класс, который ведет себя аналогично потоку Java, но он гораздо более легкий. В основном потому, что ему не приходится поддерживать свой собственный стек времени выполнения или счетчики программ.
Существует три подтипа ForkJoinTask: RecursiveAction, RecursiveTask и CountedCompleter. Выбор того или иного подтипа будет зависеть от типа задач, которые вы пишете. Изучите документацию, чтобы понять, какой из них лучше всего соответствует вашим потребностям.
Стоит отметить, что эти три класса не являются функциональными интерфейсами, главным образом потому, что ForkJoinPool был выпущен в JDK7. Поэтому, к сожалению, лямбда-выражениями воспользоваться будет нельзя. Однако платформа параллельного потока Java 8 предоставляет функциональный API для прозрачного использования ForkJoinPool.
Давайте попробуем реализовать простую задачу с помощью ForkJoinTask. Напишем вычисление чисел Фибоначчи.
Вот и все. Наша задача будет разделена на несколько подзадач, и рабочие потоки ForkJoinPool будут действовать вместе, чтобы решить их все до одной.
Обратите внимание: в примере нам пришлось использовать BigInteger ради возможности обрабатывать большие числа, но это всего лишь одна из потенциальных реализаций ForkJoinTask.
Очень хорошая практика, которую стоит применять при написании ForkJoinTasks, — это всегда писать их как чистые функции, не разделяя состояние и избегая мутации объектов. Это лучший способ гарантировать, что подзадачи выполняются безопасно и независимо.
Также имейте в виду, что ForkJoinPool позволяет отправлять не только ForkJoinTasks, но также вызываемые (Callable) или выполняемые (Runnable) задачи, поэтому вы можете применять ForkJoinPool таким же образом, как и другие существующие исполнители. Единственное отличие в том, что ваша задача не будет разделяться сама по себе, но вы можете извлечь выгоду из повышения производительности кражи работы, если будет отправлено несколько задач и у некоторых потоков будет меньше загрузка, чем у других.
Если вам хочется лучше понять параллелизм и многопоточность, а также то, почему сейчас мы делаем всё именно таким образом, а не по-другому, я бы рекомендовал прочитать книгу “Параллелизм Java на практике” Брайана Гетца.
Заключение
Как мы убедились сегодня, модель Fork Join — эффективный способ обработки задач с использованием метода “Разделяй и властвуй”. Вместе с функцией “кражи работы” этот метод делает ForkJoinPool мощным инструментом для распараллеливания Java-кода.
В наше время ForkJoinPool несложно и прозрачно применяется вместе с Java Streams или CompletableFutures, поэтому, скорее всего, нам разве что изредка придется писать собственные делимые задачи. В любом случае всегда полезно понимать, как работает этот функционал, и знать, что такая возможность есть.