Раздел 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 #практика

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

Глава 7: Stream API в экосистеме Java. Интеграционные паттерны

Примитивные стримы (IntStream, LongStream)

Одно из фундаментальных ограничений Java — отсутствие обобщений (generics) для примитивных типов. Stream<T> может содержать только ссылочные типы, что вынуждает упаковывать int в Integer, long в Long, double в Double. Эта операция, называемая боксингом (boxing), создаёт объекты-обёртки в куче, увеличивает нагрузку на GC и снижает производительность кэша процессора из-за разыменования указателей.

Примитивные стримы — IntStream, LongStream, DoubleStream — решают эту проблему через специализацию. Они работают с сырыми примитивами, избегая аллокаций и обеспечивая плотное размещение данных в памяти.


Генерация числовых последовательностей

range и rangeClosed

IntStream.range(start, end) генерирует последовательность от start (включительно) до end (исключительно). rangeClosed включает обе границы.

// Индексы для обхода: 0, 1, 2, ..., 99
IntStream indices = IntStream.range(0, 100);

// Дни месяца: 1, 2, ..., 31
IntStream days = IntStream.rangeClosed(1, 31);

Эти методы возвращают упорядоченные, конечные потоки с известным размером — оптимальные характеристики для параллелизма. Spliterator такого потока делится идеально: середина вычисляется как (start + end) >>> 1, создавая сбалансированные сегменты.

iterate и generate

Бесконечные последовательности создаются через iterate (с заданным началом и функцией следующего элемента) или generate (с поставщиком значений).

// Степени двойки: 1, 2, 4, 8, ... с ограничением
IntStream powersOfTwo = IntStream.iterate(1, n -> n * 2)
    .limit(20);

// Случайные числа
IntStream randomInts = new Random().ints();  // Бесконечный
IntStream limitedRandom = new Random().ints(100, 0, 1000);  // 100 чисел от 0 до 999

iterate не поддерживает эффективное разделение — каждый элемент зависит от предыдущего, что делает параллелизм бесполезным. Random.ints() оптимизирован для параллельного использования через специализированный


Spliterator.

Агрегация примитивов: статистика без промежуточных объектов

Примитивные стримы предоставляют терминальные операции, возвращающие примитивные значения без боксинга:

int sum = IntStream.range(1, 101).sum();  // 5050, возвращает int

long count = LongStream.range(0, 1_000_000).count();  // 1000000L

OptionalDouble average = IntStream.of(10, 20, 30).average();  // OptionalDouble[20.0]

OptionalInt max = IntStream.of(3, 1, 4, 1, 5).max();  // OptionalInt[5]
OptionalInt, OptionalLong, OptionalDouble — примитивные версии Optional, избегающие обёртки. Методы: isPresent(), getAsInt() / getAsLong() / getAsDouble(), orElse(int), orElseGet(IntSupplier).

summaryStatistics: комплексная агрегация

Для многократной статистики за один проход:

IntSummaryStatistics stats = IntStream.of(10, 20, 30, 40, 50)
    .summaryStatistics();

System.out.println(stats.getCount());    // 5
System.out.println(stats.getSum());      // 150
System.out.println(stats.getMin());      // 10
System.out.println(stats.getMax());      // 50
System.out.println(stats.getAverage());  // 30.0

IntSummaryStatistics — mutable контейнер, накапливающий данные без промежуточных коллекций. Эффективен для больших потоков: вместо boxed().collect(summarizingInt(Integer::intValue)) используется прямая агрегация примитивов.


#Java #для_новичков #beginner #stream_api #IntStream #LongStream

Преобразования между типами стримов

boxed: примитив к ссылочному

Когда нужен полноценный Stream<T> с методами объектов:

Stream<Integer> boxed = IntStream.range(0, 100).boxed();

// Применение сложных операций, недоступных для примитивов
List<String> labels = IntStream.range(0, 100)
    .boxed()
    .map(i -> "Item #" + i)
    .collect(toList());

boxed() создаёт объект для каждого примитива — дорогая операция для больших потоков. Применять осознанно, когда объектная семантика необходима.

mapToObj: примитив к произвольному объекту

Более эффективный путь создания объектов из примитивов:

List<Book> booksByIndex = IntStream.range(0, library.size())
    .mapToObj(i -> new Book("Book " + i, "Author " + i, 2000 + i))
    .collect(toList());
mapToObj(IntFunction<R>) принимает функцию, создающую объект из индекса, без промежуточного boxed().
mapToInt, mapToLong, mapToDouble: объект к примитиву

Обратное преобразование для агрегации:

// Суммарное количество страниц
int totalPages = library.stream()
    .mapToInt(Book::pageCount)  // Book -> int
    .sum();

// Средняя цена в long-центах для точности
long averagePriceCents = library.stream()
    .mapToLong(b -> b.price().multiply(BigDecimal.valueOf(100)).longValue())
    .average()
    .orElse(0L);

Эти методы критичны для производительности: mapToInt(Book::pageCount).sum() вместо map(Book::pageCount).reduce(0, Integer::sum) избегает миллионов аллокаций Integer.


Сценарий: работа с индексами

Классическая задача, не имеющая прямого решения в Stream<T> — доступ к индексу элемента. Примитивные стримы предоставляют элегантный обход:

// Нумерация элементов с сохранением порядка
List<String> numbered = IntStream.range(0, books.size())
    .mapToObj(i -> (i + 1) + ". " + books.get(i).title())
    .collect(toList());

// Поиск с индексом
OptionalInt firstExpensiveIndex = IntStream.range(0, books.size())
    .filter(i -> books.get(i).price().compareTo(THRESHOLD) > 0)
    .findFirst();  // Возвращает индекс, не объект

Для List с произвольным доступом это эффективно. Для LinkedList или потоков без размера — требуется промежуточная материализация:

// Обобщённый подход для любого Stream
Stream<IndexedValue<Book>> withIndex = StreamUtils.zipWithIndex(books.stream());

// Или через AtomicInteger (менее элегантно, но без внешних библиотек)
AtomicInteger index = new AtomicInteger(0);
List<String> numbered = books.stream()
    .map(b -> index.getAndIncrement() + ": " + b.title())
    .collect(toList());  // Не параллелизуемо!

Сценарий: статистические расчёты

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

// Корреляция двух наборов данных (упрощённо)
double[] x = fetchDataX();
double[] y = fetchDataY();

double meanX = Arrays.stream(x).average().orElse(0.0);
double meanY = Arrays.stream(y).average().orElse(0.0);

double covariance = IntStream.range(0, x.length)
    .mapToDouble(i -> (x[i] - meanX) * (y[i] - meanY))
    .average()
    .orElse(0.0);
Arrays.stream(double[]) создаёт DoubleStream без копирования массива — эффективный доступ к существующим данным.

Параллелизм примитивных стримов

IntStream.parallel(), LongStream.parallel() наследуют все особенности ForkJoinPool. Дополнительное преимущество — отсутствие boxing при слиянии результатов. sum(), average(), summaryStatistics() в параллельном режиме используют примитивные аккумуляторы, минимизируя синхронизацию.

// Эффективная параллельная агрегация
long sum = LongStream.range(0, 100_000_000)
    .parallel()
    .sum();  // Разделение по сегментам, локальные суммы, финальное слияние

#Java #для_новичков #beginner #stream_api #IntStream #LongStream

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

Глава 7: Stream API в экосистеме Java. Интеграционные паттерны

Optional и Stream — братья по духу

Optional и Stream появились в Java 8 как единое архитектурное видение: переход от императивного управления состоянием к декларативной композиции операций.

Оба API разделяют ключевые принципы: ленивость вычислений, функциональная обработка отсутствия значения (пустой поток / пустой Optional), цепочечная композиция методов. Понимание их взаимодействия открывает идиомы, устраняющие громоздкие проверки if (x != null) и if (optional.isPresent()).


Optional.stream(): мост между мирами

Метод Optional.stream(), добавленный в Java 9, — один из наиболее недооценённых инструментов. Он превращает Optional<T> в Stream<T>: непустой Optional становится потоком из одного элемента, пустой — пустым потоком. Это позволяет интегрировать Optional в потоковые конвейеры без императивных ветвлений.

Императивный антипаттерн:

List<String> names = new ArrayList<>();
Optional<User> optionalUser = userRepository.findById(userId);

if (optionalUser.isPresent()) {
    User user = optionalUser.get();
    names.add(user.name());
}
// names содержит 0 или 1 элемент, логика размазана по условию

Проблемы: мутация внешней коллекции, побочный эффект в ifPresent, невозможность дальнейшей композиции. Код читается как инструкция, а не как намерение.

Декларативная трансформация:

List<String> names = userRepository.findById(userId)
    .stream()                    // Optional<User> -> Stream<User> (0 или 1 элемент)
    .map(User::name)             // Stream<String>
    .collect(toList());          // List<String>, размер 0 или 1

Здесь отсутствие значения обрабатывается естественно: пустой Optional порождает пустой поток, map не применяется, collect возвращает пустой список. Нет условий, нет ветвлений, нет мутации внешнего состояния.


Композиция Optional в потоковых цепочках

Сила Optional.stream() проявляется в сложных конвейерах, где несколько операций могут вернуть пустой результат:

// Цепочка: заказ -> клиент -> адрес -> город
List<String> cities = orders.stream()
    .map(Order::getCustomerId)
    .map(customerRepository::findById)     // Stream<Optional<Customer>>
    .flatMap(Optional::stream)             // Stream<Customer>, пустые отфильтрованы
    .map(Customer::getAddress)
    .map(Address::getCity)
    .distinct()
    .collect(toList());

Без Optional.stream() потребовались бы вложенные filter(Optional::isPresent) и map(Optional::get), или flatMap с лямбдой, возвращающей Stream.empty() или Stream.of(value).


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

flatMap для раскрытия вложенности

Когда Optional содержит объект, который сам возвращает поток или коллекцию, flatMap объединяет уровни вложенности:

// Найти все заказы пользователя по ID, если пользователь существует
Stream<Order> userOrders = userRepository.findById(userId)
    .stream()                              // Stream<User>
    .flatMap(u -> u.getOrders().stream());  // Stream<Order>

// Сбор в список, если пользователь найден
List<Order> orders = userOrders.collect(toList());

Здесь flatMap играет двойную роль: раскрывает Optional (через stream()) и раскрывает внутреннюю коллекцию заказов. Пустой Optional прерывает цепочку, возвращая пустой поток.

Более сложный сценарий — цепочка зависимых Optional:

// Было: глубокая вложенность ifPresent
optionalUser.ifPresent(user -> 
    user.getProfile().ifPresent(profile ->
        profile.getAvatar().ifPresent(avatar ->
            process(avatar)
        )
    )
);

// Стало: плоская композиция через stream и flatMap
userRepository.findById(userId)
    .stream()
    .flatMap(u -> u.getProfile().stream())
    .flatMap(p -> p.getAvatar().stream())
    .forEach(this::process);

Каждый flatMap обрабатывает потенциальное отсутствие значения, пропуская дальнейшие операции для пустых Optional. Результат — линейная цепочка вместо пирамиды вложенности.


Stream в Optional: reduce и find

Обратное направление — терминальные операции потока, возвращающие Optional. Это естественное завершение конвейера, где результат может отсутствовать:

Optional<Book> oldest = library.stream()
    .filter(b -> b.genre() == Genre.FICTION)
    .min(comparing(Book::year));  // Optional<Book>

// Интеграция обратно в поток
List<String> oldestTitle = oldest
    .stream()                     // Optional -> Stream
    .map(Book::title)
    .collect(toList());

findFirst(), findAny(), min(), max(), reduce() — все возвращают Optional, сигнализируя о возможности пустого результата. Композиция этих операций с Optional.stream() создаёт бесшовные цепочки обработки, где отсутствие данных — нормальный сценарий, не исключительная ситуация.


Обработка коллекций в Optional

Когда Optional содержит коллекцию, которая сама может быть пустой, Optional.stream() сочетается с flatMap для нормализации:

// Репозиторий возвращает Optional<List<Order>> — неудачный дизайн, но встречается
Optional<List<Order>> optionalOrders = orderRepository.findByCustomerId(id);

// Нормализация: Optional<List<Order>> -> Stream<Order>
Stream<Order> orders = optionalOrders
    .stream()                    // Stream<List<Order>>
    .flatMap(List::stream);       // Stream<Order>, пустой если Optional пуст или список пуст

// Или в одном выражении
List<Order> result = optionalOrders.stream()
    .flatMap(List::stream)
    .filter(Order::isActive)
    .collect(toList());

Этот паттерн устраняет необходимость проверять optionalOrders.isPresent() и optionalOrders.get().isEmpty() отдельно.


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

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

Глава 7: Stream API в экосистеме Java. Интеграционные паттерны

Рефакторинг legacy-кода в проекте «Библиотека»

Представьте, что в проекте «Библиотека» накопился такой метод — подсчёт статистики по авторам с множеством условий:

public Map<String, AuthorStats> calculateAuthorStatsLegacy(List<Book> books, int minYear) {
    Map<String, AuthorStats> result = new HashMap<>();
    
    for (Book book : books) {
        // Пропускаем старые книги
        if (book.getYear() < minYear) {
            continue;
        }
        
        // Пропускаем без жанров
        if (book.getGenres().isEmpty()) {
            continue;
        }
        
        // Пропускаем, если нет цены
        Double price = book.getPrice();
        if (price == null || price <= 0) {
            continue;
        }
        
        String author = book.getAuthor();
        AuthorStats stats = result.get(author);
        
        if (stats == null) {
            stats = new AuthorStats();
            stats.firstBookYear = book.getYear();
            result.put(author, stats);
        }
        
        // Обновляем статистику
        stats.bookCount++;
        stats.totalPages += book.getPages();
        stats.totalPrice += price;
        
        // Обновляем самую старую книгу
        if (book.getYear() < stats.firstBookYear) {
            stats.firstBookYear = book.getYear();
        }
        
        // Собираем уникальные жанры
        for (String genre : book.getGenres()) {
            if (!stats.genres.contains(genre)) {
                stats.genres.add(genre);
            }
        }
    }
    
    // Пост-обработка: считаем среднее
    for (AuthorStats stats : result.values()) {
        stats.averagePrice = stats.totalPrice / stats.bookCount;
    }
    
    return result;
}

static class AuthorStats {
    int bookCount;
    int totalPages;
    double totalPrice;
    double averagePrice;
    int firstBookYear;
    Set<String> genres = new HashSet<>();
}

Характеристики кода:
40+ строк в одном методе
4 уровня вложенности в пике
Смешение фильтрации, трансформации и агрегации
Два прохода: основной цикл + пост-обработка


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

Мутация состояния на каждом шагу

Шаг 1: Выделить источник данных в stream

Не меняем логику — только оборачиваем в stream:

public Map<String, AuthorStats> calculateAuthorStatsStep1(List<Book> books, int minYear) {
    Map<String, AuthorStats> result = new HashMap<>();
    
    books.stream().forEach(book -> {  // Заменили for-each на forEach
        // ... весь прежний код без изменений ...
    });
    
    // пост-обработка
    result.values().forEach(stats -> 
        stats.averagePrice = stats.totalPrice / stats.bookCount
    );
    
    return result;
}

Что достигли: Начали привыкать к синтаксису. Никаких функциональных преимуществ ещё нет — это подготовка.


Шаг 2: Заменить простые фильтры

Выносим continue как предикаты filter. Каждый фильтр — отдельная проверка:

public Map<String, AuthorStats> calculateAuthorStatsStep2(List<Book> books, int minYear) {
    Map<String, AuthorStats> result = new HashMap<>();
    
    books.stream()
        .filter(book -> book.getYear() >= minYear)           // было: if (year < minYear) continue
        .filter(book -> !book.getGenres().isEmpty())         // было: if (genres.isEmpty()) continue
        .filter(book -> {                                    // было: if (price == null || price <= 0) continue
            Double price = book.getPrice();
            return price != null && price > 0;
        })
        .forEach(book -> {
            // ... урезанный цикл без фильтров ...
            String author = book.getAuthor();
            AuthorStats stats = result.computeIfAbsent(author, k -> {
                AuthorStats s = new AuthorStats();
                s.firstBookYear = book.getYear();
                return s;
            });
            
            stats.bookCount++;
            stats.totalPages += book.getPages();
            stats.totalPrice += book.getPrice();
            
            if (book.getYear() < stats.firstBookYear) {
                stats.firstBookYear = book.getYear();
            }
            
            book.getGenres().forEach(genre -> stats.genres.add(genre));
        });
    
    result.values().forEach(stats -> 
        stats.averagePrice = stats.totalPrice / stats.bookCount
    );
    
    return result;
}

Улучшения:
Фильтры вынесены, читаются как декларативные условия
Заменили if (stats == null) на computeIfAbsent
Жанры теперь через forEach
Проблема осталась: Сложная логика агрегации всё ещё внутри forEach с мутацией.


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

Шаг 3: Вынести преобразование в map

Выделяем «чистые» данные для агрегации — промежуточный объект:

// Вспомогательный record для промежуточных данных
record BookData(String author, int year, int pages, double price, List<String> genres) {}

public Map<String, AuthorStats> calculateAuthorStatsStep3(List<Book> books, int minYear) {
    Map<String, List<BookData>> grouped = books.stream()
        .filter(book -> book.getYear() >= minYear)
        .filter(book -> !book.getGenres().isEmpty())
        .filter(book -> book.getPrice() != null && book.getPrice() > 0)
        .map(book -> new BookData(
            book.getAuthor(),
            book.getYear(),
            book.getPages(),
            book.getPrice(),
            book.getGenres()
        ))
        .collect(Collectors.groupingBy(BookData::author));
    
    // Отдельный проход для агрегации
    Map<String, AuthorStats> result = new HashMap<>();
    grouped.forEach((author, dataList) -> {
        AuthorStats stats = new AuthorStats();
        stats.bookCount = dataList.size();
        stats.totalPages = dataList.stream().mapToInt(BookData::pages).sum();
        stats.totalPrice = dataList.stream().mapToDouble(BookData::price).sum();
        stats.averagePrice = stats.totalPrice / stats.bookCount;
        stats.firstBookYear = dataList.stream()
            .mapToInt(BookData::year)
            .min()
            .orElse(0);
        stats.genres = dataList.stream()
            .flatMap(d -> d.genres().stream())
            .collect(Collectors.toSet());
        
        result.put(author, stats);
    });
    
    return result;
}

Улучшения:
Разделили фильтрацию, группировку и агрегацию
Каждый этап — отдельный, тестируемый блок
Убрали мутацию внутри потока
Новая проблема: Два прохода по данным, избыточные промежуточные структуры.


Шаг 4: Сложная агрегация через collect

Объединяем всё в один проход с кастомным коллектором:

public Map<String, AuthorStats> calculateAuthorStatsStep4(List<Book> books, int minYear) {
    return books.stream()
        .filter(book -> book.getYear() >= minYear)
        .filter(book -> !book.getGenres().isEmpty())
        .filter(book -> book.getPrice() != null && book.getPrice() > 0)
        .collect(Collectors.groupingBy(
            Book::getAuthor,
            Collector.of(
                AuthorStats::new,           // supplier
                (stats, book) -> {          // accumulator
                    if (stats.bookCount == 0) {
                        stats.firstBookYear = book.getYear();
                    } else {
                        stats.firstBookYear = Math.min(stats.firstBookYear, book.getYear());
                    }
                    stats.bookCount++;
                    stats.totalPages += book.getPages();
                    stats.totalPrice += book.getPrice();
                    stats.genres.addAll(book.getGenres());
                },
                (left, right) -> {          // combiner для parallelStream
                    if (left.bookCount == 0) return right;
                    if (right.bookCount == 0) return left;
                    
                    AuthorStats merged = new AuthorStats();
                    merged.bookCount = left.bookCount + right.bookCount;
                    merged.totalPages = left.totalPages + right.totalPages;
                    merged.totalPrice = left.totalPrice + right.totalPrice;
                    merged.firstBookYear = Math.min(left.firstBookYear, right.firstBookYear);
                    merged.genres.addAll(left.genres);
                    merged.genres.addAll(right.genres);
                    return merged;
                },
                stats -> {                  // finisher
                    stats.averagePrice = stats.totalPrice / stats.bookCount;
                    return stats;
                }
            )
        ));
}

Достигнуто:
Один проход по данным
Нет промежуточных структур
Потенциально parallelizable
Цена: 30 строк сложного кода вместо 40 строк простого. Стало ли читаемее?


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

Шаг 5: Точка остановки — прагматичный компромисс

Вернёмся к шагу 3 и доработаем его до читаемого состояния без избыточной сложности:

public Map<String, AuthorStats> calculateAuthorStatsPragmatic(List<Book> books, int minYear) {
    // Предварительная фильтрация — ясная и тестируемая
    List<Book> validBooks = books.stream()
        .filter(this::isValidForStats)
        .filter(book -> book.getYear() >= minYear)
        .collect(Collectors.toList());
    
    // Группировка — стандартная операция
    Map<String, List<Book>> byAuthor = validBooks.stream()
        .collect(Collectors.groupingBy(Book::getAuthor));
    
    // Агрегация — явный цикл с понятной логикой
    Map<String, AuthorStats> result = new HashMap<>();
    for (Map.Entry<String, List<Book>> entry : byAuthor.entrySet()) {
        result.put(entry.getKey(), aggregateStats(entry.getValue()));
    }
    
    return result;
}

private boolean isValidForStats(Book book) {
    return !book.getGenres().isEmpty() 
        && book.getPrice() != null 
        && book.getPrice() > 0;
}

private AuthorStats aggregateStats(List<Book> authorBooks) {
    AuthorStats stats = new AuthorStats();
    stats.bookCount = authorBooks.size();
    stats.totalPages = authorBooks.stream()
        .mapToInt(Book::getPages)
        .sum();
    stats.totalPrice = authorBooks.stream()
        .mapToDouble(Book::getPrice)
        .sum();
    stats.averagePrice = stats.totalPrice / stats.bookCount;
    stats.firstBookYear = authorBooks.stream()
        .mapToInt(Book::getYear)
        .min()
        .orElse(0);
    stats.genres = authorBooks.stream()
        .flatMap(b -> b.getGenres().stream())
        .collect(Collectors.toSet());
    return stats;
}

Почему это лучше шага 4:
Методы isValidForStats и aggregateStats можно тестировать отдельно
Нет 30-строчного лямбда-ада в коллекторе
Junior-разработчик поймёт код за 5 минут
Легко добавить логирование или метрики в промежуточные точки


Критерии остановки рефакторинга

Остановитесь, если:
Код стал медленнее без причины
Появились неочевидные побочные эффекты

// Опасно: параллельный stream с непотокобезопасным accumulators
.collect(Collectors.groupingByConcurrent(
    Book::getAuthor,
    Collector.of(
        () -> new AuthorStats(),  // ОК — новый для каждого
        (stats, book) -> stats.genres.addAll(...),  // Опасно! HashSet не thread-safe
        ...
    )
))

Теряется отладочная информация
В исходном коде можно было поставить breakpoint на любой continue. В сложном коллекторе — только внутри лямбд.

Команда не владеет паттернами
Если 3 из 5 разработчиков не знают Collector.of — код неподдерживаем.


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

Задача 1: рефакторинг с остановкой

В проекте «Библиотека» найдите метод с циклом, содержащим:
2+ условия continue
мутацию аккумулятора
вложенные циклы
Проведите рефакторинг до шага 3 (группировка + явная агрегация). Остановитесь. Обоснуйте выбор.

Задача 2: сравнение читаемости

Покажите шаг 4 (полный коллектор) коллеге, не знакомому с Stream API. Засеките время, за которое он поймёт логику. Повторите с шагом 5. Зафиксируйте разницу.

Задача 3: добавление функциональности

К обоим вариантам (шаг 4 и шаг 5) добавьте требование: «пропускать авторов с менее чем 3 книгами».
В каком варианте изменение проще? В каком меньше риск регрессии?

Задача 4: документация компромисса (звёздочка)
Создайте в проекте docs/stream-guidelines.md с правилами:
когда использовать Stream
когда остановиться
примеры «хорошего», «плохого» и «достаточного» кода из вашей кодовой базы


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

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

Глава 8: За пределами коллекций. Бесконечность и I/O

Бесконечные стримы и ленивая генерация

Ранее Stream API рассматривался как инструмент обработки конечных коллекций данных. Но фундаментальная мощь потоков — в их способности моделировать бесконечные последовательности, вычисляемые по требованию. Это сдвигает фокус с "данных в памяти" к "процессу генерации", открывая паттерны для числовых последовательностей, генераторов уникальных идентификаторов, обработки потоковых источников вроде сетевых соединений.


Stream.iterate: рекурсия с состоянием

Метод Stream.iterate создаёт поток, где каждый следующий элемент вычисляется из предыдущего.

Перегрузка Java 8 принимает начальное значение и унарную функцию:

// Натуральные числа: 0, 1, 2, 3...
Stream<Integer> natural = Stream.iterate(0, n -> n + 1);

// Степени двойки: 1, 2, 4, 8...
Stream<Integer> powersOfTwo = Stream.iterate(1, n -> n * 2);

// Фибоначчи: пара (a, b) -> (b, a+b)
Stream<long[]> fibonacci = Stream.iterate(
    new long[]{0, 1}, 
    f -> new long[]{f[1], f[0] + f[1]}
);

Ключевая особенность: iterate сохраняет состояние между элементами. Это делает его непараллелизуемым — каждый элемент зависит от предыдущего, разделение невозможно. Попытка вызвать .parallel() на таком потоке не даст выигрыша, а может замедлить из-за накладных расходов.


Java 9+: iterate с предикатом остановки

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

// Числа от 0 до 99
Stream<Integer> limited = Stream.iterate(
    0,           // seed
    n -> n < 100, // hasNext (предикат продолжения)
    n -> n + 1    // next (функция следующего)
);

Это заменяет паттерн iterate(...).limit(n) более семантически ясным конструктом. Предикат проверяется перед генерацией каждого элемента, включая seed. Если hasNext(seed) ложно, поток пуст.


Stream.generate: чистая генерация без состояния

Stream.generate принимает Supplier — функцию без аргументов, возвращающую значение. Каждый вызов независим, что теоретически позволяет параллелизм (хотя реализация в OpenJDK не делит такие потоки эффективно).

// Постоянное значение
Stream<String> constants = Stream.generate(() -> "repeat");

// Случайные числа
Stream<Double> randoms = Stream.generate(Math::random);

// Уникальные ID через атомарный счётчик
AtomicLong idGenerator = new AtomicLong(0);
Stream<String> uniqueIds = Stream.generate(() -> "ID-" + idGenerator.incrementAndGet());

generate идеален для стохастических или внешне определяемых последовательностей, где нет рекуррентной зависимости. Но бесконечность требует ограничения: limit, takeWhile или короткое замыкание терминальной операцией.


takeWhile и dropWhile: предикатные границы

Java 9 добавила методы takeWhile и dropWhile, критически отличные от filter по семантике:
filter проверяет каждый элемент независимо, пропуская или отбрасывая по условию
takeWhile пропускает элементы, пока предикат истинен, и останавливает поток при первом ложном
dropWhile отбрасывает элементы, пока предикат истинен, и продолжает с первого ложного

// Сортированный поток температур
IntStream temperatures = IntStream.of(15, 18, 22, 25, 19, 12, 8);

// takeWhile: температура ниже 25
temperatures.takeWhile(t -> t < 25)
    .forEach(System.out::println);  // 15, 18, 22 — остановка на 25

// dropWhile: пропускаем прохладную погоду
temperatures.dropWhile(t -> t < 20)
    .forEach(System.out::println);  // 22, 25, 19, 12, 8

Ключевое отличие takeWhile от filter — short-circuit поведение. filter(t -> t < 25) обработал бы все элементы, включая 19, 12, 8. takeWhile останавливается, предполагая, что последовательность упорядочена и дальнейшие элементы не интересны.

Это критично для бесконечных потоков:

// Бесконечная последовательность, ограниченная условием
Stream.iterate(1, n -> n * 2)
    .takeWhile(n -> n < 1000)  // 1, 2, 4, 8, ..., 512 — остановка
    .forEach(System.out::println);

Без takeWhile пришлось бы использовать limit, требующий знания количества элементов заранее, или filter, не останавливающийся.


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

Сценарий: генерация уникальных идентификаторов

Комбинация generate с takeWhile или limit создаёт контролируемые генераторы:

// UUID с проверкой уникальности в пределах сессии
Set<String> usedIds = ConcurrentHashMap.newKeySet();
Stream<String> uniqueIds = Stream.generate(UUID::randomUUID)
    .map(UUID::toString)
    .filter(usedIds::add)  // add возвращает false если уже есть, фильтруем дубликаты
    .limit(1000);          // строго 1000 уникальных

// Или с takeWhile для условной остановки
Stream<String> idsUntilPattern = Stream.generate(this::generateId)
    .takeWhile(id -> !id.contains("STOP"));  // Генерация до спецпаттерна

Атомарные структуры (AtomicLong, ConcurrentHashMap) обеспечивают потокобезопасность при параллельном доступе, хотя generate в стандартной реализации не эффективно параллелится.


Сценарий: чтение потоковых данных

Бесконечные стримы моделируют внешние источники с неизвестным объёмом:
```
// Симуляция чтения из сокета: байты до терминатора
Stream<Byte> socketStream = Stream.generate(() -> readFromSocket())
.takeWhile(b -> b != -1); // -1 как EOF

// Обработка пакетов до специального маркера
Stream<Packet> packetStream = Stream.generate(this::readNextPacket)
.takeWhile(p -> !p.isTerminator());

Важно: такие потоки требуют управления ресурсами. Stream.generate не знает о необходимости закрыть сокет. Использование в try-with-resources или явное закрытие источника в Supplier — ответственность разработчика.


Практика: числовые последовательности с ленивостью

Сравним подходы к генерации последовательностей:

// Плохо: немедленное создание списка
List<Integer> eager = new ArrayList<>();
for (int i = 0; i < 1000000; i++) {
    if (isPrime(i)) eager.add(i);
}

// Хорошо: ленивый поток, элементы вычисляются по требованию
IntStream primes = IntStream.iterate(2, n -> n + 1)
    .filter(this::isPrime)
    .takeWhile(n -> n < 1000000);

// Потребление только первых 10, остальные никогда не вычислены
primes.limit(10).forEach(System.out::println);

Ленивость позволяет работать с "потенциально бесконечными" последовательностями, фактически обрабатывая только необходимый минимум.


Ограничения бесконечных потоков

Параллелизм: iterate не параллелится, generate — плохо. Бесконечные потоки — последовательная абстракция.
Состояние: iterate требует небольшого состояния (предыдущий элемент), но не масштабируется. Сложное состояние в Supplier требует синхронизации.
Ресурсы: бесконечность — концептуальная. Реальные источники (сокеты, файлы) требуют закрытия. Stream API не управляет жизненным циклом внешних ресурсов автоматически.


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

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

Глава 8: За пределами коллекций. Бесконечность и I/O

I/O операции как потоки данных

Stream API расширяет свою абстракцию от коллекций в памяти к внешним источникам данных. Файлы, сетевые соединения, процессы — всё это может быть представлено как Stream<String> или Stream<Byte>, с ленивой подкачкой данных по мере необходимости. Этот подход критичен для обработки данных, не помещающихся в RAM: гигабайтные логи, бесконечные потоки событий, построчная обработка без полной загрузки.


Files.lines: файловый поток строк

Метод Files.lines(Path path) — фабрика потоков для текстовых файлов. Он возвращает Stream<String>, где каждый элемент — строка файла, декодированная в указанной или платформенной кодировке (по умолчанию UTF-8).

// Базовое использование с автоматическим закрытием
try (Stream<String> lines = Files.lines(Path.of("access.log"))) {
    long errorCount = lines
        .filter(line -> line.contains("ERROR"))
        .count();
}

Критически важно: Files.lines возвращает поток, реализующий AutoCloseable. Файловый дескриптор открывается при создании потока и должен быть явно закрыт. Без try-with-resources или явного close() ресурс утекает до сборки мусора, что при высокой частоте операций приводит к исчерпанию дескрипторов ОС (Too many open files).

Паттерн обработки больших файлов:

// Фильтрация, трансформация, запись результата — всё в потоковом режиме
try (Stream<String> lines = Files.lines(Path.of("huge_input.txt"));
     BufferedWriter writer = Files.newBufferedWriter(Path.of("filtered_output.txt"))) {
    
    lines.parallel()  // Осторожно: см. предупреждение ниже
        .filter(line -> line.length() > 100)
        .map(String::toUpperCase)
        .map(line -> line + System.lineSeparator())
        .forEachOrdered(line -> {
            try {
                writer.write(line);
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        });
}

Здесь ни одна строка не хранится в памяти целиком. Files.lines использует BufferedReader с буфером 8192 байт, читая файл блоками и выдавая строки по границам \n или \r\n. Обработка происходит пакетно, с постоянным потреблением памяти независимо от размера файла.


BufferedReader.lines: низкоуровневый контроль

Для специализированных сценариев — настройка размера буфера, обработка кодировок, работа с существующим Reader:

// Кастомная кодировка и буфер
try (BufferedReader reader = new BufferedReader(
        new InputStreamReader(
            new FileInputStream("legacy.txt"), 
            StandardCharsets.ISO_8859_1
        ), 
        16384  // Удвоенный буфер для последовательного чтения
    );
    Stream<String> lines = reader.lines()) {
    
    lines.map(this::parseLegacyRecord)
         .filter(Objects::nonNull)
         .forEach(this::processRecord);
}

BufferedReader.lines() возвращает поток с теми же семантиками ленивости, но без привязки к Path — работает с любым Reader, включая StringReader, CharArrayReader, сетевые потоки через InputStreamReader.


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

Управление ресурсами и исключения

I/O потоки добавляют сложность обработки checked исключений. Лямбды в Stream API не объявляют throws, требуя обёртки:

// Проблема: readLine бросает IOException
Stream<String> lines = Files.lines(path);
lines.map(line -> {
    // Ошибка компиляции: unreported exception IOException
    return expensiveParser.parse(line);  
});

Решения:
Обёртка в runtime exception:

.lines.map(line -> {
    try {
        return expensiveParser.parse(line);
    } catch (IOException e) {
        throw new UncheckedIOException(e);
    }
})

Извлечение в метод с обёрткой:

private ParsedRecord safeParse(String line) {
    try {
        return expensiveParser.parse(line);
    } catch (IOException e) {
        throw new UncheckedIOException(e);
    }
}

// В потоке
.lines.map(this::safeParse)

Специализированный коллектор для ошибок (см. Урок 3.3): разделение успешных результатов и ошибок без прерывания потока.


Параллелизм и I/O: катастрофическая комбинация

Предупреждение из предыдущих глав приобретает критическую важность для I/O. parallelStream() на Files.lines или BufferedReader.lines() — антипаттерн с тяжёлыми последствиями:

// КАТАСТРОФА: параллельное чтение файла через common pool
try (Stream<String> lines = Files.lines(Path.of("access.log"))) {
    lines.parallel()  // Нет выигрыша, есть риск
        .map(this::blockingDatabaseLookup)  // Блокировка воркера
        .collect(toList());
}

Проблемы:
Нет разделения: BufferedReader не поддерживает trySplit(). lines().parallel() не делит файл на сегменты — он создаёт иллюзию параллелизма, фактически синхронизируя доступ к общему Reader.

Блокировка common pool: если map содержит блокирующие операции (запросы к БД, HTTP вызовы, Thread.sleep), все воркеры ForkJoinPool.commonPool() замораживаются. Другие компоненты приложения (CompletableFuture, другие parallelStream) парализуются.

Нарушение порядка: forEach в параллельном потоке не сохраняет порядок строк файла. forEachOrdered требует синхронизации, сводя на нет выигрыш.

Исключение: если обработка строк CPU-bound и дорога (сложный парсинг, криптографические операции), а чтение — отдельная стадия, можно разделить:

// Чтение последовательное, обработка параллельная — но с осторожностью
List<String> batch = new ArrayList<>(BATCH_SIZE);
try (Stream<String> lines = Files.lines(Path.of("huge.txt"))) {
    lines.forEach(line -> {
        batch.add(line);
        if (batch.size() >= BATCH_SIZE) {
            processBatchParallel(new ArrayList<>(batch));  // Копия для безопасности
            batch.clear();
        }
    });
    if (!batch.isEmpty()) processBatchParallel(batch);
}

private void processBatchParallel(List<String> batch) {
    batch.parallelStream()
        .map(this::expensiveCpuBoundTransform)
        .collect(toList());  // Результат куда-то сохраняется
}

Но лучший подход для I/O-bound задач — CompletableFuture с кастомным пулом или virtual threads (Java 21+):

// Правильно: изолированный пул для блокирующих операций
ExecutorService ioPool = Executors.newFixedThreadPool(50);

try (Stream<String> lines = Files.lines(Path.of("urls.txt"))) {
    List<CompletableFuture<Response>> futures = lines
        .map(url -> CompletableFuture.supplyAsync(
            () -> fetchHttp(url), 
            ioPool
        ))
        .collect(toList());
    
    List<Response> results = futures.stream()
        .map(CompletableFuture::join)
        .collect(toList());
} finally {
    ioPool.shutdown();
}

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

Паттерн: конвейер ETL без промежуточных файлов

Классический паттерн Extract-Transform-Load реализуется через композицию потоков:

// Извлечение из CSV
try (Stream<String> lines = Files.lines(Path.of("input.csv"));
     // Загрузка в выходной файл
     BufferedWriter writer = Files.newBufferedWriter(Path.of("output.json"))) {
    
    lines.skip(1)  // Пропуск заголовка
        .map(this::parseCsvLine)           // String -> Record
        .filter(Objects::nonNull)            // Удаление malformed
        .map(this::transformToJson)          // Record -> JSON string
        .forEach(json -> {
            try {
                writer.write(json);
                writer.newLine();
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        });
}

Память потребляется постоянно (размер буфера чтения + одна строка обработки), независимо от размера входного файла. Скорость ограничена I/O диска или сети, а не CPU.


Закрытие и обработка ошибок

При исключении в промежуточной операции поток прерывается, но ресурс в try-with-resources закрывается корректно:

try (Stream<String> lines = Files.lines(Path.of("corrupt.txt"))) {
    lines.map(this::parse)
        .filter(Objects::nonNull)
        .forEach(this::process);
    // Если parse бросает RuntimeException на 1000-й строке,
    // lines.close() вызывается автоматически
}

Но если исключение происходит в терминальной операции, а промежуточные содержат ресурсы (например, map открывает соединения), требуется явное управление:

// Антипаттерн: ресурс внутри map
lines.map(line -> {
    Connection conn = pool.borrow();  // Открытие здесь
    return query(conn, line);         // Если исключение, conn не возвращается
})

// Правильно: try-with-resources внутри лямбды, или вне потока

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