Что такое параллельные стримы

Под капотом параллельные стримы используют общий пул потоков JVM — ForkJoinPool.commonPool() и алгоритм «разделяй и властвуй» (Divide and Conquer). Для создания параллельного ст...

🟢 Junior Level

Краткий ответ для собеседования (30 секунд)

Параллельные стримы (Parallel Streams) — это механизм Stream API, позволяющий автоматически распределять обработку элементов потока между несколькими ядрами процессора без явного написания многопоточного кода (создания потоков, управления блокировками и синхронизацией).

Под капотом параллельные стримы используют общий пул потоков JVM — ForkJoinPool.commonPool() и алгоритм «разделяй и властвуй» (Divide and Conquer). Для создания параллельного стрима достаточно вызвать метод collection.parallelStream() у коллекции или переключить существующий стрим вызовом промежуточного метода .parallel().

Базовый пример на Java 21

import java.util.List;
import java.util.stream.LongStream;

public class ParallelStreamBasicsDemo {

    public static void main(String[] args) {
        // 1. Создание параллельного стрима из диапазона чисел
        long sum = LongStream.rangeClosed(1, 10_000_000)
            .parallel()
            .filter(n -> n % 2 == 0)
            .map(n -> n * 3)
            .sum();

        System.out.println("Сумма: " + sum);

        // 2. Иллюстрация многопоточного выполнения из коллекции
        List<String> items = List.of("A", "B", "C", "D", "E", "F");

        System.out.println("Обработка элементов в разных потоках:");
        items.parallelStream().forEach(item -> {
            System.out.printf("Элемент: %s, Поток: %s%n",
                item, Thread.currentThread().getName());
        });
    }
}

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

🟡 Middle Level

Механизм деления данных: `Spliterator.trySplit()`

Параллельный стрим делит исходный набор данных на подзадачи с помощью метода Spliterator.trySplit():

Исходный сплитератор пытается разделить свою последовательность пополам, возвращая новый Spliterator для первой половины, оставляя себе вторую.
Процесс повторяется рекурсивно до тех пор, пока размер порции данных (chunk) не станет достаточно мал для последовательной обработки.
Задачи передаются рабочим потокам ForkJoinPool, а результаты объединяются с помощью combiner().

                  [ 1, 2, 3, 4, 5, 6, 7, 8 ]
                             │
                    trySplit()
                   ┌─────────┴─────────┐
             [ 1, 2, 3, 4 ]       [ 5, 6, 7, 8 ]
                   │                     │
              trySplit()            trySplit()
             ┌─────┴─────┐         ┌─────┴─────┐
          [ 1, 2 ]    [ 3, 4 ]  [ 5, 6 ]    [ 7, 8 ]
             │           │         │           │
          Worker 1    Worker 2  Worker 3    Worker 4

Эффективность разделения различных структур данных:

Источник данных	Сложность `trySplit()`	Качество деления	Оценка производительности
`ArrayList`, массивы, `IntStream.range`	$O(1)$	Идеальное (деление по индексам ровно пополам)	🟢 Превосходная
`HashSet`, `TreeSet`	$O(1)$ амортизировано	Хорошее (деление по бакетам / поддеревьям)	🟡 Хорошая
`LinkedList`	$O(N)$	Ужасное (чтобы найти середину, нужно обойти $N/2$ узлов)	🔴 Катастрофическая
`Stream.iterate()`	$O(N)$	Невозможно разделить заранее (каждый элемент зависит от предыдущего)	🔴 Параллелизм бессмыслен

Общий пул `ForkJoinPool.commonPool()`

По умолчанию все параллельные стримы во всем JVM-процессе разделяют единый статический пул потоков ForkJoinPool.commonPool().

Размер пула по умолчанию: Runtime.getRuntime().availableProcessors() - 1.
Почему - 1? Поток, инициировавший вызов терминальной операции стрима (calling thread), сам становится активным участником вычислений, выполняя задачи наравне с воркерами пула.
Настройка размера пула: флаг JVM -Djava.util.concurrent.ForkJoinPool.common.parallelism=N.

🔴 Senior Level

Внутреннее устройство: `CountedCompleter` и Work-Stealing

В основе параллельной редукции лежат специальные подклассы ForkJoinTask, главным образом CountedCompleter (например, ReduceTask, ForEachTask, MatchTask).

Worker Thread 1 Deque:  [Task A] [Task B] [Task C] <- Push / Pop (LIFO)
Worker Thread 2 Deque:  (Empty) -----------------> Steal Task A (FIFO)

Модель Work-Stealing: У каждого рабочего потока есть своя двусторонняя очередь (deque). Поток берет свои задачи с конца очереди (LIFO — кэш-локальность). Если очередь пуста, он крадет («ворует») чужие задачи из начала их очередей (FIFO).
Отсутствие ожидания блокировок: CountedCompleter не блокирует поток вызовом join(). Вместо этого каждый завершившийся дочерний узел декрементирует счетчик родительской задачи (pendingCount), и последний завершившийся поток самостоятельно запускает слияние результатов (onCompletion).

Формула целесообразности параллелизма: модель $N \times Q$

По правилу Брайана Гетца (Brian Goetz, Java Language Architect), параллелизм дает преимущество, только когда: $N \times Q > 10\,000$ где:

$N$ — число элементов в потоке;
$Q$ — вычислительная сложность обработки одного элемента (такты CPU).

Если $Q$ мало (например, простое суммирование чисел i -> i + 1), $N$ должно составлять сотни тысяч или миллионы элементов. В противном случае накладные расходы на разделение задач, синхронизацию очередей ForkJoinPool, переключение контекста и слияние результатов через combiner() превысят выигрыш от параллелизма, и код выполнится медленнее обычного цикла for.

Опасности и архитектурные ловушки `commonPool()`

1. Блокирующий I/O в параллельном стриме (Starvation)

Если внутри лямбды параллельного стрима выполняется блокирующий сетевой запрос (HTTP, JDBC) или чтение файла, рабочий поток пула засыпает. Поскольку commonPool() один на всё приложение:

Другие части системы (параллельные стримы других пользователей, CompletableFuture.supplyAsync()) оказываются заблокированными из-за нехватки свободных потоков.
В веб-приложениях (Spring Boot) это приводит к лавинообразному исчерпанию ресурсов и падению throughput.

2. Изоляция в выделенном `ForkJoinPool`

Для изоляции параллельных вычислений от общего пула можно использовать трюк с отправкой задачи в отдельный пул:

import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;

public class CustomPoolStreamDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        List<Integer> data = List.of(1, 2, 3, 4, 5, 6, 7, 8);
        ForkJoinPool customPool = new ForkJoinPool(4);

        try {
            long result = customPool.submit(() ->
                // Внутри submit() parallelStream использует ForkJoinTask.getPool(),
                // то есть выполняется именно в customPool, а не в commonPool!
                data.parallelStream()
                    .mapToInt(x -> x * 2)
                    .sum()
            ).get();

            System.out.println("Результат: " + result);
        } finally {
            customPool.shutdown();
        }
    }
}

Senior Warning: Хотя трюк customPool.submit(() -> stream.parallel()) работает во всех версиях OpenJDK, он формально не гарантируется спецификацией Java (является внутренней деталью реализации).

🎯 Шпаргалка для интервью

30-секундный ответ

«Параллельные стримы автоматически разделяют обработку данных между ядрами процессора с помощью ForkJoinPool.commonPool(). Данные рекурсивно делятся пополам методом Spliterator.trySplit(), обрабатываются воркерами по алгоритму work-stealing и объединяются через combiner. Параллелизм эффективен только при соблюдении правила $N \times Q > 10\,000$ на структурах с быстрым произвольным доступом (ArrayList, массивы). Главное табу — выполнять блокирующий I/O внутри parallel stream, так как это блокирует общий пул всей JVM.»

Таблица пригодности структур данных для `parallelStream`

Источник данных	Пригодность	Причина
`int[]`, `long[]`, `double[]`	🟢 Идеально	$O(1)$ сплит, отсутствие автобоксинга, кэш-локальность
`ArrayList`	🟢 Отлично	$O(1)$ сплит по индексу, известен размер (`SIZED`)
`HashSet` / `HashMap.keySet`	🟡 Удовлетворительно	Нет индексов, но бакеты делятся достаточно хорошо
`LinkedList`	🔴 Плохо	$O(N)$ для нахождения точки деления, высокое GC-давление
`Stream.iterate()`	🔴 Недопустимо	Полностью последовательная генерация, сплит невозможен
`Files.lines()`	🔴 Опасно	Дисковый I/O, блокировка потоков `commonPool`

Топ-4 каверзных вопроса на интервью

1. Почему размер `ForkJoinPool.commonPool()` по умолчанию равен `processors - 1`, а не количеству ядер?

Поток, который вызвал терминальную операцию стрима (например, поток main или поток HTTP-запроса), сам вступает в выполнение вычислений в роли рабочего потока. Таким образом, общее количество активных потоков, участвующих в вычислении, равно (processors - 1) + 1 = processors, что точно соответствует числу доступных ядер CPU без избыточного переключения контекста.

2. Как запустить `parallelStream()` в отдельном пуле потоков и гарантируется ли это спецификацией?

Поместить выполнение стрима в задачу для кастомного пула: customPool.submit(() -> stream.parallel().collect(...)).get().
Спецификация Java SE не гарантирует, что параллельный стрим обязан выполняться в пуле вызывающего потока (это деталь реализации OpenJDK, где ForkJoinTask.getPool() проверяет текущий контекст). Однако на практике в HotSpot это работает стабильно во всех версиях.

3. Почему `Stream.iterate(0, i -> i + 1).parallel().limit(100)` работает крайне медленно или приводит к высокой нагрузке на CPU?

Stream.iterate производит элементы строго последовательно (каждый элемент является функцией от предыдущего). Метод trySplit() не может разрезать такой стрим на независимые куски, поэтому один поток генерирует элементы, а остальные потоки вынуждены постоянно синхронизироваться и ожидать. Для параллельных числовых диапазонов следует использовать IntStream.rangeClosed().

4. В чем разница между `forEach()` и `forEachOrdered()` в параллельном стриме?

forEach() передает элементы в переданный Consumer сразу по мере их готовности в любом рабочем потоке (порядок вывода непредсказуем).
forEachOrdered() принудительно соблюдает так называемый encounter order (порядок элементов в исходной коллекции). Для этого рантайм вынужден буферизовать результаты и блокировать передачу downstream до завершения предыдущих по порядку задач, что существенно снижает параллельную производительность.

Типичные ошибки и Red Flags

🚩 Вызов parallelStream() для любых мелких списков из 50 элементов в надежде «ускорить код» (в реальности код замедляется из-за overhead).
🚩 Выполнение HTTP/SQL-запросов внутри parallelStream(), приводящее к исчерпанию потоков commonPool() всего сервиса.
🚩 Мутация общего несинхронизированного состояния (например, list.add()) внутри параллельного стрима, вызывающая race conditions и ArrayIndexOutOfBoundsException.
🚩 Использование ThreadLocal внутри параллельного стрима: потоки пула переиспользуются, данные загрязняют чужие задачи (Memory Leak / Context Leak).

🟢 Junior Level

Краткий ответ для собеседования (30 секунд)

Базовый пример на Java 21

🟡 Middle Level

Механизм деления данных: Spliterator.trySplit()

Эффективность разделения различных структур данных:

Общий пул ForkJoinPool.commonPool()

🔴 Senior Level

Внутреннее устройство: CountedCompleter и Work-Stealing

Формула целесообразности параллелизма: модель $N \times Q$

Опасности и архитектурные ловушки commonPool()

1. Блокирующий I/O в параллельном стриме (Starvation)

2. Изоляция в выделенном ForkJoinPool