Реактивное программирование в Java: от теории к практике

// Традиционный подход может "зависнуть"
 public String loadUserDataBlocking(String userId) {
     // Может блокировать поток на неопределенное время
     return database.query("SELECT * FROM users WHERE id = ?", userId);
 }

// Реактивный подход гарантирует ответ
public Mono<String> loadUserDataReactive(String userId) {
    return userRepository.findById(userId) // userRepository должен быть реактивный
        .timeout(Duration.ofSeconds(3))      // Гарантия максимального времени ответа
        .onErrorReturn("Пользователь не доступен"); // Всегда возвращаем результат
}

public Mono<Order> processOrder(Order order) {
     return inventoryService.reserveItems(order) // если тут есть блокирующие вызовы то нужно добавлять .subscribeOn(Schedulers.boundedElastic()
         .transformDeferred(circuitBreaker::run) // Защита от повторяющихся сбоев
         .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))) // 3 попытки с растущей задержкой
         .timeout(Duration.ofSeconds(30))
         .onErrorResume(TimeoutException.class, e -> {
             // Переход на упрощенную логику при таймауте
             return processOrderWithLimitedFunctionality(order);
         })
         .onErrorResume(ServiceUnavailableException.class, e -> {
             return processOrderWithoutReservation(order);
         })
         .onErrorReturn(createFallbackOrder(order));
 }

@Bean
 public Scheduler elasticScheduler() {
     return Schedulers.newBoundedElastic(
         10,     //  лимит расширения при пиковой нагрузке
         1000,   //  буфер для задач при превышении лимита потоков
         "elastic-pool",
         60,    // Потоки, которые простаивают более 60 секунд, автоматически удаляются
         true);

}

public Flux<String> processBatchReactive(List<String> items) {
     return Flux.fromIterable(items)
         .flatMap(item -> 
                        Mono.fromCallable(() -> processItem(item)) // тут processItem - блокирующая операция, если тут будет реактивный метод, то subscribeOn не нужен
                                .subscribeOn(elasticScheduler())
                 // Если processItem выполняется дольше 30 секунд - операция прерывается
                 .timeout(Duration.ofSeconds(30))
                 // При ошибке будет 3 попытки с задержкой
                 .retryWhen(Retry.backoff(3, Duration.ofMillis(100)))
                 .name("processItem")
                 .metrics() // метрики нужно настраивать дополнительно
         , 
               10  // Максимум 10 одновременных операций processItem()
            )
         // Если вся обработка списка занимает больше 5 минут - прерываем
         .timeout(Duration.ofMinutes(5));
 }

// сервис для отправки заказов в брокер сообщений
@Service
public class ReactiveOrderService {
    private final StreamBridge streamBridge;

    public ReactiveOrderService(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    public Mono<Void> processOrder(Order order) {
    return Mono.fromCallable(() -> streamBridge.send("orders-out-0", order))
                  .subscribeOn(Schedulers.boundedElastic())
                  .doOnSuccess(sent -> log.info("✅ Заказ отправлен: {}", order.getId()))
                  .then();
}
// обработчик входящих сообщений
@Component
public class OrderMessageHandler {
    private final ReactiveOrderService orderService;

    public OrderMessageHandler(ReactiveOrderService orderService) {
        this.orderService = orderService;
    }

    @Bean
    public Consumer<Flux<Order>> orderProcessor() {
        return flux -> flux
            .flatMap(order -> orderService.processOrder(order));
    }
}

public class MonoUseCases {
     
     // поиск по ID - всегда один объект или null
     Mono<User> findUserById(String userId) {
         return userRepository.findById(userId); // предполагается, что userRepository реактивный
     }
     
     // создание ресурса - возвращаем созданный объект
     Mono<Order> createOrder(OrderRequest request) {
         return orderRepository.save(request.toOrder());
     }
     
     // аутентификация - возвращаем токен или ошибку
     Mono<AuthToken> login(String email, String password) {
         return authService.authenticate(email, password);
     }
     
     // валидация - успех или ошибка
     Mono<Void> validateEmail(String email) {
         return emailValidator.isValid(email) 
             ? Mono.empty() 
             : Mono.error(new InvalidEmailException());
     }
     
 }

  //  список элементов - много объектов
     Flux<Product> getAllProducts() {
         return productRepository.findAll();
     }
     
     // поиск с фильтрацией - несколько результатов
     Flux<User> findUsersByCity(String city) {
         return userRepository.findByCity(city);
     }
     
     // реальное время - поток событий
     @GetMapping(value = "/notifications", 
                 produces = MediaType.TEXT_EVENT_STREAM_VALUE)
     Flux<Notification> streamNotifications(String userId) {
         return notificationService.getUserNotifications(userId)
             .doOnCancel(() -> log.info("Клиент отключен")); // Обработка отключения
     }
     
     // IoT данные - непрерывный поток с датчиков
     Flux<SensorData> streamSensorData(String deviceId) {
     return sensorService.subscribeToDevice(deviceId)
         .sample(Duration.ofSeconds(1))     // Дросселирование - 1 значение в секунду
         .onBackpressureLatest()            // Только последнее значение при перегрузке
         .doOnSubscribe(sub -> log.info("Подписан на устройство: {}", deviceId))
         .doOnComplete(() -> log.info("Устройство {} стрим завершен", deviceId));
 }
     
     // аудит и логи - поток событий системы
     Flux<AuditEvent> getAuditLog(LocalDateTime from, LocalDateTime to) {
         return auditRepository.findByTimestampBetween(from, to); 

    }

//  блокирующий подход
 @GetMapping("/users/{id}")
 public User findUserById(String userId) {
     // Каждый запрос занимает один поток на всё время выполнения
     User user = userRepository.findById(userId); // Поток БЛОКИРОВАН на 200ms
     return user;
     // Поток освобождается только после полного выполнения
 }
 
 // ПРИ 1000 одновременных запросов:
 // 1000 потоков × 200ms = 200 секунд блокировки!
 // Сервер "захлёбывается" - кончаются потоки

// неблокирующий подход  
 @GetMapping("/users/{id}")
 public Mono<User> findUserById(String userId) {
     return userRepository.findById(userId); // реактивный userRepository
     // Поток НЕ блокируется - сразу освобождается!
     // Запрос "подписывается" на результат и ждёт его асинхронно
 }
 
 // ПРИ 1000 одновременных запросов:
 // 1 поток может обработать 10000+ таких запросов!
 // Сервер использует ресурсы эффективно

// Вся коллекция загружается в память сразу
 @GetMapping("/products")
 public List<Product> getAllProducts() {
     // Все продукты загружаются в память одновременно
     List<Product> products = productRepository.findAll(); 
     // Блокируем поток до полной загрузки всех данных
     return products;
     // При 1,000,000 товаров: 1GB памяти + блокировка на 2+ секунды
 }

//  Данные стримятся постепенно
 @GetMapping(value = "/products", produces = MediaType.APPLICATION_NDJSON_VALUE)
 public Flux<Product> getAllProductsReactive() {
     return productRepository.findAll(); реактивный productRepository
     // Данные отправляются по мере готовности
     // Поток освобождается сразу, клиент получает данные постепенно
     // При 1,000,000 товаров: 1MB буфер + неблокирующая обработка
 }

@GetMapping("/users/{id}")
public User getUser(String id) {
// Ошибка: блокируем поток до получения результата
    return userService.findById(id).block(); 
}

@GetMapping("/users/{id}")
 public Mono<User> getUser(String id) {

// Реактивно: поток сразу освобождается
     return userService.findById(id);
 }

@GetMapping("/orders")
 public void process() {
     orderService.getOrders()
         .subscribe(order -> log.info("Order: {}", order));
 }

@Bean
 @Scheduled(fixedRate = 5000)
 public Disposable backgroundCleanup() {
     // Самостоятельно создаем и управляем потоком
     return userService.cleanupExpiredSessions()
         .subscribeOn(Schedulers.boundedElastic())
         .subscribe(
             count -> log.info("Cleaned up {} sessions", count),
             error -> log.error("Cleanup failed", error)
         );
 }

 

@Component
 public class OrderMessageProcessor {
     private Disposable subscription;
     
     @EventListener(ApplicationReadyEvent.class)
     public void startProcessing() {
         // Самостоятельно создаем поток из Kafka
         this.subscription = kafkaReceiver.receive()
             .flatMap(record -> 
                 orderService.process(record.value())
                     .doOnSuccess(result -> record.receiverOffset().acknowledge())
                     .subscribeOn(Schedulers.boundedElastic())
             )
             .subscribe(
                 result -> log.debug("Processed order"),
                 error -> log.error("Order processing failed", error)
             );
     }
     
     @PreDestroy
     public void stopProcessing() {
         if (subscription != null) {
             subscription.dispose();
         }
     }
 }

public Flux<User> findAllUsers() {
     return Flux.fromIterable(userRepository.findAll()); // блокирующий JPA вызов
 }

public Flux<User> findAllUsers() {
     return Mono.fromCallable(() -> userRepository.findAll()) // Безопасный вызов
         .subscribeOn(Schedulers.boundedElastic())           // Вынос в отдельный пул потоков
         .flatMapMany(Flux::fromIterable);                   // Преобразование в поток
 }

Flux.interval(Duration.ofMillis(1))
     .map(this::process)
     .subscribe(); // Поток быстро переполнится

Flux.interval(Duration.ofMillis(1))
     .onBackpressureLatest()              // ограничиваем поток
     .flatMap(this::process)
     .subscribe();

@RestController
 @RequiredArgsConstructor
 @Slf4j
 public class OrderController {
 
     private final ClientOrderFacade clientOrderFacade;
     
     @GetMapping
     public List<Order> getAllOrders(String personId) {
         return clientOrderFacade.getAllOrders(personId);
     }
 }

@Component
 @RequiredArgsConstructor
 @Slf4j
 public class ClientOrderFacade {
 private final OrderService orderService;

private final PersonService personService;
 
     @Transactional
     public List<Order> getAllOrders(String personId) {
         Person person = personService.findById(personId);
         return orderService.findAllByPerson(person)
             .stream()
             .sorted(Comparator.comparing(Order::getCreated).reversed())
             .toList();
     }
 }

@Service
 @RequiredArgsConstructor
 @Slf4j
 public class OrderServiceImpl implements OrderService {
     private final OrderRepository orderRepository;
     
     @Override
     public Set<Order> findAllByPerson(Person person) {
         return orderRepository.findAllByPerson(person);
     }
 }

@Repository
 public interface OrderRepository extends JpaRepository<Order, Long> {
     Set<Order> findAllByPerson(Person person);

<!-- Spring WebFlux -->
 <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-webflux</artifactId>
 </dependency>
 
 <!-- Для реактивного доступа к БД -->
 <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-data-r2dbc</artifactId>
 </dependency>
 
 <!-- Драйвер для PostgreSQL -->
 <dependency>
     <groupId>io.r2dbc</groupId>
     <artifactId>r2dbc-postgresql</artifactId>
 </dependency>

@RestController
 @RequiredArgsConstructor
 @Slf4j
 public class OrderController {
 
     private final ClientOrderFacade clientOrderFacade;
 
     @GetMapping
     public Flux<Order> getAllOrders(@RequestParam String personId) {
         log.info("Поиск заказов пользователя {}", personId );
         return clientOrderFacade.getAllOrders(personId);
     }
 }

@Component
 @RequiredArgsConstructor
 @Slf4j
 public class ClientOrderFacade {
 
     private final OrderService orderService;
     private final PersonService personService;
 
     public Flux<Order> getAllOrders(String personId) {
         return personService.findById(personId)
             .switchIfEmpty(Mono.error(new RuntimeException("Пользователь не найден")))
             .flatMapMany(person -> orderService.findAllByPerson(person.getId()));
     }
 }

 

@Service
 @RequiredArgsConstructor
 @Slf4j
 public class OrderService {
 
     private final OrderRepository orderRepository;
 
     public Flux<Order> findAllByPerson(String personId) {
         return orderRepository.findAllByPersonId(personId);
     }
 }

@Repository
 public interface OrderRepository extends R2dbcRepository<Order, Long> {
 
     @Query("SELECT * FROM orders WHERE person_id = :personId ORDER BY created DESC")
     Flux<Order> findAllByPersonId(String personId);

}