<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 java.util.*;

public class Task300925 {
    public static void main(String[] args) {
        Set<Short> set = new HashSet<>();
        
        for (short i = 0; i < 10; i++) {
            set.add(i);
            set.remove(i - 1);
        }
        
        System.out.println(set.size());
    }
}

import java.util.HashSet;
import java.util.Set;
import java.util.ArrayList;
import java.util.List;

public class Main {
    public static void main(String[] args) {
        // Set: Уникальные элементы
        Set<String> uniqueNames = new HashSet<>();
        uniqueNames.add("Алексей");
        uniqueNames.add("Алексей");  // Игнорируется
        System.out.println(uniqueNames.size());  // 1

        // List: Дубликаты разрешены
        List<String> names = new ArrayList<>();
        names.add("Алексей");
        names.add("Алексей");
        System.out.println(names.size());  // 2
    }
}

import java.util.HashSet;
import java.util.LinkedHashSet;

public class Main {
    public static void main(String[] args) {
        // HashSet: Без порядка
        Set<String> hashSet = new HashSet<>();
        hashSet.add("Яблоко");
        hashSet.add("Банан");
        hashSet.add("Апельсин");
        System.out.println(hashSet);  // Может быть [Банан, Яблоко, Апельсин] — непредсказуемо

        // LinkedHashSet: Порядок вставки
        Set<String> linkedSet = new LinkedHashSet<>();
        linkedSet.add("Яблоко");
        linkedSet.add("Банан");
        linkedSet.add("Апельсин");
        System.out.println(linkedSet);  // [Яблоко, Банан, Апельсин] — сохраняет порядок
    }
}

import java.util.TreeSet;
import java.util.Set;

public class Main {
    public static void main(String[] args) {
        Set<Integer> numbers = new TreeSet<>();
        numbers.add(5);
        numbers.add(1);
        numbers.add(3);
        System.out.println(numbers);  // [1, 3, 5] — отсортировано

        // Custom Comparator для обратного порядка
        Set<Integer> reverseNumbers = new TreeSet<>((a, b) -> b - a);
        reverseNumbers.add(5);
        reverseNumbers.add(1);
        reverseNumbers.add(3);
        System.out.println(reverseNumbers);  // [5, 3, 1]
    }
}

import java.util.*;

public class Task011025 {
    public static void main(String[] args) {
        Queue<Integer> queue = new LinkedList<>();
        queue.offer(1);
        queue.offer(2);
        queue.offer(3);
        
        ((LinkedList<Integer>) queue).removeLast();
        queue.poll();
        
        System.out.println(queue.peek());
    }
}

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()).

import java.util.*;

public class Task021025 {
    public static void main(String[] args) {
        Map<String, Integer> map1 = new HashMap<>();
        Map<String, Integer> map2 = new HashMap<>();
        
        map1.put("a", 1);
        map1.put("b", 1);
        
        map2.put("b", 1);
        map2.put("a", 1);
        
        System.out.println(map1.equals(map2));
        System.out.println(Objects.equals(map1, map2));
    }
}