Реактивное программирование

Базовые операторы в Reactor: map, filter, flatMap

Операторы — это методы на Mono/Flux, которые позволяют строить конвейеры: преобразовывать, фильтровать и комбинировать данные асинхронно. Представьте их как звенья в цепи: каждый берёт входной поток, меняет его и передаёт дальше. Сегодня разберём три фундаментальных: map (преобразование элементов), filter (фильтрация) и flatMap (плоское преобразование, для слияния подпотоков). Эти операторы — основа для сложных сценариев, они решают проблемы из первого поста, позволяя писать декларативный код вместо ручных циклов и ожиданий.

Операторы в Reactor — декларативные: вы описываете, что делать с данными, а библиотека заботится об асинхронности, backpressure и ошибках. Они не меняют исходный поток (иммутабельны), а создают новый. Это делает код читаемым и тестируемым.


Map: простое преобразование элементов

Map — оператор для изменения каждого элемента потока. Он берёт входной элемент, применяет функцию и выдаёт результат. Синхронный: функция должна быть быстрой и без блокировок. Идеален для конвертации типов, вычислений или форматирования.

Пример на Flux:

import reactor.core.publisher.Flux;
Flux<String> originalFlux = Flux.just("яблоко", "банан", "вишня");
Flux<String> transformed = originalFlux.map(fruit -> fruit.toUpperCase()); // Преобразование в верхний регистр
transformed.subscribe(System.out::println); // Вывод: "ЯБЛОКО", "БАНАН", "ВИШНЯ"

Здесь map применяет лямбду к каждому элементу последовательно. Если ошибка в функции — сработает onError.

На Mono:

Mono<Integer> num = Mono.just(5).map(x -> x * 2); // Результат: 10

Почему map полезен? В традиционных подходах (как в CompletableFuture.thenApply) вы строите цепочки, но рискуете вложенностью. В Reactor map делает конвейер линейным: читается как последовательный код, но работает асинхронно. Поддерживает backpressure: если подписчик запрашивает n, map передаёт запрос upstream (источнику).


Filter: отбор элементов по условию

Filter — для пропуска только нужных элементов. Принимает предикат (функцию, возвращающую true/false) и пропускает те, для которых true. Остальные игнорируются — поток "сужается".

Пример на Flux:

Flux<Integer> numbers = Flux.range(1, 10);
Flux<Integer> evenNumbers = numbers.filter(num -> num % 2 == 0); // Только чётные
evenNumbers.subscribe(System.out::println); // Вывод: 2, 4, 6, 8, 10

Если поток пустой или ничего не проходит — onComplete сработает без onNext.

На Mono:

Mono<String> word = Mono.just("привет").filter(w -> w.length() > 7); // Не пройдёт — пустой Mono

Filter экономит ресурсы: ненужные элементы не обрабатываются дальше в цепи. В отличие от императивных циклов (где вы фильтруете в for с if), здесь всё асинхронно и с backpressure — запросы передаются источнику только для прошедших элементов.

Комбинация с map: numbers.filter(num -> num > 5).map(num -> num * 10).subscribe(); // 60, 70, 80, 90, 100
Это строит конвейер: фильтр → преобразование, без ручных переменных.


#Java #middle #Reactor #map #filter #flatMap

FlatMap: плоское преобразование для асинхронных подпотоков

FlatMap — мощный оператор для случаев, когда из одного элемента нужно создать подпоток (Publisher), и слить их в плоский результат. Это как map, но для асинхронных или множественных выходов: он "разворачивает" вложенные потоки. Полезен для запросов в цикле: например, для каждого пользователя — асинхронно запросить данные.

Пример на Flux:

Flux<String> fruits = Flux.just("яблоко", "банан");
Flux<Character> letters = fruits.flatMap(fruit -> Flux.fromArray(fruit.toCharArray())); // Из строки — поток символов
letters.subscribe(System.out::println); // Вывод: я, б, л, о, к, о, б, а, н, а, н (в возможном перемешанном порядке, если асинхронно)

Здесь flatMap берёт строку, создаёт Flux из символов и сливает всё в один поток. В отличие от map (который вернул бы Flux<Flux<Character>> — вложенный), flatMap "сплющивает".

Асинхронный пример: симулируем API-запросы.

import java.time.Duration;
Flux<String> users = Flux.just("user1", "user2");
Flux<String> data = users.flatMap(user -> Mono.just("Данные для " + user).delayElement(Duration.ofSeconds(1))); // Асинхронный подпоток с задержкой
data.subscribe(System.out::println); // Вывод через секунды: "Данные для user1", "Данные для user2" (параллельно, если scheduler позволяет)

FlatMap уважает backpressure: запрашивает у подпотоков по мере нужды. Но осторожно: если подпотоки бесконечные — рискуете перегрузкой. Параметр concurrency (flatMap(func, concurrency)) ограничивает параллелизм.

Почему flatMap решает проблемы? В традиционных подходах (циклы с Future) вы ждёте каждый запрос, блокируя. Здесь — асинхронное слияние, без ожиданий и callback-ада: цепочка читаема.


Практические советы и подводные камни

Читаемость: цепочки операторов пишите по строкам для ясности: flux.filter(...).map(...).flatMap(...);
Ошибки: если в map/flatMap исключение — onError. Используйте handle() для условной обработки.
Производительность: в flatMap устанавливайте concurrency (default 256) для контроля параллелизма: flatMap(func, 4) — max 4 подпотока одновременно.
Камень: блокирующий код в лямбдах — сломает асинхронность. Для IO — используйте flatMap с Mono.fromCallable и publishOn(Schedulers.boundedElastic()).
Тестирование: StepVerifier.create(flux.map(...)).expectNext("ЯБЛОКО").verifyComplete();


#Java #middle #Reactor #map #filter #flatMap

map: структурная трансформация

Операция map принимает Function<T, R> и применяет её к каждому элементу, заменяя его результатом функции. Это преобразование типа: из потока книг делаем поток названий, из потока строк — поток их длин, из потока заказов — поток сумм.

Stream<String> titles = library.stream()
    .map(Book::title);

Stream<Integer> lengths = titles.map(String::length);

map также stateless и ленив. Функция вызывается только при запросе элемента терминальной операцией. Стоимость — O(1) на элемент плюс стоимость функции. Функция должна быть чистой: один и тот же вход всегда даёт один и тот же выход, без модификации внешнего состояния.

Композиция map позволяет выстраивать цепочки преобразований, читаемые справа налево или через method reference:

// Цепочка преобразований: Book -> String -> Integer -> String
String summary = book.stream()
    .map(Book::author)           // Book -> String
    .map(String::toUpperCase)    // String -> String
    .map(author -> author.substring(0, 3))  // String -> String
    .map(prefix -> "Author: " + prefix)     // String -> String
    .findFirst()
    .orElse("Unknown");

Каждый map — отдельный узел конвейера, отдельный объект в памяти. Для простых преобразований это приемлемо, но для тяжёлых операций композиция нескольких map может быть менее эффективна, чем одно комплексное преобразование, из-за накладных расходов на вызовы лямбд и промежуточные объекты.


flatMap: разворачивание иерархий

flatMap — наиболее мощный и наиболее сложный из трёх примитивов. Он решает проблему, которую map решить не может: когда один элемент источника порождает ноль, один или множество элементов результата.

Сигнатура flatMap требует функцию, возвращающую Stream<R> вместо R. Операция применяет эту функцию к каждому элементу, получает потоки результатов и сливает их в единый выходной поток. Это аналог вложенных циклов или операции "расплющивания" (flattening) в функциональных языках.

Классический сценарий: коллекция внутри коллекции

Предположим, у нас есть список заказов, каждый содержит список позиций.

Нужно получить поток всех позиций всех заказов:

List<Order> orders = fetchOrders();

// Императивный подход: вложенные циклы, временная коллекция
List<OrderItem> allItems = new ArrayList<>();
for (Order order : orders) {
    for (OrderItem item : order.getItems()) {
        allItems.add(item);
    }
}

// Декларативный подход: flatMap
Stream<OrderItem> itemStream = orders.stream()
    .flatMap(order -> order.getItems().stream());

Здесь order -> order.getItems().stream() — функция, превращающая один Order в Stream<OrderItem>. flatMap выполняет слияние этих потоков, устраняя уровень вложенности. Результат — плоский поток всех элементов.

Нулевое и единичное отображение

flatMap универсален: он обрабатывает все случаи количества выходных элементов на входной:
Ноль элементов: возвращаем Stream.empty(), элемент исключается из результата
Один элемент: возвращаем Stream.of(element), эквивалентно map с последующим filter(Objects::nonNull)
Множество элементов: возвращаем collection.stream(), все элементы включаются в результат

// Фильтрация через flatMap: ноль или один элемент
Stream<Book> validBooks = rawData.stream()
    .flatMap(data -> {
        try {
            Book book = parseBook(data);  // Может бросить исключение
            return Stream.of(book);         // Один элемент при успехе
        } catch (ParseException e) {
            return Stream.empty();          // Ноль элементов при ошибке
        }
    });

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

Сложность и читаемость

Мощь flatMap оборачивается риском чрезмерной сложности.

Вложенные flatMap создают "лестницу" преобразований, сложную для восприятия:

// Антипаттерн: глубокая вложенность flatMap
Stream<String> deepStream = orders.stream()
    .flatMap(order -> order.getShipments().stream()
        .flatMap(shipment -> shipment.getPackages().stream()
            .flatMap(pkg -> pkg.getItems().stream()
                .flatMap(item -> item.getTags().stream()))));

Этот код эквивалентен четырём вложенным циклам. Он работает, но его трудно читать и отлаживать. Каждый уровень вложенности добавляет лексический контекст, и переменные верхних уровней (order, shipment) доступны во вложенных лямбдах, создавая неявные зависимости.

Рефакторинг через промежуточные методы улучшает читаемость:

// Лучше: именованные методы с ясной семантикой
Stream<String> tags = orders.stream()
    .flatMap(this::extractShipments)
    .flatMap(this::extractPackages)
    .flatMap(this::extractItems)
    .flatMap(this::extractTags);

private Stream<Shipment> extractShipments(Order order) {
    return order.getShipments().stream();
}
// ... и так далее

Или использование mapMulti (Java 16+) — операции, комбинирующей map и flatMap с более эффективной реализацией для множественного вывода:

// mapMulti: более эффективен, чем flatMap, для множественных элементов
Stream<String> tags = orders.stream()
    .<String>mapMulti((order, consumer) -> {
        for (Shipment s : order.getShipments()) {
            for (Package p : s.getPackages()) {
                for (Item i : p.getItems()) {
                    for (String tag : i.getTags()) {
                        consumer.accept(tag);  // Прямая передача без создания Stream
                    }
                }
            }
        }
    });

mapMulti избегает создания промежуточных объектов Stream для каждого элемента, что снижает нагрузку на GC при глубокой вложенности.


Анализ стоимости

filter: O(1) на элемент. Память: O(1). Параллелизм: отличный, нет синхронизации.
map: O(1) на элемент плюс стоимость функции. Память: O(1) для примитивов, O(1) для ссылок (создаётся новый объект результата). Параллелизм: отличный.
flatMap: O(1) на элемент плюс стоимость создания выходного потока плюс стоимость его обхода. Память: зависит от размера выходных потоков. Параллелизм: сложный — слияние множественных потоков требует координации.

Скрытая стоимость flatMap — в аллокациях. Каждый вызов функции создаёт новый объект Stream (обычно через Collection.stream() или Stream.of()). Для больших коллекций это давит на Young Generation. Оптимизация через mapMulti или специализированные сплитераторы уместна в критичных секциях.


Композиция как язык запросов

Комбинация filter, map, flatMap образует внутренний DSL (domain-specific language) для преобразования данных.

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

// "Найти все уникальные теги из активных заказов, отсортированные по алфавиту"
List<String> sortedUniqueTags = orders.stream()
    .filter(Order::isActive)           // которые активны
    .flatMap(o -> o.getTags().stream()) // извлечь все теги
    .distinct()                        // уникальные
    .sorted()                          // отсортированные
    .collect(toList());                // собрать в список

Каждая операция — глагол, каждый метод reference или лямбда — дополнение. Порядок операций отражает логику обработки: сначала сужаем (filter), затем расплющиваем (flatMap), затем нормализуем (distinct, sorted).

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