Як створити parallel stream

Паралельний стрім можна створити двома основними способами:

🟢 Junior Level

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

Паралельний стрім можна створити двома основними способами:

Безпосередньо з колекції: викликавши метод collection.parallelStream().
Перетворенням наявного стріма: викликавши проміжну операцію .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 у багатопотоковому сервісі, що ламає поведінку інших бібліотек.