Контролируемые побочные эффекты

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

Критерии контролируемого побочного эффекта:
Идемпотентность: повторный вызов с теми же входными данными даёт тот же результат (или эквивалентный с точки зрения системы).
Потокобезопасность: при параллельном выполнении эффект корректен без дополнительной синхронизации.
Изолированность: эффект не влияет на другие элементы потока или внешнее состояние, используемое в потоке.
Необходимость: эффект является целью операции, не побочным продуктом.

Пример: атомарный счётчик для отладки

AtomicInteger processedCount = new AtomicInteger(0);

List<Result> results = items.parallelStream()
    .map(this::heavyProcessing)
    .peek(r -> processedCount.incrementAndGet())  // Контролируемый эффект
    .collect(toList());

System.out.println("Обработано: " + processedCount.get());

AtomicInteger обеспечивает потокобезопасность инкремента. Эффект изолирован — счётчик не влияет на обработку элементов. Он идемпотентен с точки зрения бизнес-логики (конечный результат results не зависит от счётчика). Но это всё ещё антипаттерн: peek предназначен для отладки, а не для бизнес-логики, и его поведение может измениться в будущих версиях JDK.

Более чистый подход — отделение агрегации от обработки:

// Лучше: явная агрегация результата
class ProcessingResult {
    final List<Result> results;
    final int count;
    ProcessingResult(List<Result> results, int count) {
        this.results = results;
        this.count = count;
    }
}

ProcessingResult finalResult = items.parallelStream()
    .collect(() -> new ProcessingResult(new ArrayList<>(), 0),
        (acc, item) -> {
            acc.results.add(process(item));
            acc.count++;
        },
        (left, right) -> {
            left.results.addAll(right.results);
            left.count += right.count;
        });

Или использование teeing для подсчёта параллельно с сбором результатов.

Пример: логирование

orders.stream()
    .filter(o -> o.amount().compareTo(THRESHOLD) > 0)
    .peek(o -> auditLog.record("Крупный заказ", o.id()))  // Побочный эффект
    .map(this::processLargeOrder)
    .collect(toList());

Логирование — необходимый побочный эффект. Он потокобезопасен, если auditLog потокобезопасен. Он изолирован — не влияет на обработку заказа. Но размещение в peek рискованно: при оптимизации конвейера peek может быть пропущен или вызван менее раз, чем ожидается.

Надёжнее:

List<ProcessedOrder> processed = orders.stream()
    .filter(o -> o.amount().compareTo(THRESHOLD) > 0)
    .map(o -> {
        auditLog.record("Крупный заказ", o.id());  // Явный эффект в map
        return processLargeOrder(o);
    })
    .collect(toList());

Теперь логирование гарантировано для каждого прошедшего фильтр элемента, хотя map с побочным эффектом всё ещё нарушает чистоту функции.


Граница допустимого

Правило практики: побочные эффекты допустимы только в терминальных операциях, работающих с уже потокобезопасными внешними системами. forEach для отправки сообщений в Kafka, collect в ConcurrentHashMap, reduce с атомарными операциями — приемлемы. Побочные эффекты в промежуточных операциях (map, filter, flatMap) — почти всегда ошибка.
Исключение — кэширование вычислений внутри операции, но и здесь предпочтительны чистые функции с мемоизацией вне потока.


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

Проблема 3: groupingBy ломается

public void demonstrateGroupingProblem() {
    List<Book> books = Arrays.asList(
        new Book("Война и мир", "Толстой"),
        new Book("Война и мир", "Достоевский"),
        new Book("Война и мир", "Тургенев")
    );
    
    Map<Book, List<Book>> byBook = books.stream()
        .collect(Collectors.groupingBy(Function.identity()));
    
    System.out.println("Групп: " + byBook.size());
    byBook.forEach((k, v) -> System.out.println(k + " -> " + v.size()));
}

Ожидаемо: 3 группы (все разные авторы) или 1 группа (если equals/hashCode считают их одинаковыми).
Фактически: Непредсказуемо из-за отсутствующего hashCode.


Исправление: корректные equals и hashCode

Правильная реализация по всем полям

Обновите класс Book:

@Override
public boolean equals(Object o) {
    if (this == o) return true;
    if (!(o instanceof Book)) return false;
    Book book = (Book) o;
    return title.equals(book.title) && 
           author.equals(book.author);  // Теперь оба поля!
}

@Override
public int hashCode() {
    return Objects.hash(title, author);  // Консистентно с equals
}

Контракт equals и hashCode:
Рефлексивность: x.equals(x) всегда true
Симметричность: x.equals(y) ⇔ y.equals(x)
Транзитивность: x.equals(y) и y.equals(z) → x.equals(z)
Консистентность: повторные вызовы дают тот же результат
С null: x.equals(null) всегда false
Критически: Если x.equals(y), то x.hashCode() == y.hashCode()

Проверка исправления

Повторите все демонстрации

public void demonstrateFixed() {
    List<Book> books = Arrays.asList(
        new Book("Война и мир", "Толстой"),
        new Book("Война и мир", "Достоевский"),
        new Book("1984", "Оруэлл")
    );
    
    // distinct() — теперь 3 уникальные книги
    System.out.println("distinct: " + books.stream().distinct().count());
    
    // toMap — теперь 3 записи
    Map<Book, Integer> stock = books.stream()
        .collect(Collectors.toMap(
            Function.identity(),
            b -> 10,
            Integer::sum
        ));
    System.out.println("Map size: " + stock.size());
    
    // Поиск работает
    Book lookup = new Book("Война и мир", "Толстой");
    System.out.println("Найдено: " + stock.get(lookup));
}

Ожидаемый результат: 3 уникальных книги, Map с 3 записями, поиск находит значение.


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

Альтернативные стратегии equals/hashCode

Стратегия 1: Только неизменяемый идентификатор (ISBN)

Если Book имеет уникальный ISBN:

private final String isbn;  // уникальный, неизменяемый

@Override
public boolean equals(Object o) {
    if (this == o) return true;
    if (!(o instanceof Book)) return false;
    return isbn.equals(((Book) o).isbn);
}

@Override
public int hashCode() {
    return isbn.hashCode();
}

Плюсы: Эффективно, надёжно. Минусы: Требует уникального поля.

Стратегия 2: Все поля

Плюсы: Корректно для value objects. Минусы: Медленнее, чувствительно к изменениям.

Стратегия 3: Не переопределять (только identity)

Если Book — entity с уникальным ID в базе данных, и вы никогда не сравниваете разные объекты с одинаковым ID:

// Нет equals/hashCode — используем Object

Плюсы: Просто. Минусы: distinct(), toMap, groupingBy работают по ссылке, не по содержимому.

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

Задача 1: воспроизвести все три проблемы

В проекте «Библиотека» создайте класс BrokenBook с equals только по title и без hashCode.

Продемонстрируйте:
distinct() «проглатывает» книги или нет (зависит от хэш-коллизий)
toMap создаёт больше записей, чем ожидается
groupingBy разбивает на неожиданное количество групп
Зафиксируйте наблюдения в комментариях.

Задача 2: исправить и сравнить

Создайте FixedBook с корректной парой equals/hashCode. Повторите те же операции.

Убедитесь, что:
Толстой, Достоевский, Тургенев — три разные книги с одинаковым названием
distinct() оставляет все три
toMap создаёт три записи
Поиск по ключу находит правильное значение

Задача 3: специфический equals для бизнес-логики

Добавьте в Library метод findByTitle(String title), который находит все книги с данным названием (независимо от автора).
Реализуйте через filter с кастомным предикатом, не через equals:

public List<Book> findByTitle(String title) {
    return books.stream()
        .filter(b -> b.getTitle().equals(title))
        .collect(Collectors.toList());
}

Объясните, почему это безопаснее, чем менять equals для всего класса.

Задача 4: кеширование hashCode (звёздочка)

Если поля title и author неизменяемы, закешируйте hashCode:

private int hashCode;  // 0 = не вычислено

@Override
public int hashCode() {
    int result = hashCode;
    if (result == 0) {
        result = Objects.hash(title, author);
        hashCode = result;
    }
    return result;
}

Сравните производительность при частом использовании в HashMap для миллиона операций.

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

Раздел 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