Реактивное программирование в Java: библиотека Reactor

Реактивное программирование в Java с Project Reactor

Реактивное программирование — это парадигма разработки, ориентированная на асинхронные потоки данных и распространение изменений. В Java эта концепция реализована через спецификацию Reactive Streams и её самую популярную реализацию — библиотеку Project Reactor (часть экосистемы Spring WebFlux).

В отличие от традиционного императивного кода, где программа последовательно выполняет команды, реактивный подход позволяет обрабатывать данные по мере их поступления, не блокируя потоки выполнения. Это особенно важно для высоконагруженных систем, микросервисов и приложений с большим количеством одновременных запросов.

Основные компоненты Reactor

Библиотека предоставляет два основных типа реактивных последовательностей:

• Mono<T> — представляет собой поток, который может содержать 0 или 1 элемент. Используется для единичных асинхронных операций (например, HTTP-запрос, получение записи из БД).
• Flux<T> — представляет собой поток, который может содержать от 0 до N элементов. Подходит для коллекций, потоков событий или данных, поступающих частями.

Оба типа реализуют интерфейс Publisher из спецификации Reactive Streams и поддерживают механизм backpressure (противодавление) — возможность подписчика контролировать скорость получения данных.

Пример: создание и обработка Flux

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

import reactor.core.publisher.Flux;

public class ReactorExample {
    public static void main(String[] args) {
        Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);

        numbers
            .map(n -> n * 2)          // преобразование
            .filter(n -> n % 3 == 0)   // фильтрация
            .subscribe(
                data -> System.out.println("Получено: " + data),
                error -> System.err.println("Ошибка: " + error),
                () -> System.out.println("Завершено!")
            );
    }
}

В этом примере:

• Flux.just() создаёт поток из пяти чисел.
• map() умножает каждый элемент на 2.
• filter() оставляет только числа, кратные 3.
• subscribe() запускает обработку — выводит результат, обрабатывает ошибки и сигнализирует о завершении.

Асинхронность и работа с Mono

Реактор позволяет выполнять асинхронные операции без блокировки потоков. Например, симуляция запроса к удалённому сервису:

import reactor.core.publisher.Mono;
import java.time.Duration;

public class AsyncExample {
    public static void main(String[] args) throws InterruptedException {
        Mono<String> asyncResult = Mono.fromSupplier(() -> {
            try {
                Thread.sleep(1000); // имитация задержки
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Данные получены";
        }).subscribeOn(Schedulers.boundedElastic()); // выполняем в другом потоке

        System.out.println("Запрос отправлен...");
        
        asyncResult.subscribe(System.out::println);
        
        Thread.sleep(2000); // ждём завершения
    }
}

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

Обработка ошибок в реактивных потоках

В отличие от try-catch, в реактивном программировании ошибки обрабатываются через операторы:

Flux<String> fluxWithError = Flux.just("A", "B", "C")
    .map(s -> {
        if ("B".equals(s)) throw new RuntimeException("Ошибка на B");
        return s.toLowerCase();
    })
    .onErrorResume(e -> Flux.just("fallback1", "fallback2"));

fluxWithError.subscribe(
    System.out::println,
    err -> System.err.println("Это не вызовется, т.к. ошибка перехвачена")
);

Оператор onErrorResume позволяет подменить ошибочный поток на резервный. Существуют также onErrorReturn (вернуть одно значение) и retry() (повторить операцию).

Backpressure: управление нагрузкой

Одно из главных преимуществ реактивного подхода — защита от перегрузки. Подписчик может запрашивать данные порциями:

Flux.range(1, 100)
    .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            request(5); // запрашиваем только 5 элементов
        }

        @Override
        protected