import reactor.core.scheduler.Schedulers;

Schedulers.parallel(); // Готов к использованию

Schedulers не создают потоки заранее — ленивые, как Mono/Flux.

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

Flux<Integer> flux = Flux.range(1, 3)
    .map(i -> { // На subscribeOn
        System.out.println("Map1 на потоке: " + Thread.currentThread().getName());
        return i * 10;
    })
    .publishOn(Schedulers.boundedElastic()) // Переключаем downstream
    .delayElements(Duration.ofSeconds(1)) // На boundedElastic
    .map(i -> { // На boundedElastic
        System.out.println("Map2 на потоке: " + Thread.currentThread().getName());
        return i + 1;
    });

flux.subscribeOn(Schedulers.single()) // Upstream на single
    .subscribe(System.out::println);

Вывод покажет: Map1 на "single-1", delay и Map2 на "boundedElastic-X". Это асинхронно: основной поток не блокируется.

Flux.range(1, 10)
    .flatMap(i -> Mono.just(i).delayElement(Duration.ofMillis(100)), 2) // Max 2 параллельно
    .subscribeOn(Schedulers.parallel())
    .blockLast(); // Для теста, в prod не блокируйте!

Это ограничивает: не 10 задержек сразу, а по 2.

import org.springframework.web.reactive.function.client.WebClient;

WebClient client = WebClient.builder()
    .baseUrl("https://api.example.com")
    .defaultHeader("Authorization", "Bearer token")
    .build();

Это готово к запросам. Для кодеков (JSON) — добавьте Jackson или Gson.

Mono<String> responseMono = client.get()
    .uri("/users/{id}", 1) // URI с параметрами
    .retrieve()
    .bodyToMono(String.class); // Декодируем в String

responseMono.subscribe(System.out::println); // Асинхронно: ответ придёт позже

Для JSON: bodyToMono(User.class) с Jackson.

Flux<User> usersFlux = Flux.just(new User("Alice"), new User("Bob"));

Mono<Void> postMono = client.post()
    .uri("/users/batch")
    .body(usersFlux, User.class) // Тело как Flux
    .retrieve()
    .bodyToMono(Void.class);

postMono.subscribe(); // Отправка асинхронно

WebClient уважает backpressure: если сервер шлёт Flux (SSE), клиент запрашивает по мере обработки.

С Schedulers: по умолчанию на event-loop, но publishOn для кастом.

client.get()
    .uri("/data")
    .retrieve()
    .onStatus(HttpStatusCode::is4xxClientError, resp -> Mono.error(new BadRequestException("Ошибка клиента")))
    .bodyToFlux(Data.class)
    .doOnError(e -> System.err.println("Сбой: " + e)) // Из поста 10
    .retry(3) // Retry на сетевые ошибки
    .subscribe(data -> process(data));

Это интегрирует с Reactor: ошибки как onError, retry для flaky API.

Для стриминга: bodyToFlux для больших данных, без загрузки всего в память.

import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

import java.time.Duration;

public class Task141025 {
    public static void main(String[] args) {
        WebClient webClient = WebClient.builder()
                .baseUrl("http://example.com")
                .build();

        Mono<String> result = webClient.get()
                .uri("/api/data")
                .retrieve()
                .bodyToMono(String.class)
                .timeout(Duration.ofMillis(100))
                .onErrorResume(throwable -> {
                    System.out.println("Timeout occurred!");
                    return Mono.just("Fallback");
                })
                .flatMap(response -> {
                    System.out.println("Processing: " + response);
                    return Mono.just("Processed: " + response);
                });

        System.out.println("Final: " + result.block());
    }
}