🌊 Раздел 8 · Вопрос #11

Как создать parallel stream

Параллельный стрим можно создать двумя основными способами:

🌊 Stream API 📄 Исходный Markdown

🟢 Junior Level

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

Параллельный стрим можно создать двумя основными способами:

  1. Напрямую из коллекции: вызвав метод collection.parallelStream().
  2. Преобразованием существующего стрима: вызвав промежуточную операцию .parallel() на любом экземпляре Stream, IntStream, LongStream или DoubleStream.

Оба способа эквивалентны по поведению: под капотом они вызывают StreamSupport.stream(spliterator, true), передавая булев флаг параллелизма parallel = true. Проверить, является ли стрим параллельным, можно методом stream.isParallel().

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

import java.util.Arrays;
import java.util.List;
import java.util.stream.IntStream;
import java.util.stream.Stream;

public class CreateParallelStreamDemo {

    public static void main(String[] args) {
        // Способ 1: Из коллекции через parallelStream()
        List<String> names = List.of("Анна", "Борис", "Виктор", "Дарья");
        names.parallelStream()
            .map(String::toUpperCase)
            .forEach(System.out::println);

        // Способ 2: Через метод .parallel() на существующем стриме
        Stream.of("Красный", "Зеленый", "Синий")
            .parallel()
            .forEach(color -> System.out.printf("%s в потоке %s%n",
                color, Thread.currentThread().getName()));

        // Способ 3: Из примитивного диапазона
        long sum = IntStream.rangeClosed(1, 1_000_000)
            .parallel()
            .sum();
        System.out.println("Сумма: " + sum);

        // Способ 4: Из обычного массива
        int[] numbers = {10, 20, 30, 40, 50};
        Arrays.stream(numbers)
            .parallel()
            .forEach(n -> System.out.print(n + " "));
    }
}

🟡 Middle Level

Низкоуровневое создание через StreamSupport

Класс java.util.stream.StreamSupport предоставляет низкоуровневые фабричные методы для создания стримов из любых пользовательских источников с помощью Spliterator:

public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel)

Реализация метода по умолчанию в интерфейсе java.util.Collection:

default Stream<E> parallelStream() {
    return StreamSupport.stream(spliterator(), true);
}

Правило «Последний вызов побеждает» (Last Call Wins)

В Stream API не существует частично параллельных потоков. Конвейер стрима не может выполнять одну часть операций параллельно, а вторую — последовательно в рамках одного пайплайна:

List<Integer> list = List.of(1, 2, 3, 4, 5);

list.stream()
    .parallel()      // Запрос параллельного режима
    .filter(x -> x > 2)
    .sequential()    // Переключение обратно в последовательный режим!
    .map(x -> x * 2)
    .forEach(System.out::println);

В данном примере весь стрим выполнится строго последовательно. Флаг параллелизма — это единое булево состояние (parallel), хранящееся в голове пайплайна (AbstractPipeline.sourceStage.parallel). Каждая терминальная операция опрашивает этот единый флаг в момент своего запуска: побеждает метод, вызванный последним перед терминальной операцией.

Оптимизация unordered() для параллельных стримов

Если источник стрима обладает характеристикой упорядоченности (ORDERED, например List, LinkedHashSet или отсортированный массив), рантайм ForkJoinPool вынужден тратить ресурсы на соблюдение исходного порядка элементов при объединении результатов.

Если порядок вывода не важен, вызов .unordered() снимает это ограничение:

List<String> largeList = getLargeDataset();

// unordered() позволяет потокам сливать результаты без сохранения порядка вставки
List<String> result = largeList.parallelStream()
    .unordered()
    .distinct() // distinct() без порядка работает в разы быстрее!
    .toList();

🔴 Senior Level

Управление пулом исполнения: изоляция в Custom ForkJoinPool

По умолчанию все параллельные стримы разделяют общесистемный ForkJoinPool.commonPool(). В высоконагруженных серверных средах (например, в микросервисах на Spring Boot) это может привести к взаимной блокировке задач.

Для изоляции вычислений задачу оборачивают в выделенный пул:

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

public class CustomForkJoinPoolDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        List<Integer> data = IntStream.rangeClosed(1, 100_000).boxed().toList();

        // Создаем выделенный пул из 4 рабочих потоков
        ForkJoinPool dedicatedPool = new ForkJoinPool(4);

        try {
            long total = dedicatedPool.submit(() ->
                // Внутри submit() поток-воркер принадлежит dedicatedPool.
                // AbstractPipeline определяет текущий контекст через ForkJoinTask.getPool()!
                data.parallelStream()
                    .filter(x -> x % 2 == 0)
                    .mapToLong(x -> (long) x * x)
                    .sum()
            ).get();

            System.out.println("Результат в кастомном пуле: " + total);
        } finally {
            dedicatedPool.shutdown();
        }
    }
}

Механизм перехвата пула:

Когда вызывается терминальная операция параллельного стрима, рантайм вызывает ForkJoinTask.fork(). Реализация ForkJoinTask проверяет:

  • Если текущий вызывающий поток является экземпляром ForkJoinWorkerThread, задача регистрируется в очереди текущего пула (Thread.currentThread().pool).
  • Если поток является внешним (например, main или HTTP-воркер Tomcat), задача отправляется в глобальный ForkJoinPool.commonPool().

Важно: Передача parallelStream() в кастомный пул через customPool.submit() является недокументированным поведением реализации OpenJDK HotSpot. Спецификация языка формально этого не гарантирует, однако в реальных проектах это общепринятый стандарт изоляции нагрузки.

Настройка глобального параллелизма через JVM Flag

Размер глобального пула ForkJoinPool.commonPool() настраивается системным свойством JVM:

-Djava.util.concurrent.ForkJoinPool.common.parallelism=8

По умолчанию значение равно Runtime.getRuntime().availableProcessors() - 1.

Senior Warning: Изменение этого системного свойства глобально влияет на все компоненты JVM, включая CompletableFuture (без указания Executor) и параллельные компиляторы/рантаймы библиотек. В production менять этот флаг не рекомендуется; лучше использовать отдельные пулы потоков (ExecutorService / Custom ForkJoinPool).

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

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

«Параллельный стрим создается либо методом collection.parallelStream(), либо добавлением промежуточной операции .parallel() к любому стриму. Под капотом они вызывают StreamSupport.stream(spliterator, true). В пайплайне не бывает частичного параллелизма: действует правило «последний вызов побеждает» — если в конце цепочки стоит .sequential(), весь конвейер выполнится последовательно. По умолчанию вычисления идут в ForkJoinPool.commonPool(), но для изоляции нагрузки их можно отправить в отдельный ForkJoinPool через customPool.submit(...).»

Сравнение способов создания

Способ создания Пример Когда использовать
Collection.parallelStream() list.parallelStream() Стандартный способ для любых коллекций
Stream.parallel() IntStream.range(1, 100).parallel() Для диапазонов, массивов, генераторов
StreamSupport.stream(...) StreamSupport.stream(mySpliterator, true) При создании кастомных структур данных и сплитераторов
customPool.submit(() -> ...) pool.submit(() -> s.parallel().collect(...)) Для изоляции тяжелой нагрузки от commonPool

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

1. Что произойдет, если в стриме написать .parallel().filter(...).sequential().map(...).parallel()?

  • Стрим выполнится полностью параллельно. В Stream API состояние параллелизма задается единым флагом в корневом объекте AbstractPipeline. Метод, вызванный последним перед терминальной операцией, перезаписывает этот флаг. Частичного параллелизма внутри одного стрима не существует.

2. В чём разница между list.parallelStream() и StreamSupport.stream(list.spliterator(), true)?

  • Никакой разницы нет. По умолчанию реализация метода Collection.parallelStream() в JDK состоит ровно из одной строки: return StreamSupport.stream(spliterator(), true);.

3. Зачем вызывать .unordered() перед .parallel() на упорядоченных коллекциях?

  • По умолчанию у списков и отсортированных множеств установлен флаг Spliterator.ORDERED. Параллельный стрим при наличии этого флага обязан сохранять исходный порядок элементов при слиянии подзадач, что приводит к блокировкам и синхронизациям в ForkJoinPool. Вызов .unordered() снимает это требование, позволяя потокам сбрасывать результаты в произвольном порядке, что дает значительный прирост скорости при операциях distinct(), limit() и findAny().

4. Почему в контейнерах (Docker/Kubernetes) commonPool() может получить меньше потоков, чем ожидалось?

  • До Java 10 (и некоторых ранних версий Java 8/11) JVM не умела корректно читать cgroups лимиты CPU контейнера и определяла ядра хостовой машины.
  • В современных JVM (Java 17/21) Runtime.getRuntime().availableProcessors() учитывает CPU limits контейнера (--cpus). Если контейнеру выделено 2 ядра CPU, размер commonPool() составит 2 - 1 = 1 рабочий поток (плюс вызывающий поток), из-за чего параллелизм практически не даст выигрыша.

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

  • 🚩 Попытка распараллелить только часть конвейера вызовами .parallel() и .sequential() в надежде, что один фильтр выполнится параллельно, а маппинг последовательно.
  • 🚩 Забывание вызвать shutdown() на экземпляре new ForkJoinPool(), приводящее к утечке системных потоков и памяти.
  • 🚩 Запуск параллельного стрима без терминальной операции (стрим просто не запустится из-за ленивости).
  • 🚩 Изменение глобального системного свойства ForkJoinPool.common.parallelism в многопоточном сервисе, что ломает поведение других библиотек.

Связанные темы