Реактивное программирование
Базовые операторы в Reactor: map, filter, flatMap
Операторы — это методы на Mono/Flux, которые позволяют строить конвейеры: преобразовывать, фильтровать и комбинировать данные асинхронно. Представьте их как звенья в цепи: каждый берёт входной поток, меняет его и передаёт дальше. Сегодня разберём три фундаментальных: map (преобразование элементов), filter (фильтрация) и flatMap (плоское преобразование, для слияния подпотоков). Эти операторы — основа для сложных сценариев, они решают проблемы из первого поста, позволяя писать декларативный код вместо ручных циклов и ожиданий.
Операторы в Reactor — декларативные: вы описываете, что делать с данными, а библиотека заботится об асинхронности, backpressure и ошибках. Они не меняют исходный поток (иммутабельны), а создают новый. Это делает код читаемым и тестируемым.
Map: простое преобразование элементов
Map — оператор для изменения каждого элемента потока. Он берёт входной элемент, применяет функцию и выдаёт результат. Синхронный: функция должна быть быстрой и без блокировок. Идеален для конвертации типов, вычислений или форматирования.
Пример на Flux:
На Mono:
Почему map полезен? В традиционных подходах (как в CompletableFuture.thenApply) вы строите цепочки, но рискуете вложенностью. В Reactor map делает конвейер линейным: читается как последовательный код, но работает асинхронно. Поддерживает backpressure: если подписчик запрашивает n, map передаёт запрос upstream (источнику).
Filter: отбор элементов по условию
Filter — для пропуска только нужных элементов. Принимает предикат (функцию, возвращающую true/false) и пропускает те, для которых true. Остальные игнорируются — поток "сужается".
Пример на Flux:
На 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
Базовые операторы в 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); // Не пройдёт — пустой MonoFilter экономит ресурсы: ненужные элементы не обрабатываются дальше в цепи. В отличие от императивных циклов (где вы фильтруете в 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
👍1
FlatMap: плоское преобразование для асинхронных подпотоков
FlatMap — мощный оператор для случаев, когда из одного элемента нужно создать подпоток (Publisher), и слить их в плоский результат. Это как map, но для асинхронных или множественных выходов: он "разворачивает" вложенные потоки. Полезен для запросов в цикле: например, для каждого пользователя — асинхронно запросить данные.
Пример на Flux:
Асинхронный пример: симулируем API-запросы.
Почему 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
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
👍2