🔬 Stream API: что происходит внутри пайплайна

Вы вызываете .stream().filter().map().collect() и получаете результат. Но между вызовом и результатом лежит архитектура, которую стоит понимать, если вы хотите писать эффективный код, а не просто «работающий».

🔹 Pipeline — это не цепочка вызовов

Каждая промежуточная операция не выполняет вычислений. Она возвращает новый объект Stream, который хранит ссылку на предыдущий стейдж. Фактически вы строите связный список из объектов AbstractPipeline.

Иерархия выглядит так:

BaseStream
└─ AbstractPipeline
├─ ReferencePipeline.Head // источник
├─ ReferencePipeline.StatelessOp // filter, map, flatMap...
└─ ReferencePipeline.StatefulOp // sorted, distinct, limit...

Head — точка входа, обёртка над Spliterator. Каждый StatelessOp / StatefulOp — узел, который знает только своего предшественника (previousStage) и свою операцию.

Ничего не вычисляется до вызова терминальной операции — это ленивость на уровне структуры данных, а не какая-то магия отложенных функций.

🔹 Spliterator — настоящий движок

Это не просто итератор, а абстракция, отвечающая за обход (tryAdvance, forEachRemaining), расщепление для параллелизма (trySplit), характеристики источника и оценку размера.

Характеристики — критически важны. От них зависит, какие оптимизации применит пайплайн.

Например:

→ SIZED + SUBSIZED → toArray() аллоцирует массив точного размера без ресайза.
→ DISTINCT на источнике превращает .distinct() в no-op.
→ SORTED позволяет пропустить sorted().

Флаги пробрасываются и модифицируются на каждом стейдже: filter() сбрасывает SIZED, map() — SORTED и DISTINCT.

🔹 Sink chain: push-модель

При терминальной операции пайплайн собирается с конца к началу в цепочку Sink-ов. Каждый Sink реализует begin(size), accept(element), end() и cancellationRequested(). Стейдж оборачивает downstream через opWrapSink().

Данные идут поэлементно: элемент входит в первый sink и проталкивается до конца цепочки. Не «сначала весь filter, потом весь map». Исключение — stateful-операции: sorted() копит всё в буфер — это барьер в пайплайне.

🔹 Short-circuiting: нюанс

limit, findFirst, anyMatch останавливают обход через cancellationRequested(). Но forEachRemaining() этот флаг не проверяет — AbstractPipeline переключается на поэлементный tryAdvance при наличии short-circuiting в цепочке. limit(10) на бесконечном источнике сработает, но ценой потери batch-оптимизации.

🔹 Параллельные стримы

parallelStream() использует ForkJoinPool.commonPool(), trySplit() рекурсивно делит данные, задачи раскидываются через CountedCompleter.

Что важно:

— sorted() в parallel дорого, т.к. требует merge-фазы
— distinct() на UNORDERED → ConcurrentHashMap, на ORDERED — linked-структура с барьерами
— limit(n) + parallel + ORDERED — ловушка: буферизация убивает параллелизм
— unordered() перед limit()/distinct() — не стилистика, а указание рантайму выбрать дешёвый алгоритм

💬 Какой самый неочевидный баг ловили из-за механик стримов?

══════ Навигация ══════
Вакансии • Задачи • Собесы

🐸 Библиотека джависта

#CoreJava