Раздел 8. Stream API и функциональный стиль в Java

Глава 6: Parallel Stream

Механика ForkJoinPool.commonPool()

Вызов .parallelStream() или .parallel() на потоке создаёт иллюзию простоты: система сама разделит работу между ядрами процессора, ускорит выполнение, вернёт результат. Реальность сложнее. Параллельные потоки в Java построены на ForkJoinPool — специализированном механизме для задач с шаблоном "разделяй и властвуй", и их эффективность зависит от понимания внутренней механики.


ForkJoinPool: архитектура общего пула

ForkJoinPool.commonPool() — статический пул, создаваемый при загрузке класса ForkJoinTask. Его размер по умолчанию равен количеству доступных процессоров минус один (оставляя один поток для основной работы), но не менее одного. Максимальный размер ограничен 32767 потоками, но на практике редко превышает десятки.

Этот пул — разделяемый ресурс. Он используется не только для parallelStream, но и для CompletableFuture.async, Arrays.parallelSort, RecursiveTask и других компонентов стандартной библиотеки. Исчерпание пула блокирующими задачами парализует всё приложение.


Разделение данных: Spliterator.trySplit()

Ключ к параллелизму — способность разделить источник данных на независимые части. Эту функцию выполняет Spliterator.trySplit():

public interface Spliterator<T> {
    Spliterator<T> trySplit();  // Возвращает новый Spliterator для части данных или null
    void forEachRemaining(Consumer<? super T> action);
    boolean tryAdvance(Consumer<? super T> action);
    long estimateSize();
    int characteristics();
}

Метод trySplit() пытается разделить оставшиеся элементы пополам. Если успешно — возвращает новый Spliterator для первой половины, текущий продолжает обрабатывать вторую. Если данных мало или разделение невозможно — возвращает null, сигнализируя, что эту часть нужно обрабатывать последовательно.

Качество разделения определяет эффективность параллелизма:
Идеальные источники (ArrayList, массивы, IntStream.range): знают свой размер, поддерживают произвольный доступ. ArrayListSpliterator вычисляет середину как (lo + hi) >>> 1 и создаёт новый сплитератор для поддиапазона за O(1).
Приемлемые источники (HashSet, TreeSet): HashSet разделяется по бакетам хеш-таблицы. Разделение неравномерное (некоторые бакеты пусты), но работает. TreeSet использует структуру дерева для разделения.
Проблемные источники (LinkedList, Stream.iterate, Stream.generate, BufferedReader.lines()): не поддерживают произвольный доступ. LinkedList должен проходить узлы от начала до середины для разделения — O(n) на каждый split.

Stream.iterate и generate вообще не делятся, возвращая null из trySplit(). Параллельная обработка таких источников сводится к последовательной с накладными расходами на координацию.


Работа воркеров и кража задач

ForkJoinPool использует модель "work-stealing" (кража работы). Каждый поток-пул имеет локальную двустороннюю очередь (deque) задач. Новые задачи добавляются в голову очереди владельцем, выполняются с головы (LIFO — последняя добавленная первой). Когда поток опустошает свою очередь, он "ворует" задачи с хвоста очереди другого потока (FIFO — старые задачи), уменьшая contention.

Алгоритм для parallelStream:
Инициация: терминальная операция оборачивает конвейер в ForkJoinTask и отправляет в пул.
Разделение: корневой Spliterator делится рекурсивно, пока части достаточно малы или достигнут лимит параллелизма.
Выполнение: воркеры забирают подзадачи, обрабатывают свои сегменты данных через Spliterator.forEachRemaining.
Слияние: результаты подзадач комбинируются через Collector.combiner или аналогичный механизм.
Завершение: финальный результат возвращается вызывающему потоку.

Критично понимать: разделение происходит до выполнения, не во время. Поток разбивается на сегменты, затем каждый сегмент обрабатывается целиком одним воркером. Это не "потоковая" параллелизация, где элементы распределяются по ядрам по мере готовности, а "батчевая": данные разделены, затем обработаны.


#Java #для_новичков #beginner #stream_api #ForkJoinPool #parallelStream

Опасность блокирующих задач

Самый разрушительный антипаттерн — блокирующие операции внутри parallelStream.

Рассмотрим сценарий:

List<Result> results = urls.parallelStream()
    .map(url -> {
        try {
            return httpClient.fetch(url);  // Блокирующий HTTP-запрос, 500мс
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    })
    .collect(toList());

При 100 URL и пуле размером 8 все воркеры быстро блокируются в ожидании сети. Оставшиеся 92 URL стоят в очереди. Но хуже: другие компоненты приложения, использующие commonPool (CompletableFuture, другие parallelStream), не получают потоков. Система "замораживается" — не от зависания, а от исчерпания ресурса.

Ещё хуже с Thread.sleep:

// Имитация тяжёлой работы
IntStream.range(0, 1000).parallel()
    .map(i -> {
        try { Thread.sleep(100); } catch (InterruptedException e) { }
        return i * 2;
    })
    .collect(toList());

Все воркеры спят. Никакая другая задача в пуле не выполняется. Это эквивалентно deadlock для всего, что зависит от commonPool.

Диагностика в production:
Мониторинг ForkJoinPool.commonPool() через JMX: getActiveThreadCount(), getQueuedTaskCount(), getStealCount().
Thread dumps: поиск потоков с именем вида ForkJoinPool.commonPool-worker-N, ожидающих в Object.wait(), Thread.sleep(), или блокирующих I/O.
Профилирование: высокое время ожидания в ForkJoinTask.join() при отсутствии CPU-bound работы указывает на блокировки.


Альтернативы для блокирующих операций

Для I/O-bound задач parallelStream неприменим.

Используйте:
CompletableFuture с кастомным пулом:

ExecutorService ioPool = Executors.newFixedThreadPool(50);  // Много потоков, не боится блокировок

List<CompletableFuture<Result>> futures = urls.stream()
    .map(url -> CompletableFuture.supplyAsync(() -> fetch(url), ioPool))
    .collect(toList());

List<Result> results = futures.stream()
    .map(CompletableFuture::join)
    .collect(toList());

ioPool.shutdown();
Virtual Threads (Java 21+):
java
Copy
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    List<StructuredTaskScope.Subtask<Result>> subtasks = urls.stream()
        .map(url -> scope.fork(() -> fetch(url)))
        .collect(toList());
    
    scope.join().throwIfFailed();
    
    return subtasks.stream()
        .map(StructuredTaskScope.Subtask::get)
        .collect(toList());
}

Виртуальные потоки не блокируют носитель потоков ОС при блокировке Java-потока, делая блокирующие операции дешёвыми.


Контроль над commonPool

Размер пула можно настроить через системное свойство:

// При запуске JVM
-Djava.util.concurrent.ForkJoinPool.common.parallelism=16

// Или программно, но только до первого использования пула
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "16");

Но увеличение размера не решает проблему блокирующих задач — оно лишь откладывает исчерпание. Правильное решение — изоляция: блокирующие задачи в отдельном пуле, CPU-bound задачи в commonPool.


#Java #для_новичков #beginner #stream_api #ForkJoinPool #parallelStream

Раздел 8. Stream API и функциональный стиль в Java

Глава 6: Parallel Stream

Условия эффективности параллельного stream

Параллельные потоки ускоряют не всегда. В худшем случае они замедляют выполнение, увеличивают потребление памяти и вносят race conditions. Эффективность зависит от четырёх факторов: характеристик источника, стоимости операции над элементом, чистоты функций и структуры конвейера.


Источник данных: качество разделения

Первый и решающий фактор — способность источника к эффективному разделению. Как обсуждалось ранее, Spliterator.trySplit() определяет, насколько равномерно данные распределятся между воркерами.
Идеальные источники демонстрируют три свойства: точное знание размера (SIZED), поддержка произвольного доступа (RANDOM_ACCESS), быстрое разделение (SUBSIZED).

ArrayList, массивы примитивов, IntStream.range обладают всеми тремя. Их разделение работает за константное время, создавая сбалансированные сегменты.

// ArrayList: O(1) разделение, равномерная нагрузка
List<Book> books = new ArrayList<>(100000);
books.parallelStream()  // Эффективен
    .map(this::expensiveAnalysis)
    .collect(toList());

Приемлемые источники работают хуже, но применимы. HashSet разделяется по бакетам хеш-таблицы. Если распределение хешей равномерное и заполнение высокое, разделение качественное. Но при коллизиях или неравномерном заполнении некоторые сегменты становятся существенно больше других, нарушая балансировку.

Проблемные источники лишены возможности эффективного разделения. LinkedList требует O(n) для поиска середины. Stream.iterate и Stream.generate не делятся вообще — каждый элемент порождается последовательно, и весь поток обрабатывается одним воркером, независимо от вызова parallel().

// LinkedList: разделение O(n), часто деградирует к последовательному
LinkedList<Book> linkedBooks = new LinkedList<>();
linkedBooks.parallelStream()  // Нет выигрыша, возможен проигрыш
    .map(this::expensiveAnalysis)
    .collect(toList());

// Stream.iterate: не делится, parallel бесполезен
Stream.iterate(0, n -> n + 1)
    .parallel()  // Игнорируется
    .limit(1000)
    .map(this::expensiveComputation)
    .collect(toList());

Источники ввода-вывода (Files.lines, BufferedReader.lines) представляют особый случай. Они не поддерживают разделение, но могут быть обёрнуты в Stream с буферизацией. Параллелизм здесь достигается через промежуточную коллекцию: чтение последовательное, обработка параллельная.


Стоимость операции: порог эффективности

Даже при идеальном источнике параллелизм имеет накладные расходы: создание задач ForkJoinTask, разделение Spliterator, синхронизация при слиянии результатов, кэш-коэрентность между ядрами. Эти затраты должны компенсироваться выигрышем от параллельного выполнения.

Эмпирический порог — порядка 10 микросекунд на элемент. Если операция дешевле, накладные расходы перевешивают выгоду. Если дороже — параллелизм эффективен.

// Слишком дёшево: параллелизм замедлит
List<Integer> doubled = numbers.parallelStream()
    .map(n -> n * 2)  // Одна инструкция процессора
    .collect(toList());

// Достаточно дорого: параллелизм ускорит
List<Result> analyzed = documents.parallelStream()
    .map(doc -> nlpPipeline.analyze(doc))  // 50-100 мс на документ
    .collect(toList());

Сложность оценки в том, что "стоимость" включает не только CPU-инструкции, но и кэш-промахи, аллокации, вызовы методов. Профилирование (JMH, async-profiler) необходимо для точного определения порога в конкретном контексте.


#Java #для_новичков #beginner #stream_api #ForkJoinPool #parallelStream

Отсутствие shared mutable state

Параллелизм превращает скрытые баги в явные катастрофы. Код, работающий корректно в последовательном потоке, может давать неверные результаты или зависать при parallel().

Race condition в accumulator:

// Антипаттерн: общий счётчик
AtomicInteger counter = new AtomicInteger(0);

List<Result> results = data.parallelStream()
    .map(d -> {
        counter.incrementAndGet();  // Потокобезопасен, но не бесплатен
        return process(d);
    })
    .collect(toList());

AtomicInteger обеспечивает корректность, но каждый incrementAndGet() требует атомарной операции и кэш-коэрентности между ядрами. При высоком contention производительность падает ниже последовательной версии.

Непотокобезопасная коллекция в collect:

// Катастрофа: ArrayList не потокобезопасен
List<Result> results = data.parallelStream()
    .map(this::process)
    .collect(ArrayList::new, List::add, List::addAll);  // Race condition!

Стандартный toList() использует Collector с правильным combiner, создающим локальные ArrayList для каждого сегмента и сливающим их в конце. Прямое использование ArrayList::new в collect нарушает этот протокол.

Изменяемые ключи в groupingBy:

// Опасность: изменяемый ключ после группировки
Map<MutableAuthor, List<Book>> byAuthor = books.parallelStream()
    .collect(groupingBy(MutableAuthor::new));

// Позже в другом потоке...
byAuthor.keySet().iterator().next().setName("New");  // Непредсказуемое поведение

Параллелизм увеличивает вероятность одновременного доступа к изменяемым структурам. Иммутабельность ключей и элементов становится не рекомендацией, а требованием.


#Java #для_новичков #beginner #stream_api #ForkJoinPool #parallelStream

Отсутствие барьеров в конвейере

Барьер (synchronization point) — операция, требующая видимости всех элементов или глобальной координации. Барьеры разрушают параллелизм, заставляя воркеров ждать друг друга.

sorted: полный барьер

Операция sorted() требует всех элементов для сортировки. В параллельном потоке каждый сегмент сортируется локально, затем результаты сливаются (merge) в глобально отсортированную последовательность. Слияние требует координации и дополнительной памяти.

// Барьер: локальная сортировка + глобальное слияние
List<Book> sorted = books.parallelStream()
    .sorted(comparing(Book::year))  // Накладные расходы на merge
    .collect(toList());

Если конвейер содержит sorted без последующих операций, параллелизм может быть оправдан для дорогих предшествующих операций. Но sorted + limit — особенно неэффективная комбинация: все элементы сортируются, хотя нужны только первые N.

limit и findFirst: частичные барьеры

limit(n) в параллельном потоке создаёт глобальный счётчик оставшихся элементов. Когда один воркер достигает лимита, другие должны быть уведомлены для остановки. Это требует синхронизации и снижает эффективность.
findFirst() требует упорядоченности. Если источник ORDERED, воркеры должны координироваться для определения, кто нашёл "первый" элемент.

Если источник неупорядочен или порядок не важен, findAny() предпочтительнее — он возвращает любой найденный элемент без координации.

// Плохо: упорядоченный источник + findFirst в parallel
Optional<Book> first = books.parallelStream()
    .filter(b -> b.year() > 2000)
    .findFirst();  // Требует проверки всех предшествующих сегментов

// Лучше: findAny для неупорядоченных задач
Optional<Book> any = books.parallelStream()
    .filter(b -> b.year() > 2000)
    .findAny();  // Первый найденный в любом сегменте

distinct: барьер с состоянием

distinct() в параллельном потоке требует глобального множества уникальных элементов, доступного всем воркерам. Реализация использует ConcurrentHashMap, что добавляет накладные расходы на синхронизацию. Для больших потоков с высокой кардинальностью это может быть медленнее последовательной версии с HashSet.


Когда parallelStream применим

Суммируя критерии, эффективный сценарий для parallelStream:
Источник: ArrayList, массив, IntStream.range с большим размером (10 000+ элементов)
Операция: CPU-bound, дорогая (> 10 мкс), без блокировок
Конвейер: stateless операции, без sorted, limit, distinct или с ними в конце
Данные: иммутабельные, без shared mutable state
Цель: агрегация в коллектор с эффективным combiner
Пример подходящей задачи: анализ миллиона документов, извлечение признаков, подсчёт статистики.
Пример неподходящей задачи: фильтрация списка идентификаторов с простым предикатом, преобразование в строки, лимит первых десяти.


Измерение прежде оптимизации

Предположения о производительности часто ошибочны. Единственный надёжный метод — измерение с репрезентативными данными:

// JMH-бенчмарк для сравнения
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
public class StreamBenchmark {
    
    @State(Scope.Thread)
    public static class Data {
        List<Book> books = generateBooks(100000);
    }
    
    @Benchmark
    public List<Result> sequential(Data d) {
        return d.books.stream()
            .map(this::expensiveTransform)
            .collect(toList());
    }
    
    @Benchmark
    public List<Result> parallel(Data d) {
        return d.books.parallelStream()
            .map(this::expensiveTransform)
            .collect(toList());
    }
}

Микробенчмарки без JMH ненадёжны из-за JIT-оптимизаций, GC-пauses и прогрева кэша. Реальное приложение требует мониторинга в production: метрики latency, throughput, использование CPU, профилирование hot paths.


#Java #для_новичков #beginner #stream_api #ForkJoinPool #parallelStream

Раздел 8. Stream API и функциональный стиль в Java

Глава 6: Parallel Stream

Гонка параллельных stream в «Библиотеке»

Подготовка: расширение модели Book

Добавьте поле для имитации дорогой операции:

public class Book {
    private final String title;
    private final String author;
    private final int year;
    private final List<String> genres;
    private final int complexity;  // «сложность» проверки, влияет на время фильтрации
    
    public Book(String title, String author, int year, 
                List<String> genres, int complexity) {
        this.title = title;
        this.author = author;
        this.year = year;
        this.genres = genres;
        this.complexity = complexity;
    }
    
    // геттеры...
    
    public boolean isExpensiveFantasyCheck() {
        // Имитация дорогой операции: чем выше complexity, тем дольше
        try {
            Thread.sleep(complexity);  // миллисекунды
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return genres.contains("Фантастика");
    }
}

Базовый эксперимент: поиск фантастики

Метод для измерения

public class ParallelStreamBenchmark {
    
    public static List<String> findFantasyTitlesSequential(List<Book> books) {
        return books.stream()
            .filter(Book::isExpensiveFantasyCheck)
            .map(Book::getTitle)
            .collect(Collectors.toList());
    }
    
    public static List<String> findFantasyTitlesParallel(List<Book> books) {
        return books.parallelStream()
            .filter(Book::isExpensiveFantasyCheck)
            .map(Book::getTitle)
            .collect(Collectors.toList());
    }
    
    public static long measure(Runnable task) {
        long start = System.nanoTime();
        task.run();
        long end = System.nanoTime();
        return (end - start) / 1_000_000;  // миллисекунды
    }
}

Три сценария: малые, средние, большие данные

Сценарий А: 10 элементов, лёгкая операция (complexity = 0)

List<Book> tinyLibrary = IntStream.range(0, 10)
    .mapToObj(i -> new Book("Книга " + i, "Автор " + i, 
        2000 + i, Arrays.asList(i % 2 == 0 ? "Фантастика" : "Роман"), 0))
    .collect(Collectors.toList());

long seqTime = measure(() -> findFantasyTitlesSequential(tinyLibrary));
long parTime = measure(() -> findFantasyTitlesParallel(tinyLibrary));

System.out.println("10 элементов, лёгкая операция:");
System.out.println("  Последовательно: " + seqTime + " мс");
System.out.println("  Параллельно: " + parTime + " мс");
System.out.println("  Накладка: " + (parTime - seqTime) + " мс");

Ожидаемый результат: Параллельная версия медленнее в 2–10 раз.

Почему:
Разбиение потока на сегменты (splitting)
Передача задач в ForkJoinPool
Синхронизация при сборке результата (ConcurrentHashMap для toList)
Координация потоков дороже, чем сама работа

Сценарий Б: 10 000 элементов, лёгкая операция

List<Book> mediumLibrary = IntStream.range(0, 10_000)
    .mapToObj(i -> new Book("Книга " + i, "Автор " + i,
        2000 + i % 100, 
        Arrays.asList(i % 3 == 0 ? "Фантастика" : "Роман"), 0))
    .collect(Collectors.toList());

// те же измерения

Ожидаемый результат: Примерно паритет или лёгкий проигрыш параллельной версии.

Почему:
Накладка на координацию амортизируется
Но операция всё ещё слишком лёгкая — переключение контекста не окупается

Сценарий В: 100 000 элементов, тяжёлая операция (complexity = 1 мс)

List<Book> largeLibrary = IntStream.range(0, 100_000)
    .mapToObj(i -> new Book("Книга " + i, "Автор " + i,
        2000 + i % 100,
        Arrays.asList(i % 3 == 0 ? "Фантастика" : "Роман"), 1))
    .collect(Collectors.toList());

// те же измерения

Ожидаемый результат: Параллельная версия быстрее в 2–4 раза (на многоядерной машине).

Почему:
Общая работа: 100 000 × 1 мс = 100 секунд последовательно
На 8 ядрах: теоретически ~12.5 секунд + накладка
Накладка амортизируется на большом объёме вычислений


#Java #для_новичков #beginner #stream_api #ForkJoinPool #parallelStream #практика

Проблема с неправильными источниками

Демонстрация: LinkedList vs ArrayList

List<Book> arrayList = new ArrayList<>(largeLibrary);
List<Book> linkedList = new LinkedList<>(largeLibrary);

// ArrayList — хорошее разбиение
long arrayListParallel = measure(() -> 
    arrayList.parallelStream().filter(...).collect(...));

// LinkedList — плохое разбиение
long linkedListParallel = measure(() ->
    linkedList.parallelStream().filter(...).collect(...));

System.out.println("ArrayList parallel: " + arrayListParallel + " мс");
System.out.println("LinkedList parallel: " + linkedListParallel + " мс");

Ожидаемый результат: LinkedList в 3–10 раз медленнее из-за невозможности эффективного разбиения.


Интеграция в проект «Библиотеке»

Задача: параллельный анализ большой библиотеки

Добавьте в Library метод для статистики с тяжёлыми вычислениями:

public Map<String, Double> analyzeGenreComplexityParallel() {
    return books.parallelStream()
        .collect(Collectors.groupingByConcurrent(
            book -> book.getGenres().get(0),  // основной жанр
            Collectors.averagingInt(Book::getComplexity)
        ));
}

Сравните с последовательной версией на 100 000 книгах.


Практические задания

Задача 1: неправильное использование

Напишите код, который демонстрирует проблемы:
parallelStream().forEach(System.out::println) — беспорядок в выводе
parallelStream().sorted() — избыточная синхронизация
parallelStream() для записи в общий StringBuilder — race condition
Объясните, почему каждый случай ломается.

Задача 2: правильный кастомный коллектор (звёздочка)

Реализуйте parallelStream-совместимый коллектор для Map<Author, List<Book>>, который:
Использует ConcurrentHashMap для аккумулятора
Корректно обрабатывает combiner для слияния частичных результатов
Сохраняет порядок книг внутри автора
Сравните производительность с groupingByConcurrent.

#Java #для_новичков #beginner #stream_api #ForkJoinPool #parallelStream #практика