Строчка list.parallelStream() выглядит безобидно и даже современно. На четырёхъядерной машине она обещает четырёхкратный прирост, а приносит то четырёхкратное замедление на мелкой коллекции, то остановку целого сервиса, если в конвейер затесался вызов к базе. Разрыв между обещанием и результатом объясняется не качеством реализации, а механикой общего пула ForkJoinPool, которого у этой строчки ровно один на всю виртуальную машину. Ниже разбор того, как параллельные стримы делят работу, какие задачи им идут впрок, и когда вместо них нужен явный пул, а когда обычный исполнитель.
parallelStream делит работу через общий пул ForkJoinPool
Параллельный стрим исполняется в общем пуле, который называется commonPool. Его параллелизм по умолчанию равен числу доступных ядер минус один: поток, запустивший вычисление, сам становится рабочим, и вычитание не даёт ядру простаивать. Настройка меняется системным свойством java.util.concurrent.ForkJoinPool.common.parallelism, но применяется ко всему пулу сразу, то есть ко всем параллельным стримам процесса, включая чужие.
Деление работы устроено как двоичное расщепление. Источник режется надвое, пока оценка размера куска не опустится до доли от общего; цель задана как четырёхкратный параллелизм, то есть на четырёх ядрах работа распиливается примерно на шестнадцать листьев. Каждый лист проходит конвейер независимо, результаты сшиваются в обратном порядке, и цена сшивки входит в общее время. Задачи лежат в локальных очередях рабочих потоков, а освободившийся поток крадёт половину очереди соседа, так работает переманивание работы. От источника зависит разбиваемость: список режется ровно и предсказуемо, множество с его корзинами хуже, а стрим, построенный из итератора, делится наугад и баланс страдает.
Общий пул обслуживает не только стримы. В нём исполняются и асинхронные ступени CompletableFuture, запущенные без явного исполнителя. Это значит, что любой параллельный стрим в приложении делит ёмкость с любой асинхронной операцией, и конфликт между подсистемами заложен на уровне конструкции, а не возникает случайно.
Глубина деления предсказуема: каждый шаг уменьшает кусок вдвое, и для миллиона элементов путь до листа занимает около двадцати распилов. Мелкие коллекции страдают громче всего: на источнике из восьми элементов деление либо не стартует вовсе, либо стартует и съедает больше, чем вся работа. Экономика расщепления видна прямо в коде фреймворка: цель листьев задана, размер листа вычисляется делением оценки источника на цель, и маленькая коллекция просто не нарабатывает достаточную массу.
Терминальная операция держит ещё одну деталь. Короткие проверки вроде anyMatch заканчиваются на первом подходящем результате, отменяя недоделанные куски, и параллельность помогает маленьким порциям пробегать источник с нескольких сторон одновременно. Полная свёртка, sum или сбор, обязана дождаться всех листов, и быстрый источник может отправить машину скучать в ожидании последнего, самого медленного куска.
Окупаемость параллельного стрима зависит от размера и стоимости элемента
Документация параллельных фреймворков формулирует правило большого пальца: произведение числа элементов N на стоимость обработки одного элемента Q должно быть не меньше десятка тысяч. Проще говоря, миллион дешёвых элементов может окупиться, тысяча средних нет, а десять дорогих никогда. Под это же правило попадает примитивность: стрим из int по массиву делится и считается быстро, стрим из упакованных Integer платит за боксинг на каждом шаге, и выигрыш от параллельности растворяется в упаковке.
Хороший кандидат выглядит так:
long parallel = IntStream.range(0, 500_000_000)
.mapToLong(i -> i ^ (i << 1))
.parallel()
.sum();На четырёх ядрах такой конвейер ускоряется почти линейно, с поправкой на сшивку результатов и запуск задач. Плотный вычислительный цикл по массиву это лучший сценарий: работа делится ровно, листья одинаково тяжелы, терминальная сумма ассоциативна и сшивается без конфликтов. Напротив, упорядоченный поиск по большому источнику с findFirst ловит себя за хвост: первых подходящих элементов может быть много, а выбрать нужно строго первый, и конвейер ждёт координации. Замена на findAny с пометкой unordered снимает часть цены.
Пара замеров для калибровки ожиданий. Примитивный стрим по ста миллионам элементов с плотным отображением на скромной рабочей машине проходит за секунды последовательно и ужимается примерно вчетверо параллельно, минус немного на сшивку. Тот же объём в упакованных значениях тяжелее обоих вариантов на порядок, и параллельность съедается упаковкой. Тысяча элементов со средней стоимостью обработки в микросекунду стоит ровно на границе правила: выгода если и есть, то проценты, и её легко не заметить за шумом. Числа у каждой машины свои, форма зависимости устойчива.
Сбор результатов тоже участвует в экономике. Свёртка в карту требует слияния частичных результатов, и дорогой сборщик съедает выигрыш на отображении; мелкие карты сливаются дёшево, крупные с вложенными структурами платят заметно. Всё, что работает после листьев, оплачивается по тарифу сшивки.
Блокирующие вызовы в общем пуле отравляют всё приложение целиком
Общий пул один на всю машину, и это главная ловушка конструкции. Рабочих потоков в нём мало, по числу ядер минус один, и стрим, элементы которого ходят в базу или по сети, занимает всех рабочих на время каждого вызова. Второй параллельный стрим, запущенный в этот момент, стоит в очереди. Третий, из другого модуля, тоже. Потоки общего пула фоновые, сервис не упадёт, он просто перестанет отвечать, и в дампе потоков картина узнаваемая: строки commonPool-worker стоят в ожидании чтения из сокета, а рядом ждут их заказчики.
Хуже простого ожидания выглядит самозапирание. Параллельный стрим, внутри листа которого ждут результата другого параллельного вычисления, легко занимает всех рабочих и требует ещё одного. Свободных нет, внешний вызов не завершён, ждать некому. Диагноз ставится дампом потоков, лечение всегда одно: убрать блокировку из общего пула и вынести её в отдельный исполнитель.
// так делать не надо
list.parallelStream()
.map(this::fetchFromRemoteDb) // блокирует рабочего на десятки миллисекунд
.toList();Правка на две строки в далёком модуле превращает весь сервис в очередь к базе. Именно из-за этой особенности параллельные стримы часто запрещают корпоративным стилем целиком, и запрет не так глуп, как кажется: дисциплина одного ключевого слова дешевле, чем дисциплина каждого вызова.
Диагностика такой беды не требует редких инструментов. Дамп потоков командой jcmd 4321 Thread.print показывает, чем заняты рабочие общего пула, и если вместо вычислений там чтение из сокета или ожидание блокировки базы, виновник найден. В записях рекордера рабочие общего пула видны как обычные потоки, и их состояние в окне инцидента читается так же ясно.
Явный ForkJoinPool и границы его полезности для параллельных стримов
Способ запустить стрим в собственном пуле известен давно: обернуть всё вычисление в отправку пулу.
ForkJoinPool pool = new ForkJoinPool(16);
long found = pool.submit(() ->
items.parallelStream()
.filter(this::isHeavy)
.count())
.get();
pool.shutdown();Приём работает, потому что отправленная задача исполняется внутри пула, и расщепление обслуживают его потоки. Плюсы очевидны: вычисление получает выделенную ёмкость, параллелизм задаётся под задачу, в дампе потоков виден собственный пул. Минусы тише, но весомее. Приём опирается на детали поведения, которые документация прямо не обещает; потоки собственного пула не фоновые, пул нужно закрывать самому; и главное, выделенный пул не отменяет блокировку внутри листа, он лишь изолирует её последствия от остального приложения.
Для честной работы с блокировками внутри ForkJoin существует интерфейс ManagedBlocker: он временно расширяет пул на время ожидания. Стримы этим механизмом пользоваться не умеют, и оборачивание каждого вызова в ManagedBlocker сводит декларативность конвейера на нет. Наблюдение за собственным пулом устроено проще, чем кажется: у ForkJoinPool есть открытые методы мониторинга очередей и параллелизма, и одна метрика на пул в журнале приложения расскажет о его жизни больше, чем любой спор в команде. Пулов в приложении держат мало, по одному на выделенную партию, и закрывают вместе с задачей: потерянный пул, чьи потоки не фоновые, переживает и тест, и развертывание, и репутацию автора.
Практика выстоялась простая: собственный пул оправдан для тяжёлых вычислительных партий, где нужно гарантировать ресурсы, а для блокирующих нагрузок стримы не используют вовсе.
Замеры разных типов задач от мелкой коллекции до тяжёлой обработки
Честный ответ на вопрос о пользе даёт только замер, и каркас JMH делает его дешёвым:
@Benchmark
public long parallelSum() {
return IntStream.range(0, 10_000_000)
.parallel()
.map(x -> x * 3)
.sum();
}Типичные результаты по четырём классам задач складываются в узнаваемую картину. Коллекция из десятка элементов проигрывает параллельной версии в разы, расходы на расщепление и сшивку больше всей работы. Массив из сотен миллионов примитивов с плотным вычислением ускоряется почти линейно, оставаясь единственным случаем, где стрим безусловно выигрывает. Упакованные значения съедают разы боксингом, и параллельность уже не спасает. Блокирующие вызовы по каждому элементу дают пропускную способность в количество рабочих пула, то есть на четыре ядра около трёх одновременных обращений, и рядом с исполнителем на пятьдесят потоков это выглядит печально.
Отдельная история ожидания. Сто обращений к внешнему сервису по двести миллисекунд каждое на параллельном стриме выполнятся партиями по количеству рабочих, около трёх на четырёх ядрах, и займут около семи секунд. Тот же список на исполнителе с пятьюдесятью потоками укладывается в один заход около четырёхсот миллисекунд плюс накладные. На полностью вычислительной задаче расклад обратный, и стрим выигрывает у исполнителя экономией переключений контекста.
Отдельно ведёт себя порядок. Поиск первого элемента по упорядоченному параллельному источнику платит за координацию, а сбор в упорядоченную структуру платит за слияние кусков. Пометка unordered и выбор findAny возвращают конвейеру скорость, если порядок в задаче не продиктован смыслом.
Правила выбора между parallelStream и ручным пулом
Выбор можно свести к пяти правилам, каждое проверено чужими инцидентами:
- считать parallelStream оптимизацией для тяжёлых вычислительных задач с хорошо делимым источником, а не выключателем скорости по умолчанию;
- подтверждать окупаемость замером на репрезентативном размере с помощью JMH до попадания в основную ветку;
- выносить любые блокирующие вызовы из стримов в исполнители с собственной ёмкостью и собственной политикой отказов;
- не вкладывать параллельные вычисления друг в друга в общем пуле, чтобы не запереть пул на самом себе;
- фиксировать зависимость горячего пути от общего пула комментарием, чтобы будущий читатель видел цену решения и не добавлял соседа.
Пятый пункт списка звучит скромно, а экономит дорого: большинство инцидентов с общим пулом начиналось с безобидной пометки parallel в чужом модуле, о которой не знал никто, включая автора.
Вывод из всей истории двойственный. Для плотной обработки больших массивов параллельный стрим остаётся одним из самых лаконичных инструментов языка: одна пометка, и четыре ядра работают вместо одного. Для всего остального он источник ответственности без полномочий: поведение зависит от общего пула, пул делится со всей машиной, а цена ошибки ложится на невиновных. С выходом виртуальных потоков в двадцать первом выпуске JDK у блокирующих нагрузок появился штатный инструмент, и параллельным стримам не нужно тащить роль, о которой их никто не просил. Явный пул остаётся для редких случаев, когда вычислительной партии гарантируют ресурсы, и для таких случаев он работает хорошо.