Mono<String> coldMono = Mono.fromCallable(() -> {
     System.out.println("Генерация данных...");
     return "Данные";
});

coldMono.subscribe(System.out::println); // Вывод: "Генерация данных..." и "Данные"
coldMono.subscribe(System.out::println); // Снова: "Генерация данных..." и "Данные"
Каждый subscribe() вызывает fromCallable заново — данные свежие, но если это запрос в API, будет два вызова.

   Flux<Integer> coldFlux = Flux.range(1, 3).doOnSubscribe(sub -> System.out.println("Новая подписка!"));
   coldFlux.subscribe(val -> System.out.println("Подписчик 1: " + val));
   coldFlux.subscribe(val -> System.out.println("Подписчик 2: " + val));
   // Вывод: "Новая подписка!" + 1,2,3 для первого; "Новая подписка!" + 1,2,3 для второго

Каждый подписчик получает полный поток независимо. Полезно для idempotent операций (без side-effects), как чтение статичных данных.
Асинхронный пример: coldFlux = Flux.interval(Duration.ofSeconds(1)).take(3); // Каждый subscribe() запускает свой таймер.

ConnectableFlux<Integer> hotFlux = Flux.push(sink -> {
// Симулируем внешний источник
new Thread(() -> {
   for (int i = 1; i <= 3; i++) {
   try { Thread.sleep(1000); } catch (InterruptedException e) {}
   sink.next(i);
}
   sink.complete();
   }).start();
});

hotFlux.subscribe(val -> System.out.println("Подписчик 1: " + val));
Thread.sleep(1500); // Ждём, чтобы пропустить начало
hotFlux.subscribe(val -> System.out.println("Подписчик 2: " + val));
hotFlux.connect(); // Запуск горячего

// Вывод примерно: Подписчик 1: 1 (1с), 2 (2с), 3 (3с); Подписчик 2: 2 (присоединился после 1), 3

Второй пропустил 1 — поток общий. Connect() — триггер для ConnectableFlux (обёртка для горячих).

Flux.interval(Duration.ofSeconds(1)) — бесконечный таймер, вещает независимо.
Оператор share(): Делает холодный горячим.
Flux<Integer> shared = coldFlux.share();
shared.subscribe(...); // Запускает
shared.subscribe(...); // Присоединяется к существующему

import java.util.concurrent.*;

public class Task241025 {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(2);
        
        System.out.print(queue.offer(1) + " ");
        System.out.print(queue.offer(2) + " ");
        System.out.print(queue.offer(3, 100, TimeUnit.MILLISECONDS) + " ");
        
        System.out.print(queue.poll() + " ");
        System.out.print(queue.offer(3) + " ");
        System.out.print(queue.poll() + " ");
        System.out.print(queue.poll() + " ");
    }
}