import reactor.core.publisher.Flux;
import java.io.IOException;

Flux<String> riskyFlux = Flux.just("данные1", "данные2").map(data -> {
    if (data.equals("данные2")) throw new IOException("Сбой ввода-вывода");
    return data.toUpperCase();
});

riskyFlux
    .doOnError(e -> System.err.println("Лог: " + e.getMessage())) // Логируем
    .onErrorMap(e -> new RuntimeException("Обёрнутая ошибка: " + e)) // Преобразуем
    .subscribe(
        System.out::println,
        error -> System.err.println("Финальная ошибка: " + error) // onError в подписке
    );

// Вывод: "ДАННЫЕ1", потом лог "Сбой ввода-вывода", и финальная "Обёрнутая ошибка: ..."

Здесь doOnError срабатывает перед onErrorMap, а подписка ловит модифицированную ошибку. Это асинхронно: если ошибка в асинхронном подпотоке (flatMap), Reactor передаёт её downstream (дальше по цепи) без блокировок.

Mono<String> httpMono = Mono.fromCallable(() -> {
    // Симулируем HTTP-запрос
    throw new RuntimeException("Сервер не отвечает");
}).onErrorReturn("Кэшированные данные");

httpMono.subscribe(System.out::println); // Вывод: "Кэшированные данные", потом onComplete

Здесь ошибка преобразуется в значение, поток завершается успешно.

Flux<Integer> calcFlux = Flux.range(1, 5).map(i -> {
    if (i == 3) throw new ArithmeticException("Деление на ноль");
    return 10 / (i - 3); // Симуляция
}).onErrorResume(e -> {
    if (e instanceof ArithmeticException) {
        return Flux.just(0, 0); // Fallback на нули
    } else {
        return Flux.error(e); // Пропустить другие ошибки
    }
});

calcFlux.subscribe(System.out::println); // Вывод: элементы до ошибки, потом 0, 0, onComplete

Mono<String> flakyMono = Mono.defer(() -> {
    if (Math.random() > 0.3) throw new RuntimeException("Временный сбой");
    return Mono.just("Успех");
}).retry(3); // Повторить 3 раза

flakyMono.subscribe(System.out::println, Throwable::printStackTrace);

Здесь retry повторяет весь Mono при ошибке, до успеха или исчерпания попыток. Если все попытки fail — финальный onError.

import reactor.util.retry.Retry;

flakyMono.retryWhen(Retry.backoff(3, Duration.ofSeconds(1))); // 3 попытки с задержкой 1с, 2с, 4с

import reactor.core.publisher.Mono;

public class MonoExample {
    public static void main(String[] args) {
        Mono.just("Start")
            .doOnNext(System.out::print)
            .then(Mono.fromRunnable(() -> System.out.print("Middle")))
            .thenReturn("End")
            .subscribe(System.out::print);
    }
}