Що таке паралельні стріми

Під капотом паралельні стріми використовують спільний пул потоків 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 — кеш-локальність). Якщо черга порожня, він краде («краде / steal») чужі завдання з початку їхніх черг (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