Что такое реактивное программирование в Java и когда его применяют?

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

Реактивное программирование (Reactive Programming) — это парадигма, которая сосредоточена на потоках данных и распространении изменений. Вместо императивного подхода «вычисли, а потом используй», реактивный подход строится на том, что система автоматически реагирует на изменения данных.

Основные концепции

Реактивные потоки (Streams) — это основа реактивного программирования. Поток может испускать значения, ошибки или сигнал завершения.

// Простой пример с RxJava
Observable.just(1, 2, 3)
    .map(x -> x * 2)
    .filter(x -> x > 2)
    .subscribe(
        value -> System.out.println("Значение: " + value),
        error -> System.err.println("Ошибка: " + error),
        () -> System.out.println("Завершено")
    );

Основные библиотеки

RxJava — самая популярная библиотека реактивного программирования для Java.

// Пример обработки данных
Observable<User> userStream = Observable
    .fromIterable(getUserIds())
    .flatMap(userId -> getUserFromApi(userId))
    .filter(user -> user.isActive())
    .timeout(5, TimeUnit.SECONDS)
    .retry(3);

userStream.subscribe(
    user -> processUser(user),
    error -> handleError(error)
);

Project Reactor — реактивная библиотека от Spring, часть Spring WebFlux.

// Монада Mono — для одного значения
Mono<User> userMono = userRepository.findById(1L)
    .flatMap(user -> enrichUserData(user))
    .timeout(Duration.ofSeconds(5))
    .onErrorResume(this::handleError);

// Flux — для множества значений
Flux<User> userFlux = userRepository.findAll()
    .buffer(10)
    .flatMap(batch -> processBatch(batch))
    .doOnError(e -> logger.error("Ошибка: ", e))
    .retry(2);

Когда применяют реактивное программирование

1. Высоконагруженные системы

• Обработка большого количества параллельных запросов
• Минимальное использование потоков благодаря event-loop модели

@GetMapping("/users/{id}")
public Mono<ResponseEntity<UserDto>> getUser(@PathVariable Long id) {
    return userService.findById(id)
        .map(ResponseEntity::ok)
        .defaultIfEmpty(ResponseEntity.notFound().build());
}

2. Микросервисная архитектура

• Асинхронные вызовы между сервисами
• Non-blocking I/O для лучшей производительности

3. WebSocket и real-time приложения

• Трансляция событий клиентам в реальном времени
• Bi-directional communication

@RestController
public class EventController {
    private final EventService eventService;
    
    @GetMapping("/events/stream")
    public Flux<ServerSentEvent<EventDto>> streamEvents() {
        return eventService.getEventStream()
            .map(event -> ServerSentEvent.builder(event).build())
            .doOnCancel(() -> logger.info("Клиент отключился"));
    }
}

4. Большие объёмы данных (Big Data)

• Обработка потоков данных
• Backpressure — управление скоростью обработки

Flux.range(1, 1000000)
    .onBackpressureBuffer(100, BufferOverflowStrategy.DROP_OLDEST)
    .parallel()
    .runOn(Schedulers.parallel())
    .map(this::processData)
    .sequential()
    .subscribe();

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

• map — преобразование значений
• filter — фильтрация элементов
• flatMap — плоское преобразование
• reduce — агрегация значений
• merge — объединение нескольких потоков
• zip — комбинирование данных из разных потоков
• timeout — установка таймаута
• retry — повторные попытки при ошибке

Преимущества

• Высокая производительность и масштабируемость
• Неблокирующие операции (non-blocking I/O)
• Удобная обработка асинхронных операций
• Встроенная обработка ошибок
• Функциональный стиль кода

Недостатки

• Кривая обучения
• Сложность отладки
• Может быть излишним для простых приложений

Реактивное программирование — мощный инструмент для систем, требующих высокой пропускной способности и низких задержек.