Flux<integer> fastFlux = Flux.range(1, 1000); // Быстрый источник
fastFlux.subscribe(new BaseSubscriber<integer>() {

@Override
protected void hookOnSubscribe(Subscription subscription) {
        request(10); // Запрашиваем только 10 элементов сначала
}

@Override
protected void hookOnNext(Integer value) {
          System.out.println(value);
          if (value % 10 == 0) request(10); // Запрашиваем ещё по мере обработки
          }
});

List<String> fruits = Arrays.asList("яблоко", "банан", "вишня");

Iterator<String> iterator = fruits.iterator();

while (iterator.hasNext()) {
    String fruit = iterator.next(); // "Вытягиваем" следующий элемент
    System.out.println(fruit.toUpperCase());
}

Flux<String> pushFlux = Flux.fromIterable(Arrays.asList("яблоко", "банан", "вишня"))
                    .map(String::toUpperCase); // Преобразование в потоке

pushFlux.subscribe(
               fruit -> System.out.println("Получено из push: " + fruit), // Реакция на толкание
               Throwable::printStackTrace, // Обработка ошибок
               () -> System.out.println("Push завершён")
);

WebClient.create()
     .get()
     .uri("https://api.example.com/fruits")
     .retrieve()
     .bodyToFlux(String.class) // Flux толкает строки из ответа
     .subscribe(fruit -> System.out.println("Push из API: " + fruit));

Flux.fromIterable(fruits)
     .map(String::toUpperCase)
     .flatMap(fruit -> sendToApi(fruit)) // flatMap толкает в асинхронный API
     .subscribe(result -> System.out.println("Ответ: " + result));

import reactor.core.publisher.Flux; // Flux реализует Publisher

Flux<Integer> publisher = Flux.range(1, 5); // Издатель: поток от 1 до 5

publisher.subscribe(new Subscriber<Integer>() {
    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription s) {
        this.subscription = s;
        s.request(1); // Запрашиваем первый элемент
    }

    @Override
    public void onNext(Integer item) {
        System.out.println("Получено: " + item);
        subscription.request(1); // Запрашиваем следующий
    }

    @Override
    public void onError(Throwable t) {
        System.err.println("Ошибка: " + t);
    }

    @Override
    public void onComplete() {
        System.out.println("Завершено");
    }
});

import reactor.core.publisher.Flux;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

Flux<Integer> fastPublisher = Flux.range(1, 1000); // Быстрый источник: 1000 элементов

fastPublisher.subscribe(new Subscriber<Integer>() {
    private Subscription sub;

    @Override
    public void onSubscribe(Subscription s) {
        sub = s;
        sub.request(5); // Сначала запрашиваем 5 элементов
    }

    @Override
    public void onNext(Integer item) {
        System.out.println("Обработано: " + item);
        // Симулируем медленную обработку: Thread.sleep(100); (но в реальности избегайте блокировок!)
        sub.request(1); // После каждого — запрашиваем следующий
    }

    @Override
    public void onError(Throwable t) { /* обработка */ }
    @Override
    public void onComplete() { System.out.println("Готово"); }
});

Здесь подписчик контролирует темп: запросил 5 — получил 5, обработал — запросил ещё. Если не request() — поток остановится. Это асинхронно: издатель не блокируется, а ждёт сигналов.

Flux.interval(Duration.ofMillis(1)) // Бесконечный поток: элемент каждую мс
    .onBackpressureDrop(dropped -> System.out.println("Сброшено: " + dropped)) // Стратегия: сбрасываем лишнее
    .subscribe(item -> {
        try { Thread.sleep(100); } catch (InterruptedException e) {} // Медленный подписчик
        System.out.println("Обработано: " + item);
    });

Здесь быстрый Flux "замедляется" под подписчика: лишние элементы сбрасываются, система не падает.

Ещё опция — cancel(): подписчик может полностью отменить подписку, если данных слишком много или не нужно.

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.5.0</version> <!-- версия может быть неактуальной -->
</dependency>

import reactor.core.publisher.Mono; import reactor.core.publisher.Flux;

Mono<String> simpleMono = Mono.just("Привет из Reactor");

Подписка: subscribe() вызывает реакции.

simpleMono.subscribe(
    value -> System.out.println("Значение: " + value), // onNext: реакция на элемент
    error -> System.err.println("Ошибка: " + error),   // onError
    () -> System.out.println("Завершено")             // onComplete
);

Здесь: "Значение: Привет из Reactor" и "Завершено". Подписка асинхронна — код после subscribe() идёт сразу, без ожидания. Если ошибка (например, Mono.error(new RuntimeException("Бум"))), сработает onError.

Mono<String> asyncMono = Mono.delay(Duration.ofSeconds(1)).map(ignore -> "Готово после секунды");

asyncMono.subscribe(System.out::println); // Вывод после 1 сек

Flux<Integer> numbersFlux = Flux.range(1, 5);

numbersFlux.subscribe(
    value -> System.out.println("Элемент: " + value), // onNext для каждого
    error -> System.err.println("Ошибка: " + error),
    () -> System.out.println("Завершено")
);

Вывод: 1, 2, 3, 4, 5 и "Завершено". Асинхронно: элементы "текут" по мере готовности.

numbersFlux.subscribe(new BaseSubscriber<Integer>() {
    @Override
    protected void hookOnSubscribe(Subscription subscription) {
        request(2); // Запрашиваем по 2 элемента
    }
    @Override
    protected void hookOnNext(Integer value) {
        System.out.println("Получено: " + value);
        request(1); // После обработки — следующий
    }
});

Это предотвращает перегрузку: Flux выдаёт элементы порциями.

Flux<Long> tickingFlux = Flux.interval(Duration.ofMillis(500)).take(5); // 5 тиков каждые 0.5 сек

tickingFlux.subscribe(System.out::println); // 0, 1, 2, 3, 4 с паузами

import reactor.core.publisher.Flux;

Flux<String> fruitsFlux = Flux.just("яблоко", "банан", "вишня");
fruitsFlux.subscribe(
   fruit -> System.out.println("Съедено: " + fruit), // onNext: реакция на каждый элемент
   error -> System.err.println("Проблема: " + error.getMessage()), // onError: если ошибка
   () -> System.out.println("Фрукты закончились") // onComplete: завершение
);

Вывод: "Съедено: яблоко", "Съедено: банан", "Съедено: вишня", "Фрукты закончились". Если добавить ошибку — Flux.just("яблоко").concatWith(Flux.error(new RuntimeException("Гнилой фрукт"))) — сработает onError, и onComplete не вызовется.

Disposable disposable = fruitsFlux.subscribe(...);
disposable.dispose(); // Отмена: поток остановится, onComplete не сработает.
Это полезно для UI или долгоживущих потоков: отпишись, когда компонент уничтожен, чтобы избежать утечек памяти.

import reactor.core.publisher.BaseSubscriber;

Flux<Integer> numbersFlux = Flux.range(1, 10).map(i -> {
   if (i == 5) throw new RuntimeException("Ошибка на 5"); // Симулируем ошибку
   return i;
});

numbersFlux.subscribe(new BaseSubscriber<Integer>() {

  @Override
  protected void hookOnSubscribe(Subscription subscription) {
  System.out.println("Подписка готова");
  request(3); // Запрашиваем первые 3
}

@Override
protected void hookOnNext(Integer value) {
   System.out.println("Элемент: " + value);
   request(1); // Запрашиваем по одному дальше
}

@Override
   protected void hookOnError(Throwable throwable) {
   System.err.println("Ошибка: " + throwable.getMessage());
}

@Override
   protected void hookOnComplete() {
   System.out.println("Цикл завершён");
}
});

Вывод: "Подписка готова", "Элемент: 1", "Элемент: 2", "Элемент: 3", "Элемент: 4", "Ошибка: Ошибка на 5". onComplete не сработает, потому что ошибка. Без ошибки — все 10 элементов и "Цикл завершён".

Это демонстрирует контроль: request() управляет темпом, как в backpressure (пост 5). Если не request() — только "Подписка готова".

Mono<String> riskyMono = Mono.fromCallable(() -> {
   if (Math.random() > 0.5) throw new RuntimeException("Сбой");
   return "Успех";
}).onErrorReturn("Fallback").retry(2); // Retry 2 раза

riskyMono.subscribe(
   System.out::println,
   error -> System.out.println("Не удалось: " + error),
   () -> System.out.println("Завершено")
);

Если сбой — retry, потом fallback. onComplete сработает только при успехе или fallback.

Для onComplete: используйте doFinally(Runnable) — сработает всегда, после onComplete или onError. Полезно для закрытия ресурсов: doFinally(() -> connection.close()).