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

public class Book {}

xml<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
</dependencies>
spring-boot-starter-aop включает всё необходимое для АОП.

javaimport org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.EnableAspectJAutoProxy;

@SpringBootApplication
@EnableAspectJAutoProxy // Включает автоматическое создание прокси для аспектов
public class AopApplication {
    public static void main(String[] args) {
        SpringApplication.run(AopApplication.class, args);
    }
}
Это говорит Spring: "Используй АОП с прокси".

javaimport org.springframework.stereotype.Service;

@Service // Обозначает, что это сервис, Spring создаст экземпляр
public class MyService {
    public String doSomething(String input) {
        return "Результат: " + input.toUpperCase(); // Простая логика
    }
}

javaimport org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.springframework.stereotype.Component;

@Aspect // Это аспект
@Component // Spring зарегистрирует его
public class LoggingAspect {
    @Before("execution(* com.example.service.*.*(..))") // Pointcut: все методы в пакете service
    public void logBefore(JoinPoint joinPoint) { // JoinPoint — информация о точке
        System.out.println("Вызов метода: " + joinPoint.getSignature().getName());
        System.out.println("Аргументы: " + Arrays.toString(joinPoint.getArgs()));
    }
}
Здесь @Before значит "выполни перед методом". execution — выражение для pointcut: * значит любой возврат, com.example.service..(..) — любой класс в пакете service, любой метод с любыми аргументами.

Если вызвать myService.doSomething("hello"), в консоли увидишь лог перед результатом.

javaimport org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.stereotype.Component;

@Aspect
@Component
public class TimingAspect {
    @Around("execution(* com.example.service.*.*(..))")
    public Object measureTime(ProceedingJoinPoint joinPoint) throws Throwable {
        long start = System.currentTimeMillis();
        Object result = joinPoint.proceed(); // Выполняет оригинальный метод
        long end = System.currentTimeMillis();
        System.out.println("Время выполнения: " + (end - start) + " мс");
        return result; // Возвращает результат метода
    }
}
ProceedingJoinPoint позволяет контролировать вызов: можно пропустить метод, изменить аргументы или результат. Идеально для транзакций или кэша.

import reactor.core.publisher.Flux;
Flux<String> originalFlux = Flux.just("яблоко", "банан", "вишня");
Flux<String> transformed = originalFlux.map(fruit -> fruit.toUpperCase()); // Преобразование в верхний регистр
transformed.subscribe(System.out::println); // Вывод: "ЯБЛОКО", "БАНАН", "ВИШНЯ"

Здесь map применяет лямбду к каждому элементу последовательно. Если ошибка в функции — сработает onError.

Mono<Integer> num = Mono.just(5).map(x -> x * 2); // Результат: 10

Flux<Integer> numbers = Flux.range(1, 10);
Flux<Integer> evenNumbers = numbers.filter(num -> num % 2 == 0); // Только чётные
evenNumbers.subscribe(System.out::println); // Вывод: 2, 4, 6, 8, 10

Если поток пустой или ничего не проходит — onComplete сработает без onNext.

Mono<String> word = Mono.just("привет").filter(w -> w.length() > 7); // Не пройдёт — пустой Mono

Flux<String> fruits = Flux.just("яблоко", "банан");
Flux<Character> letters = fruits.flatMap(fruit -> Flux.fromArray(fruit.toCharArray())); // Из строки — поток символов
letters.subscribe(System.out::println); // Вывод: я, б, л, о, к, о, б, а, н, а, н (в возможном перемешанном порядке, если асинхронно)

Здесь flatMap берёт строку, создаёт Flux из символов и сливает всё в один поток. В отличие от map (который вернул бы Flux<Flux<Character>> — вложенный), flatMap "сплющивает".

import java.time.Duration;
Flux<String> users = Flux.just("user1", "user2");
Flux<String> data = users.flatMap(user -> Mono.just("Данные для " + user).delayElement(Duration.ofSeconds(1))); // Асинхронный подпоток с задержкой
data.subscribe(System.out::println); // Вывод через секунды: "Данные для user1", "Данные для user2" (параллельно, если scheduler позволяет)

FlatMap уважает backpressure: запрашивает у подпотоков по мере нужды. Но осторожно: если подпотоки бесконечные — рискуете перегрузкой. Параметр concurrency (flatMap(func, concurrency)) ограничивает параллелизм.

import java.util.HashSet;
import java.util.Set;

public class Main {
    public static void main(String[] args) {
        Set<String> fruits = new HashSet<>();
        fruits.add("Яблоко");
        fruits.add("Банан");
        fruits.add("Яблоко");  // Игнорируется

        System.out.println(fruits.size());  // 2
        System.out.println(fruits.contains("Банан"));  // true

        for (String fruit : fruits) {
            System.out.println(fruit);  // Порядок непредсказуем
        }
    }
}


Вывод: Размер 2, содержит "Банан", элементы в случайном порядке.

import java.util.LinkedHashSet;
import java.util.Set;

public class Main {
    public static void main(String[] args) {
        Set<String> fruits = new LinkedHashSet<>();
        fruits.add("Яблоко");
        fruits.add("Банан");
        fruits.add("Апельсин");

        for (String fruit : fruits) {
            System.out.println(fruit);  // Яблоко, Банан, Апельсин — порядок вставки
        }
    }
}

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);

        for (Integer num : numbers) {
            System.out.println(num);  // 1, 3, 5 — отсортировано
        }
    }
}

import reactor.core.publisher.Flux;
import java.time.Duration;
Flux<String> first = Flux.just("Шаг 1a", "Шаг 1b").delayElements(Duration.ofSeconds(1)); // Задержка для симуляции
Flux<String> second = Flux.just("Шаг 2a", "Шаг 2b");
Flux<String> combined = Flux.concat(first, second);

combined.subscribe(System.out::println); // Вывод: "Шаг 1a" (через 1с), "Шаг 1b" (ещё 1с), "Шаг 2a", "Шаг 2b"

Здесь concat гарантирует последовательность: второй Flux стартует только после onComplete первого. Если ошибка в первом — весь поток прервётся onError.

Mono<String> m1 = Mono.just("A").delayElement(Duration.ofSeconds(1));
Mono<String> m2 = Mono.just("B");
Flux<String> seq = Flux.concat(m1, m2); // Mono как Flux с одним элементом

Flux<String> slow = Flux.just("Медленный 1", "Медленный 2").delayElements(Duration.ofSeconds(2));
Flux<String> fast = Flux.just("Быстрый A", "Быстрый B").delayElements(Duration.ofMillis(500));
Flux<String> merged = Flux.merge(slow, fast);

merged.subscribe(System.out::println); // Возможный вывод: "Быстрый A" (0.5с), "Быстрый B" (ещё 0.5с), "Медленный 1" (2с), "Медленный 2" (ещё 2с)

Здесь merge отдает элементы, как только они готовы — параллельно. Если ошибка в одном — весь поток onError (по умолчанию), но можно настроить.

Flux<Integer> xCoords = Flux.just(1, 2, 3);
Flux<Integer> yCoords = Flux.just(10, 20, 30, 40); // Лишний элемент игнорируется
Flux<String> points = Flux.zip(xCoords, yCoords, (x, y) -> "(" + x + ", " + y + ")");

points.subscribe(System.out::println); // Вывод: "(1, 10)", "(2, 20)", "(3, 30)"

Здесь zip ждёт пару: если один медленный — задерживает. Для >2 потоков: zip(tuple -> ..., flux1, flux2, flux3).

Flux<String> stockA = Flux.just("A:100", "A:110").delayElements(Duration.ofSeconds(1));
Flux<String> stockB = Flux.just("B:200").delayElements(Duration.ofSeconds(2));
Flux<String> latest = Flux.combineLatest(stockA, stockB, (a, b) -> a + " + " + b);

latest.subscribe(System.out::println); // Вывод примерно: "A:100 + B:200" (после 2с), "A:110 + B:200" (ещё 1с после)

Здесь combineLatest реагирует на изменения: при обновлении A использует последний B. Для >2: combineLatest(tuple -> ..., fluxes).