Какие плюсы и минусы Spring Reactor?

Spring Reactor: Плюсы и минусы

Что такое Spring Reactor?

Spring Reactor - это реактивная библиотека для асинхронного программирования в Java. Реализует спецификацию Reactive Streams.

// Imperative (традиционный способ)
List<User> users = userRepository.findAll();
for (User user : users) {
    System.out.println(user.getName());
}

// Reactive (Spring Reactor)
Flux<User> users = userRepository.findAllReactive();
users.subscribe(user -> System.out.println(user.getName()));

Ключевые компоненты

Mono - реактивный тип для 0 или 1 элемента. Flux - реактивный тип для 0..N элементов.

// Mono (как Optional)
Mono<User> user = userRepository.findByIdReactive(1L);
user.subscribe(
    data -> System.out.println("User: " + data),
    error -> System.err.println("Error: " + error),
    () -> System.out.println("Complete")
);

// Flux (как List)
Flux<User> users = userRepository.findAllReactive();
users.subscribe(
    user -> System.out.println(user.getName()),
    error -> System.err.println("Error: " + error),
    () -> System.out.println("All users loaded")
);

ПЛЮСЫ Spring Reactor

1. Неблокирующий I/O (Non-blocking)

Плюс: Один поток может обрабатывать множество requests.

// Традиционный Spring MVC
@RestController
public class UserController {
    @GetMapping("/{id}")
    public User getUser(@PathVariable Long id) {
        return userRepository.findById(id).orElseThrow();
        // Поток БЛОКИРУЕТСЯ, ожидая результата БД
        // Если 1000 requests - нужно 1000 потоков!
    }
}

// Spring WebFlux + Reactor
@RestController
public class UserController {
    @GetMapping("/{id}")
    public Mono<User> getUser(@PathVariable Long id) {
        return userRepository.findByIdReactive(id);
        // Один поток обрабатывает N requests!
        // Асинхронный результат
    }
}

// Без Reactor: 10,000 requests → 10,000 потоков → 10GB памяти
// С Reactor: 10,000 requests → 10-100 потоков → 100MB памяти

2. Backpressure (управление потоком данных)

Плюс: Consumer может сказать Producer: "Дай мне 100 элементов, а не все сразу".

// Без Backpressure (может быть OutOfMemory)
Flux<DataChunk> dataStream = fetchLargeDataset();
List<DataChunk> allData = new ArrayList<>();
dataStream.subscribe(chunk -> allData.add(chunk));  // Все в памяти!

// С Backpressure
Flux<DataChunk> dataStream = fetchLargeDataset();
dataStream
    .buffer(100)  // Буферизируем по 100
    .subscribe(chunks -> processChunk(chunks));

// Или более контролируемо
dataStream
    .limitRate(100)  // Обрабатываем 100 за раз
    .subscribe(chunk -> processChunk(chunk));

Результат: Можно обрабатывать файлы в 1GB+ без загрузки всего в памяти.

3. Лучше читаемость с pipeline паттерном

Плюс: Функциональный стиль делает код понятнее.

// Reactor pipeline
Flux.range(1, 100)
    .filter(n -> n % 2 == 0)        // Только чётные
    .map(n -> n * n)                 // Возвести в квадрат
    .take(10)                        // Первые 10
    .subscribe(System.out::println); // 4, 16, 36, ...

// С реальными данными
Flux.fromIterable(users)
    .filter(u -> u.getAge() > 18)
    .map(u -> new UserDTO(u.getName(), u.getEmail()))
    .flatMap(dto -> enrichWithStats(dto))  // Асинхронное обогащение
    .collect(Collectors.toList())
    .subscribe(dtos -> System.out.println(dtos));

4. Композиция асинхронных операций

Плюс: Легко комбинировать разные асинхронные источники.

// Множественные асинхронные calls
Mono<User> user = userService.getUser(1L);
Mono<List<Order>> orders = orderService.getOrders(1L);
Mono<Profile> profile = profileService.getProfile(1L);

// Объединить все три
Mono<UserWithData> combined = Mono.zip(user, orders, profile)
    .map(tuple -> new UserWithData(
        tuple.getT1(),  // User
        tuple.getT2(),  // Orders
        tuple.getT3()   // Profile
    ));

combined.subscribe(data -> System.out.println(data));

// Параллельное выполнение: все три вызова идут параллельно!
// Не последовательно: user → orders → profile

5. Отличная интеграция с Spring

Плюс: Spring WebFlux полностью построен на Reactor.

// Spring Boot с Reactor
@SpringBootApplication
public class ReactiveApp {
    public static void main(String[] args) {
        SpringApplication.run(ReactiveApp.class, args);
    }
}

// Реактивный controller
@RestController
@RequestMapping("/api/users")
public class UserController {
    @PostMapping
    public Mono<ResponseEntity<UserDTO>> createUser(@RequestBody UserDTO dto) {
        return userService.createUserAsync(dto)
            .map(user -> ResponseEntity.status(HttpStatus.CREATED).body(user))
            .onErrorResume(err -> Mono.just(
                ResponseEntity.status(HttpStatus.CONFLICT).build()
            ));
    }
    
    @GetMapping("/{id}")
    public Mono<User> getUser(@PathVariable Long id) {
        return userService.getUserAsync(id);
    }
    
    @GetMapping
    public Flux<User> getAllUsers() {
        return userService.getAllUsersAsync();
    }
}

// Реактивный сервис
@Service
public class UserService {
    private final ReactiveUserRepository repository;
    
    public Mono<User> getUserAsync(Long id) {
        return repository.findById(id);
    }
    
    public Flux<User> getAllUsersAsync() {
        return repository.findAll();
    }
}

// Реактивный репозиторий
public interface ReactiveUserRepository extends ReactiveCrudRepository<User, Long> {
    Flux<User> findByAgeGreaterThan(int age);
    Mono<User> findByEmail(String email);
}

6. Server-Sent Events (SSE) и WebSockets

Плюс: Идеально для real-time приложений.

// Server-Sent Events
@GetMapping("/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> streamEvents() {
    return Flux.interval(Duration.ofSeconds(1))
        .map(i -> "Event " + i)
        .doOnCancel(() -> System.out.println("Client disconnected"));
}

// Клиент получает: Event 0, Event 1, Event 2, ...

// WebSocket с Reactor
@Component
public class WebSocketHandler implements WebSocketHandler {
    @Override
    public Mono<Void> handle(WebSocketSession session) {
        return session.send(
            Flux.interval(Duration.ofSeconds(1))
                .map(i -> session.textMessage("Message " + i))
        ).and(
            session.receive()
                .doOnNext(msg -> System.out.println("Received: " + msg))
                .doOnComplete(() -> System.out.println("Closed"))
        );
    }
}

7. Лучшая масштабируемость

Плюс: Меньше потоков = больше одновременных соединений.

# Spring MVC (блокирующий)
server.tomcat.threads.max=200  # Макс 200 потоков
# Макс ~200 одновременных requests

# Spring WebFlux (неблокирующий)
server.netty.threads.event-loop.preferred-count=4  # 4 потока event loop
# Может обработать 100,000+ одновременных requests!

МИНУСЫ Spring Reactor

1. Кривая обучения очень крутая

Минус: Нужно переучиться мышлению.

// Традиционное мышление
User user = getUserFromDB();
List<Order> orders = getOrdersForUser(user.getId());
for (Order order : orders) {
    System.out.println(order);
}

// Реактивное мышление (Reactor)
Mono<User> user = getUserFromDBAsync();
Flux<Order> orders = user.flatMapMany(u -> getOrdersForUserAsync(u.getId()));
orders.subscribe(System.out::println);

// Нужно понимать:
// - Mono vs Flux
// - flatMap vs map vs flatMapMany
// - subscribe vs block
// - backpressure
// - scheduling

Результат: До неделю легко писать неправильный код!

2. Дебаггинг очень сложный

Минус: Stack trace неполный, асинхронный поток выполнения.

// Это выбросит ошибку, но где?
Flux.range(1, 100)
    .map(n -> {
        if (n == 50) throw new RuntimeException("Error at 50");
        return n;
    })
    .subscribe(
        System.out::println,
        error -> error.printStackTrace()  // Где ошибка? Stack trace короткий!
    );

// Нужно использовать doOnError
Flux.range(1, 100)
    .map(n -> {
        if (n == 50) throw new RuntimeException("Error at 50");
        return n;
    })
    .doOnError(error -> System.err.println("Error: " + error))
    .onErrorResume(error -> Flux.empty())
    .subscribe(System.out::println);

3. Несовместимость с блокирующим кодом

Минус: Нельзя смешивать blocking и non-blocking.

// ПЛОХО: blocking call в Reactor
Flux<User> users = userService.getAllUsers();
users.map(user -> {
    // Это БЛОКИРУЕТ event loop!
    String enrichedData = blockingExternalAPI(user.getId());
    return new UserDTO(user, enrichedData);
}).subscribe(System.out::println);

// ХОРОШО: offload blocking на отдельный scheduler
Flux<User> users = userService.getAllUsers();
users.map(user -> new UserDTO(user))
    .publishOn(Schedulers.boundedElastic())  // Для blocking операций
    .flatMap(dto -> {
        String enrichedData = blockingExternalAPI(dto.getId());
        return Mono.just(new EnrichedDTO(dto, enrichedData));
    })
    .subscribeOn(Schedulers.parallel())
    .subscribe(System.out::println);

4. Трудно интегрировать с legacy code

Минус: Если есть блокирующие зависимости (JDBC и т.д.).

// Legacy repository (блокирующий)
public interface UserRepository extends JpaRepository<User, Long> {}

// Нельзя использовать в WebFlux напрямую
@Service
public class UserService {
    private final UserRepository repository;  // Blocking!
    
    public Flux<User> getAllUsers() {
        // Как вернуть Flux из blocking метода?
        return Flux.fromIterable(repository.findAll());  // Всё равно блокирует!
    }
}

// Нужно обёртка
@Service
public class UserService {
    private final UserRepository repository;
    
    public Flux<User> getAllUsers() {
        return Flux.defer(() -> 
            Flux.fromIterable(repository.findAll())
        ).subscribeOn(Schedulers.boundedElastic());  // Offload на отдельный поток
    }
}

5. Memory leaks если неправильно использовать

Минус: Подписки нужно отписывать.

// ПЛОХО: утечка памяти
Flux<DataChunk> stream = fetchDataStream();
stream.subscribe(chunk -> process(chunk));
// Кто отписывается? Никто! Memory leak!

// ХОРОШО: используй Disposable
Disposable subscription = stream.subscribe(chunk -> process(chunk));
// ...
subscription.dispose();  // Явная отписка

// ХОРОШО: используй try-with-resources (если поддерживается)
try (Disposable subscription = stream.subscribe(...)) {
    // ...
}  // автоматически dispose

6. Performance overhead для простых операций

Минус: Для простого кода может быть медленнее.

// Простой case (нет асинхронности)
@GetMapping("/users/{id}")
public Mono<User> getUser(@PathVariable Long id) {
    return userRepository.findById(id);
    // Reactor добавляет overhead для простой синхронной операции
}

// Это может быть медленнее традиционного
@GetMapping("/users/{id}")
public ResponseEntity<User> getUser(@PathVariable Long id) {
    return userRepository.findById(id)
        .map(ResponseEntity::ok)
        .orElseGet(() -> ResponseEntity.notFound().build());
}

7. Конкурирующие решения

Минус: Kotlin coroutines, Virtual Threads (Java 21) могут быть проще.

// Kotlin coroutines (проще!)
suspend fun getUser(id: Long): User {
    return userRepository.findByIdAsync(id)
}

// Java 21 Virtual Threads (ещё проще!)
var user = userRepository.findById(id).orElseThrow();
// Нет нужды в Reactor, виртуальные потоки очень лёгкие

Сравнение: Reactor vs традиционный Spring MVC

Best Practices для Reactor

✅ Делай так:

// 1. Используй Mono/Flux правильно
Mono<User> user = userRepository.findById(1L);
Flux<Order> orders = orderRepository.findAll();

// 2. Не вызывай block() в production
// User user = mono.block();  // NO!
Mono<User> monoUser = mono;  // YES

// 3. Обрабатывай ошибки
mono.onErrorResume(err -> Mono.just(defaultValue))
    .subscribe();

// 4. Используй backpressure
flux.buffer(100).subscribe();  // Не все в памяти

// 5. Offload blocking операции
flux.publishOn(Schedulers.boundedElastic())
    .flatMap(item -> blockingCall(item))
    .subscribe();

❌ Не делай так:

// 1. Не смешивай blocking и non-blocking
flux.map(item -> blockingCall(item));  // Замораживает event loop!

// 2. Не создавай новую Flux без причины
for (User user : users) {
    Flux.just(user).subscribe();  // Неправильно!
}

// 3. Не забывай отписываться
subscribe();  // Память может утечь

// 4. Не игнорируй ошибки
subscribe(data -> {}, err -> {});  // Хотя бы логируй

Когда использовать Reactor?

✅ Использовать:

• High-throughput API (10k+ requests/sec)
• Real-time приложения (WebSockets, SSE)
• Микросервисы с много асинхронных операций
• Stream processing больших данных

❌ Не использовать:

• CRUD приложения (простой JDBC sufficient)
• Legacy монолит
• Если весь стек blocking (JDBC, другие библиотеки)
• Если нет опыта с async кодом

Заключение

Spring Reactor - отличный инструмент для высоконагруженных систем, но с высокой ценой на обучение и сложность. Рекомендую использовать если:

1. Реально нужна масштабируемость
2. Есть асинхронные зависимости (API calls)
3. Команда готова учиться
4. Production имеет высокую нагрузку

Для простых CRUD приложений лучше придерживаться Spring MVC с JDBC или JPA!