Реактивное программирование в Java с Project Reactor и RxJava

// Блокирующий подход
String result = blockingService.getData();
System.out.println(result);
 
// Реактивный подход
reactiveService.getData()
    .subscribe(result -> System.out.println(result));
// Продолжаем выполнять другие операции, не дожидаясь результата

// Project Reactor пример
Flux<String> names = Flux.just("Алиса", "Боб", "Чарли");
Mono<String> singleName = Mono.just("Дэвид");

// RxJava пример
Observable<String> names = Observable.just("Алиса", "Боб", "Чарли");
Single<String> singleName = Single.just("Дэвид");
Maybe<String> maybeName = Maybe.just("Ева"); // может быть пустым
Completable operation = Completable.complete(); // только успех/ошибка

Flux.range(1, 100)
    .filter(n -> n % 2 == 0)    // четные числа
    .map(n -> n * n)            // возвести в квадрат
    .take(10)                   // взять первые 10
    .subscribe(System.out::println);

Observable.range(1, 100)
    .filter(n -> n % 2 == 0)
    .map(n -> n * n)
    .take(10)
    .subscribe(System.out::println);

// Project Reactor
Flux.range(1, 10).collectList().block(); // возвращает List<Integer>
 
// RxJava
Observable.range(1, 10).toList().blockingGet(); // возвращает List<Integer>

Flux.range(1, 1000)
    .publishOn(Schedulers.parallel())
    .map(i -> compute(i))
    .subscribeOn(Schedulers.elastic())
    .subscribe();

Observable.range(1, 1000)
    .observeOn(Schedulers.computation())
    .map(i -> compute(i))
    .subscribeOn(Schedulers.io())
    .subscribe();

// Project Reactor
Flux<String> names = Flux.just("Анна", "Борис", "Виктор");
Flux<Integer> ages = Flux.just(25, 30, 35);
 
Flux.zip(names, ages, (name, age) -> name + ": " + age + " лет")
    .subscribe(System.out::println);

// RxJava
Observable<String> names = Observable.just("Анна", "Борис", "Виктор");
Observable<Integer> ages = Observable.just(25, 30, 35);
 
Observable.zip(names, ages, (name, age) -> name + ": " + age + " лет")
    .subscribe(System.out::println);

// Project Reactor
Flux<String> interval1 = Flux.interval(Duration.ofMillis(100)).map(i -> "A" + i);
Flux<String> interval2 = Flux.interval(Duration.ofMillis(150)).map(i -> "B" + i);
 
Flux.combineLatest(interval1, interval2, (a, b) -> a + ":" + b)
    .take(5)
    .subscribe(System.out::println);

// RxJava
Observable<String> source1 = Observable.interval(100, TimeUnit.MILLISECONDS)
    .map(i -> "Источник 1: " + i).take(5);
Observable<String> source2 = Observable.interval(150, TimeUnit.MILLISECONDS)
    .map(i -> "Источник 2: " + i).take(5);
 
Observable.merge(source1, source2)
    .subscribe(System.out::println);

// Преобразование Flux в Publisher
Flux<String> flux = Flux.just("данные");
Publisher<String> publisher = flux;
 
// Преобразование Publisher в Flux
Flux<String> convertedBack = Flux.from(publisher);

// Преобразование Flowable в Publisher
Flowable<String> flowable = Flowable.just("данные");
Publisher<String> publisher = flowable;
 
// Преобразование Publisher в Flowable
Flowable<String> convertedBack = Flowable.fromPublisher(publisher);

StepVerifier.create(Flux.just("a", "b", "c"))
    .expectNext("a")
    .expectNext("b")
    .expectNext("c")
    .verifyComplete();

StepVerifier.withVirtualTime(() -> Flux.interval(Duration.ofHours(1)).take(2))
    .expectSubscription()
    .expectNoEvent(Duration.ofHours(1))
    .expectNext(0L)
    .thenAwait(Duration.ofHours(1))
    .expectNext(1L)
    .verifyComplete();

Observable<String> source = Observable.just("Alpha", "Beta", "Gamma");
TestObserver<String> testObserver = source.test();
 
testObserver.assertValues("Alpha", "Beta", "Gamma")
    .assertComplete()
    .assertNoErrors();

Flux.just("100", "200", "three", "400")
    .map(Integer::parseInt)
    .onErrorReturn(0)  // При ошибке вернет 0 и завершится
    .subscribe(System.out::println);
 
Flux.just("100", "200", "three", "400")
    .map(Integer::parseInt)
    .onErrorResume(e -> Flux.just(-1, -2))  // При ошибке продолжит с новыми элементами
    .subscribe(System.out::println);
 
Flux.just("100", "200", "three", "400")
    .flatMap(s -> Mono.fromCallable(() -> Integer.parseInt(s))
        .onErrorResume(e -> Mono.just(-1))  // Локальная обработка ошибок
    )
    .subscribe(System.out::println);

Observable.just("100", "200", "three", "400")
    .map(Integer::parseInt)
    .onErrorReturnItem(0)
    .subscribe(System.out::println);
 
Observable.just("100", "200", "three", "400")
    .map(Integer::parseInt)
    .onErrorResumeNext(Observable.just(-1, -2))
    .subscribe(System.out::println);

// Project Reactor с экспоненциальной задержкой
Flux.just("url1", "url2")
    .flatMap(url -> callExternalService(url))
    .retryBackoff(3, Duration.ofSeconds(1), Duration.ofSeconds(10))
    .subscribe(System.out::println);
 
// RxJava с простым повтором
Observable.just("url1", "url2")
    .flatMap(url -> callExternalService(url))
    .retry(3)
    .subscribe(System.out::println);

// Блокирующий подход: ~500ms (100мс на каждый запрос)
String result1 = service1.getData();  // 100ms
String result2 = service2.getData();  // 100ms
String result3 = service3.getData();  // 100ms
String result4 = service4.getData();  // 100ms
String result5 = service5.getData();  // 100ms
 
// Реактивный подход: ~100-120ms (все запросы параллельно)
Mono<String> result1 = service1.getDataReactive();
Mono<String> result2 = service2.getDataReactive();
Mono<String> result3 = service3.getDataReactive();
Mono<String> result4 = service4.getDataReactive();
Mono<String> result5 = service5.getDataReactive();
 
Mono.zip(result1, result2, result3, result4, result5)
.map(tuple -> combineResults(tuple.getT1(), tuple.getT2(), tuple.getT3(), tuple.getT4(), tuple.getT5()))
.subscribe(finalResult -> sendResponse(finalResult));

// Обработка потока событий от датчиков с агрегацией по 5-секундным интервалам
Flux.from(sensorEventSource) // источник событий
.window(Duration.ofSeconds(5))
.flatMap(window -> window.groupBy(SensorEvent::getSensorId)
    .flatMap(group -> group.reduce(SensorData::combine))
)
.subscribe(aggregatedData -> {
    analyzeAndStore(aggregatedData);
    sendAlertIfNeeded(aggregatedData);
});

// Пример механизма ограничения скорости обработки с динамическим буфером
Flux<ClientRequest> requestSource = ...;
 
requestSource
.onBackpressureBuffer(10000, BufferOverflowStrategy.DROP_OLDEST)
.concatMap(request -> processRequest(request)
    .timeout(Duration.ofSeconds(3), Mono.empty())
    .onErrorResume(e -> {
        logError(e);
        return Mono.empty();
    })
, 100) // Максимум 100 параллельных запросов
.subscribe();

// Добавление контекста к реактивному потоку для отладки
Mono.just(1)
.map(i -> i + 1)
.checkpoint("After first map") // добавляем метку
.map(i -> i * 2)
.checkpoint("After second map")
.subscribe();

// Неправильно: блокирующая операция внутри реактивного потока
Flux.range(1, 100)
.map(i -> {
    // Здесь блокируемся – плохая идея
    Thread.sleep(100);
    return i * 2;
})
.subscribe();
 
// Правильно: блокирующая операция выполняется в отдельном пуле потоков
Flux.range(1, 100)
.publishOn(Schedulers.boundedElastic()) // Специальный планировщик для блокирующих операций
.map(i -> {
    Thread.sleep(100);
    return i * 2;
})
.subscribe();

// Пример использования оператора timeout с резервной цепочкой
Mono<Response> result = callExternalService(request)
.timeout(Duration.ofSeconds(3), 
         Mono.just(new Response("Временный ответ пока сервис недоступен")))
.doOnError(e -> logWarning("Ошибка обработки запроса", e));

// Буфер с хронометражем: собирает максимум 100 событий или
// ждёт 500мс, что наступит раньше
eventStream
    .bufferTimeout(100, Duration.ofMillis(500))
    .flatMap(batch -> processBatchOfEvents(batch))
    .subscribe();

// Реализация предохранителя с Project Reactor
Flux<String> protectedCall = Flux.defer(() -> callToExternalService())
    .transform(CircuitBreakerOperator.of(circuitBreaker))
    .onErrorResume(CircuitBreakerOpenException.class, 
        ex -> Flux.just("Цепь разомкнута, используем резервный ответ"));

// Кэширование с автоматической инвалидацией
class ReactiveCachingService {
    private final Map<String, Mono<Data>> cache = new ConcurrentHashMap<>();
    private final Flux<CacheInvalidationEvent> invalidationEvents;
 
    public Mono<Data> getData(String key) {
        return Mono.defer(() -> cache.computeIfAbsent(key, 
            k -> fetchFromDatabase(k).cache()));
    }
 
    public ReactiveCachingService(Flux<CacheInvalidationEvent> events) {
        this.invalidationEvents = events;
        
        // Подписываемся на события инвалидации
        events.subscribe(event -> cache.remove(event.getKey()));
    }
}

// Адаптация блокирующего кода для реактивного использования
public class UserServiceAdapter {
    private final BlockingUserService legacyService;
    
    public Mono<User> findUser(long userId) {
        // Выполняем блокирующий вызов в специальном пуле потоков,
        // чтобы не блокировать события-луп
        return Mono.fromCallable(() -> legacyService.findUser(userId))
            .subscribeOn(Schedulers.boundedElastic());
    }
}

@RestController
public class ReactiveWrapperController {
    private final LegacyUserService userService;
    
    @GetMapping("/users")
    public Flux<User> getAllUsers() {
        // Оборачиваем блокирующий вызов, но уже предоставляем
        // реактивный API для клиентов
        return Flux.defer(() -> Flux.fromIterable(userService.getAllUsers()))
            .subscribeOn(Schedulers.boundedElastic());
    }
}

// R2DBC с PostgreSQL
ConnectionFactory connectionFactory = ConnectionFactories.get(
    "r2dbc:postgresql://localhost:5432/mydb"
);
 
Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
Flux<Map<String, Object>> resultFlux = connectionMono
    .flatMapMany(connection ->
        Flux.from(connection
            .createStatement("SELECT * FROM users WHERE age > $1")
            .bind("$1", 18)
            .execute())
    )
    .flatMap(result -> 
        Flux.from(result.map((row, metadata) -> {
            Map<String, Object> map = new HashMap<>();
            for (String column : metadata.getColumnNames()) {
                map.put(column, row.get(column));
            }
            return map;
        }))
    );

public interface UserRepository extends ReactiveCrudRepository<User, Long> {
    Flux<User> findByAgeGreaterThan(int age);
    Mono<User> findByUsername(String username);
}
 
// Использование
userRepository.findByAgeGreaterThan(18)
    .map(user -> convertToDto(user))
    .subscribe(userDto -> log.info("Found user: {}", userDto));

// Реактивный MongoDB драйвер
ReactiveMongoTemplate template;
 
Flux<Document> userDocuments = template.find(
    Query.query(Criteria.where("active").is(true)), 
    Document.class, 
    "users"
);
 
// Spring Data Reactive MongoDB
public interface ReactiveUserRepository extends ReactiveMongoRepository<User, String> {
    Flux<User> findByActiveTrue();
    Mono<Long> countByActiveTrue();
}

// Ограничение до максимум 50 параллельных запросов
Flux<User> batchedUserFlux = userIds
    .flatMap(id -> userRepository.findById(id), 50)
    .onBackpressureBuffer(1000, BufferOverflowStrategy.ERROR);

// Потенциальный гибридный подход с Loom и Reactor
public Mono<Result> processData(String input) {
    return Mono.fromCompletionStage(() -> {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            // Используем виртуальные потоки для параллелизма
            var subtask1 = scope.fork(() -> callService1(input));
            var subtask2 = scope.fork(() -> callService2(input));
            
            scope.join();           // Ждём завершения всех задач
            scope.throwIfFailed();  // Пробрасываем ошибку если что-то пошло не так
            
            // Объединяем результаты и возвращаем
            return combineResults(subtask1.get(), subtask2.get());
        }
    });
}

// Пример реактивного Spring Cloud Gateway
@Bean
public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
    return builder.routes()
        .route("path_route", r -> r.path("/api/customers/**")
            .filters(f -> f
                .circuitBreaker(c -> c.setName("customerCircuitBreaker")
                    .setFallbackUri("forward:/fallback"))
                .rateLimit(c -> c.setRateLimiter(redisRateLimiter())
                    .setKeyResolver(userKeyResolver()))
            )
            .uri("lb://customer-service"))
        .build();
}

// Реактивный контроллер на WebFlux
@RestController
@RequestMapping("/api/products")
public class ProductController {
    private final ProductService productService;
    
    @GetMapping
    public Flux<ProductDTO> getAllProducts() {
        return productService.findAllProducts()
            .map(this::convertToDto);
    }
    
    @GetMapping(value = "/{id}/updates", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ProductUpdate> streamUpdates(@PathVariable String id) {
        return productService.streamUpdatesForProduct(id)
            .delayElements(Duration.ofSeconds(1)); // имитация задержки обновлений
    }
 
    @PostMapping
    public Mono<ResponseEntity<ProductDTO>> createProduct(@RequestBody Mono<ProductDTO> productDto) {
        return productDto
            .map(this::convertToEntity)
            .flatMap(productService::save)
            .map(this::convertToDto)
            .map(dto -> ResponseEntity.created(URI.create("/api/products/" + dto.getId())).body(dto));
    }
}

// До: блокирующий сервис обработки платежей
public PaymentResult processPayment(Payment payment) {
    User user = userRepository.findById(payment.getUserId());
    Account account = accountRepository.findById(user.getAccountId());
    boolean valid = validationService.validatePayment(payment, account);
    if (valid) {
        TransactionResult result = paymentGateway.executeTransaction(payment);
        notificationService.sendReceipt(payment, result);
        return PaymentResult.success(result.getTransactionId());
    }
    return PaymentResult.error("Invalid payment");
}
 
// После: реактивный сервис обработки платежей
public Mono<PaymentResult> processPayment(Payment payment) {
    return userRepository.findById(payment.getUserId())
        .flatMap(user -> accountRepository.findById(user.getAccountId()))
        .flatMap(account -> validationService.validatePayment(payment, account)
            .flatMap(valid -> {
                if (valid) {
                    return paymentGateway.executeTransaction(payment)
                        .flatMap(result -> notificationService.sendReceipt(payment, result)
                            .thenReturn(PaymentResult.success(result.getTransactionId())));
                }
                return Mono.just(PaymentResult.error("Invalid payment"));
            })
        );
}