Реактивное программировани­е с Kafka Stream и Spring WebFlux

// Пример реактивного кода с использованием Flux
Flux.just("Привет", "Реактивный", "Мир")
    .map(String::toUpperCase)
    .filter(s -> s.length() > 5)
    .subscribe(
        data -> System.out.println("Получено: " + data),
        error -> System.err.println("Ошибка: " + error),
        () -> System.out.println("Завершено")
    );

// Пример простого Producer на Java
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "message-key", "message-value"));
producer.close();

// Пример контроллера в Spring WebFlux
@RestController
public class UserController {
    
    private final UserRepository userRepository;
    
    public UserController(UserRepository userRepository) {
        this.userRepository = userRepository;
    }
    
    @GetMapping("/users")
    public Flux<User> getAllUsers() {
        return userRepository.findAll();
    }
    
    @GetMapping("/users/{id}")
    public Mono<User> getUserById(@PathVariable String id) {
        return userRepository.findById(id);
    }
}

// Функциональный стиль в Spring WebFlux
@Configuration
public class RouterConfig {
    
    @Bean
    public RouterFunction<ServerResponse> route(UserHandler userHandler) {
        return RouterFunctions
            .route(GET("/users").and(accept(APPLICATION_JSON)), userHandler::getAllUsers)
            .andRoute(GET("/users/{id}").and(accept(APPLICATION_JSON)), userHandler::getUserById);
    }
}

@Configuration
public class KafkaProducerConfig {
    
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    
    @Bean
    public SenderOptions<String, String> senderOptions() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return SenderOptions.create(props);
    }
    
    @Bean
    public ReactiveKafkaProducerTemplate<String, String> reactiveKafkaProducerTemplate(
            SenderOptions<String, String> senderOptions) {
        return new ReactiveKafkaProducerTemplate<>(senderOptions);
    }
}

@Configuration
public class KafkaConsumerConfig {
    
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    
    @Bean
    public ReceiverOptions<String, String> receiverOptions() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return ReceiverOptions.create(props);
    }
    
    @Bean
    public ReactiveKafkaConsumerTemplate<String, String> reactiveKafkaConsumerTemplate(
            ReceiverOptions<String, String> receiverOptions) {
        return new ReactiveKafkaConsumerTemplate<>(receiverOptions);
    }
}

// Асинхронная отправка сообщения в Kafka
reactiveKafkaProducerTemplate.send("topic-name", "payload")
    .doOnSuccess(result -> log.info("Сообщение успешно отправлено"))
    .subscribe();

// Реализация обратного давления в реактивном потребителе Kafka
return reactiveKafkaConsumerTemplate
    .receiveAutoAck()
    .map(ConsumerRecord::value)
    .onBackpressureBuffer(1000) // Буферизация до 1000 сообщений
    .doOnNext(this::processMessage);

// Создание из одного или нескольких элементов
Flux.just("A", "B", "C");
 
// Создание из коллекции
Flux.fromIterable(Arrays.asList("X", "Y", "Z"));
 
// Создание пустой последовательности
Mono.empty();
 
// Создание последовательности, завершающейся ошибкой
Flux.error(new RuntimeException("Что-то пошло не так"));
 
// Создание с отложенным вычислением (ленивая инициализация)
Mono.defer(() -> callExpensiveOperation());
 
// Генерация последовательности с помощью функции
Flux.generate(sink -> {
    sink.next(ThreadLocalRandom.current().nextInt());
    if (shouldStop()) sink.complete();
});

// Преобразование каждого элемента
flux.map(item -> item.toUpperCase());
 
// Плоское отображение (для работы с вложенными потоками)
flux.flatMap(item -> callAsyncService(item));
 
// Преобразование Flux в Mono, содержащий список всех элементов
flux.collectList();
 
// Преобразование элементов в пары с индексами
flux.index();
 
// Группировка элементов
flux.buffer(5); // Группы по 5 элементов
 
// Объединение последовательных дублирующихся элементов
flux.distinctUntilChanged();

// Отбор по предикату
flux.filter(item -> item.length() > 3);
 
// Ограничение количества элементов
flux.take(5);
 
// Пропуск определённого количества элементов
flux.skip(2);
 
// Первый элемент
flux.next();
 
// Отбор уникальных элементов
flux.distinct();
 
// Выборка элементов, пока выполняется условие
flux.takeWhile(item -> item.startsWith("A"));

// Последовательное объединение
Flux.concat(flux1, flux2);
 
// Параллельное объединение (элементы смешиваются в порядке эмиссии)
Flux.merge(flux1, flux2);
 
// Объединение по принципу "молнии" (чередование элементов)
flux1.zipWith(flux2, (item1, item2) -> item1 + "-" + item2);
 
// Комбинирование последнего элемента одного потока с другим потоком
flux1.withLatestFrom(flux2, (item1, item2) -> item1 + ":" + item2);

// Перехват и обработка ошибок
flux.onErrorReturn("Значение по умолчанию");
 
// Замена потока с ошибкой на другой поток
flux.onErrorResume(e -> Flux.just("Резервные данные"));
 
// Повторная попытка при ошибке
flux.retry(3); // 3 попытки
 
// Более гибкая стратегия повторных попыток с экспоненциальной задержкой
flux.retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
     .maxBackoff(Duration.ofSeconds(10)));

// Выполнение действия для каждого элемента
flux.doOnNext(item -> log.info("Обработка элемента: {}", item));
 
// Выполнение действия при завершении потока
flux.doOnComplete(() -> log.info("Поток завершен"));
 
// Выполнение действия при ошибке
flux.doOnError(e -> log.error("Произошла ошибка: {}", e.getMessage()));
 
// Выполнение действия перед подпиской
flux.doOnSubscribe(s -> log.info("Новая подписка"));
 
// Выполнение действия для всех типов событий (next/complete/error)
flux.doOnEach(signal -> log.debug("Сигнал: {}", signal));
 
// Выполнение действия при отмене подписки
flux.doOnCancel(() -> log.warn("Подписка отменена"));

// Потенциально опасное использование без ограничения параллелизма
flux.flatMap(item -> callExpensiveService(item));
 
// Более безопасное использование с ограничением параллелизма
flux.flatMap(item -> callExpensiveService(item), 10);

// Блокирующий подход
String result = blockingService.getData(); // Поток блокируется здесь
processResult(result);
 
// Неблокирующий подход
nonBlockingService.getData()
    .subscribe(result -> processResult(result)); // Поток не блокируется
// Код продолжает выполняться сразу

package com.example.reactiveapp.model;
 
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
 
import java.math.BigDecimal;
import java.time.LocalDateTime;
 
@Data
@NoArgsConstructor
@AllArgsConstructor
public class ProductEvent {
    private String id;
    private String name;
    private String category;
    private BigDecimal price;
    private EventType eventType;
    private LocalDateTime timestamp;
    
    public enum EventType {
        CREATED, UPDATED, DELETED
    }
}

package com.example.reactiveapp.config;
 
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate;
import reactor.kafka.sender.SenderOptions;
 
import java.util.HashMap;
import java.util.Map;
 
@Configuration
public class KafkaProducerConfig {
 
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
 
    @Bean
    public SenderOptions<String, String> senderOptions() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.ACKS_CONFIG, "all"); // Наиболее надежная конфигурация
        props.put(ProducerConfig.RETRIES_CONFIG, 3);  // Количество повторных попыток
        
        return SenderOptions.create(props);
    }
 
    @Bean
    public ReactiveKafkaProducerTemplate<String, String> reactiveKafkaProducerTemplate(
            SenderOptions<String, String> senderOptions) {
        return new ReactiveKafkaProducerTemplate<>(senderOptions);
    }
}

package com.example.reactiveapp.config;
 
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import reactor.kafka.receiver.ReceiverOptions;
 
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
 
@Configuration
public class KafkaConsumerConfig {
 
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;
    
    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;
 
    @Bean
    public ReceiverOptions<String, String> receiverOptions() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        
        return ReceiverOptions.create(props);
    }
 
    @Bean
    public ReceiverOptions<String, String> productEventsReceiverOptions(ReceiverOptions<String, String> receiverOptions) {
        return receiverOptions.subscription(Collections.singleton("product-events"));
    }
 
    @Bean
    public ReactiveKafkaConsumerTemplate<String, String> reactiveKafkaConsumerTemplate(
            ReceiverOptions<String, String> productEventsReceiverOptions) {
        return new ReactiveKafkaConsumerTemplate<>(productEventsReceiverOptions);
    }
}

package com.example.reactiveapp.service;
 
import com.example.reactiveapp.model.ProductEvent;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.kafka.sender.SenderResult;
 
@Service
@Slf4j
public class ProductEventProducer {
 
    private final ReactiveKafkaProducerTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;
    
    private static final String TOPIC = "product-events";
 
    public ProductEventProducer(ReactiveKafkaProducerTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
        this.objectMapper = new ObjectMapper();
        this.objectMapper.registerModule(new JavaTimeModule());
    }
 
    public Mono<SenderResult<Void>> sendProductEvent(ProductEvent event) {
        return Mono.fromCallable(() -> objectMapper.writeValueAsString(event))
            .flatMap(eventJson -> 
                kafkaTemplate.send(TOPIC, event.getId(), eventJson)
                    .doOnSuccess(r -> log.info("Sent event: {} with offset: {}", 
                        event, r.recordMetadata().offset()))
                    .doOnError(e -> log.error("Error sending event: {}", event, e))
            );
    }
}

package com.example.reactiveapp.service;
 
import com.example.reactiveapp.model.ProductEvent;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
 
import javax.annotation.PostConstruct;
 
@Service
@Slf4j
public class ProductEventConsumer {
 
    private final ReactiveKafkaConsumerTemplate<String, String> consumerTemplate;
    private final ObjectMapper objectMapper;
    private final Sinks.Many<ProductEvent> eventSink;
    
    public ProductEventConsumer(ReactiveKafkaConsumerTemplate<String, String> consumerTemplate) {
        this.consumerTemplate = consumerTemplate;
        this.objectMapper = new ObjectMapper();
        this.objectMapper.registerModule(new JavaTimeModule());
        this.eventSink = Sinks.many().multicast().onBackpressureBuffer();
    }
 
    @PostConstruct
    public void consumeEvents() {
        consumerTemplate.receiveAutoAck()
            .doOnNext(record -> log.info("Received key={}, value={} from topic={}, partition={}, offset={}",
                record.key(), record.value(), record.topic(), record.partition(), record.offset()))
            .flatMap(this::handleEvent)
            .subscribe();
    }
    
    private Mono<ProductEvent> handleEvent(ConsumerRecord<String, String> record) {
        return Mono.fromCallable(() -> {
                ProductEvent event = objectMapper.readValue(record.value(), ProductEvent.class);
                log.info("Processing event: {}", event);
                eventSink.tryEmitNext(event);
                return event;
            })
            .doOnError(e -> log.error("Error processing event from record: {}", record.value(), e))
            .onErrorResume(e -> Mono.empty())
            .subscribeOn(Schedulers.boundedElastic());
    }
    
    public Flux<ProductEvent> getEventStream() {
        return eventSink.asFlux();
    }
}

package com.example.reactiveapp.controller;
 
import com.example.reactiveapp.model.ProductEvent;
import com.example.reactiveapp.service.ProductEventConsumer;
import com.example.reactiveapp.service.ProductEventProducer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
 
import java.time.LocalDateTime;
 
@RestController
@RequestMapping("/api/products")
@RequiredArgsConstructor
@Slf4j
public class ProductController {
 
    private final ProductEventProducer producer;
    private final ProductEventConsumer consumer;
 
    @PostMapping("/events")
    public Mono<ResponseEntity<Void>> publishEvent(@RequestBody ProductEvent event) {
        event.setTimestamp(LocalDateTime.now());
        
        return producer.sendProductEvent(event)
            .thenReturn(ResponseEntity.accepted().build())
            .doOnError(e -> log.error("Failed to send event", e))
            .onErrorReturn(ResponseEntity.internalServerError().build());
    }
 
    @GetMapping(value = "/events/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ProductEvent> streamEvents() {
        return consumer.getEventStream();
    }
}

package com.example.reactiveapp.config;
 
import com.example.reactiveapp.model.ProductEvent;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class KafkaSerializationConfig {
 
    @Bean
    public ObjectMapper objectMapper() {
        ObjectMapper mapper = new ObjectMapper();
        mapper.registerModule(new JavaTimeModule());
        return mapper;
    }
 
    @Bean
    public Serializer<ProductEvent> productEventSerializer(ObjectMapper objectMapper) {
        return new Serializer<ProductEvent>() {
            @Override
            public byte[] serialize(String topic, ProductEvent data) {
                try {
                    return objectMapper.writeValueAsBytes(data);
                } catch (Exception e) {
                    throw new RuntimeException("Error serializing ProductEvent", e);
                }
            }
        };
    }
 
    @Bean
    public Deserializer<ProductEvent> productEventDeserializer(ObjectMapper objectMapper) {
        return new Deserializer<ProductEvent>() {
            @Override
            public ProductEvent deserialize(String topic, byte[] data) {
                try {
                    return objectMapper.readValue(data, ProductEvent.class);
                } catch (Exception e) {
                    throw new RuntimeException("Error deserializing ProductEvent", e);
                }
            }
        };
    }
}

package com.example.reactiveapp;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
public class ReactiveKafkaApplication {
 
    public static void main(String[] args) {
        SpringApplication.run(ReactiveKafkaApplication.class, args);
    }
}

package com.example.reactiveapp.service;
 
import com.example.reactiveapp.model.ProductEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;
 
import java.math.BigDecimal;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
 
@Service
@Slf4j
public class ProductAnalyticsService {
 
    private final ProductEventConsumer eventConsumer;
    private final Map<String, AtomicReference<BigDecimal>> categoryAvgPriceMap = new ConcurrentHashMap<>();
    private final Sinks.Many<Tuple2<String, BigDecimal>> priceUpdates = Sinks.many().multicast().onBackpressureBuffer();
 
    public ProductAnalyticsService(ProductEventConsumer eventConsumer) {
        this.eventConsumer = eventConsumer;
        
        // Подписка на поток событий для расчета статистики
        eventConsumer.getEventStream()
            .filter(event -> event.getEventType() != ProductEvent.EventType.DELETED)
            .groupBy(ProductEvent::getCategory)
            .flatMap(this::processCategoryEvents)
            .subscribe();
    }
    
    private Flux<Tuple2<String, BigDecimal>> processCategoryEvents(Flux<ProductEvent> categoryEvents) {
        String category = categoryEvents.blockFirst().getCategory();
        
        return categoryEvents
            .window(Duration.ofSeconds(5))
            .flatMap(window -> window.collectList()
                .filter(list -> !list.isEmpty())
                .map(events -> {
                    BigDecimal sum = events.stream()
                        .map(ProductEvent::getPrice)
                        .reduce(BigDecimal.ZERO, BigDecimal::add);
                    
                    BigDecimal avg = sum.divide(BigDecimal.valueOf(events.size()), 2, BigDecimal.ROUND_HALF_UP);
                    
                    // Обновляем значение в мэпе и публикуем обновление
                    categoryAvgPriceMap.computeIfAbsent(category, k -> new AtomicReference<>(avg))
                        .set(avg);
                    
                    log.info("Average price for category {}: {}", category, avg);
                    
                    Tuple2<String, BigDecimal> update = Tuples.of(category, avg);
                    priceUpdates.tryEmitNext(update);
                    
                    return update;
                })
            );
    }
    
    public Flux<Tuple2<String, BigDecimal>> getCategoryPriceUpdates() {
        return priceUpdates.asFlux();
    }
    
    public Mono<Map<String, BigDecimal>> getCurrentCategoryPrices() {
        Map<String, BigDecimal> result = new HashMap<>();
        categoryAvgPriceMap.forEach((category, price) -> 
            result.put(category, price.get()));
        return Mono.just(result);
    }
    
    // Получение количества событий по категориям за последние N секунд
    public Flux<Map<String, Long>> getEventCountByCategory(int seconds) {
        return eventConsumer.getEventStream()
            .window(Duration.ofSeconds(seconds), Duration.ofSeconds(seconds))
            .flatMap(window -> window
                .groupBy(ProductEvent::getCategory)
                .flatMap(group -> group.count().map(count -> 
                    Tuples.of(group.key(), count)))
                .collectMap(Tuple2::getT1, Tuple2::getT2)
            );
    }
}

@RestController
@RequestMapping("/api/analytics")
@RequiredArgsConstructor
@Slf4j
public class AnalyticsController {
 
    private final ProductAnalyticsService analyticsService;
    
    @GetMapping(value = "/prices/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Tuple2<String, BigDecimal>> streamPriceUpdates() {
        return analyticsService.getCategoryPriceUpdates();
    }
    
    @GetMapping("/prices/current")
    public Mono<Map<String, BigDecimal>> getCurrentPrices() {
        return analyticsService.getCurrentCategoryPrices();
    }
    
    @GetMapping(value = "/events/count", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Map<String, Long>> getEventCounts(@RequestParam(defaultValue = "10") int window) {
        return analyticsService.getEventCountByCategory(window);
    }
}

package com.example.reactiveapp.service;
 
import com.example.reactiveapp.model.ProductDetail;
import com.example.reactiveapp.model.ProductEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
 
import java.time.Duration;
 
@Service
@Slf4j
public class ProductEnrichmentService {
 
    private final WebClient webClient;
    private final ProductEventConsumer eventConsumer;
    
    public ProductEnrichmentService(ProductEventConsumer eventConsumer, WebClient.Builder webClientBuilder) {
        this.eventConsumer = eventConsumer;
        this.webClient = webClientBuilder.baseUrl("https://api.external-service.com").build();
        
        // Подписываемся на события и обогащаем их
        this.eventConsumer.getEventStream()
            .filter(event -> event.getEventType() == ProductEvent.EventType.CREATED)
            .flatMap(this::enrichProduct)
            .subscribe(
                enriched -> log.info("Обогащены данные продукта: {}", enriched),
                error -> log.error("Ошибка при обогащении данных", error)
            );
    }
    
    private Mono<ProductDetail> enrichProduct(ProductEvent event) {
        return webClient.get()
            .uri("/products/{id}/details", event.getId())
            .retrieve()
            .bodyToMono(ProductDetail.class)
            .doOnNext(detail -> log.info("Получены детали для продукта {}: {}", event.getId(), detail))
            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
                .maxBackoff(Duration.ofSeconds(5)))
            .onErrorResume(e -> {
                log.error("Не удалось получить детали продукта {}: {}", event.getId(), e.getMessage());
                return Mono.empty();
            });
    }
}

package com.example.reactiveapp.service;
 
import com.example.reactiveapp.model.ProductEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
 
import javax.annotation.PostConstruct;
import java.util.function.Predicate;
 
@Service
@Slf4j
public class EventRouter {
 
    private final ProductEventConsumer eventConsumer;
    private final Sinks.Many<ProductEvent> createdEvents = Sinks.many().multicast().onBackpressureBuffer();
    private final Sinks.Many<ProductEvent> updatedEvents = Sinks.many().multicast().onBackpressureBuffer();
    private final Sinks.Many<ProductEvent> deletedEvents = Sinks.many().multicast().onBackpressureBuffer();
    
    public EventRouter(ProductEventConsumer eventConsumer) {
        this.eventConsumer = eventConsumer;
    }
    
    @PostConstruct
    public void setupRouting() {
        // Маршрутизация по типу события
        eventConsumer.getEventStream()
            .doOnNext(event -> routeEvent(event))
            .subscribe();
    }
    
    private void routeEvent(ProductEvent event) {
        switch (event.getEventType()) {
            case CREATED:
                createdEvents.tryEmitNext(event);
                break;
            case UPDATED:
                updatedEvents.tryEmitNext(event);
                break;
            case DELETED:
                deletedEvents.tryEmitNext(event);
                break;
        }
    }
    
    public Flux<ProductEvent> getCreatedEvents() {
        return createdEvents.asFlux();
    }
    
    public Flux<ProductEvent> getUpdatedEvents() {
        return updatedEvents.asFlux();
    }
    
    public Flux<ProductEvent> getDeletedEvents() {
        return deletedEvents.asFlux();
    }
    
    // Позволяет получать поток событий по произвольному предикату
    public Flux<ProductEvent> getEventsByPredicate(Predicate<ProductEvent> predicate) {
        return eventConsumer.getEventStream().filter(predicate);
    }
}

productService.getProduct(id)
    .onErrorReturn(ProductNotFoundException.class, Product.notFound())
    .onErrorResume(DatabaseException.class, e -> 
        backupProductService.getProduct(id))
    .doOnError(e -> log.error("Ошибка при получении продукта: {}", id, e))
    .retry(3)
    .timeout(Duration.ofSeconds(5));

@Bean
public Flux<ReceiverRecord<String, String>> kafkaEvents(
        ReactiveKafkaConsumerTemplate<String, String> template) {
    return template.receive()
        .doOnNext(record -> log.info("Получено: {}", record))
        .flatMap(record -> processRecord(record)
            .doOnSuccess(v -> record.receiverOffset().acknowledge())
            .onErrorResume(e -> {
                log.error("Ошибка обработки: {}", record.value(), e);
                return sendToDlq(record)
                    .then(Mono.fromRunnable(() -> 
                        record.receiverOffset().acknowledge()));
            })
        )
        .onErrorContinue((e, o) -> 
            log.error("Неперехваченная ошибка: {}", o, e));
}
 
private Mono<Void> sendToDlq(ReceiverRecord<String, String> record) {
    return reactiveKafkaProducerTemplate
        .send("dlq-topic", record.key(), record.value())
        .then();
}

// Потенциально опасный код без ограничения параллелизма
flux.flatMap(item -> heavyProcessing(item));
 
// Контролируемый параллелизм
flux.flatMap(item -> heavyProcessing(item), 10);

// Создание планировщика с ограниченным пулом потоков
Scheduler boundedScheduler = Schedulers.newBoundedElastic(
    10, // максимум 10 потоков
    100, // максимум 100 задач в очереди
    "bounded-elastic",
    60 // TTL потока в секундах
);
 
// Использование планировщика для параллельной обработки
flux.flatMap(item -> 
    Mono.fromCallable(() -> heavyProcessing(item))
        .subscribeOn(boundedScheduler)
);

// Оптимизация продюсера
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
props.put(ProducerConfig.LINGER_MS_CONFIG, 5); // Ждать 5мс для группировки сообщений
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy"); // Сжатие данных
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); // 32MB буфер памяти
 
// Оптимизация консьюмера
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024); // Минимум 1KB данных за запрос
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500); // Но не ждать дольше 500мс
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500); // Получать не более 500 записей

// Неоптимально - создает отдельный Mono для каждого элемента
flux.flatMap(item -> validateAndProcess(item));
 
// Лучше - группирует элементы для пакетной обработки
flux.buffer(100)
    .flatMap(batch -> validateAndProcessBatch(batch));
 
// Еще лучше для подходящих операций - обработка целыми партициями
flux.window(Duration.ofSeconds(1), 1000)
    .flatMap(window -> window.collectList().flatMap(this::processBatch));

@Test
void testProductEventProcessing() {
    // Подготовка тестовых данных
    ProductEvent event = new ProductEvent("1", "Test", "Category", 
        BigDecimal.TEN, ProductEvent.EventType.CREATED, LocalDateTime.now());
    
    // Имитация отправки в Kafka
    when(kafkaTemplate.send(anyString(), anyString(), anyString()))
        .thenReturn(Mono.empty());
    
    // Проверка результата с помощью StepVerifier
    StepVerifier.create(productService.processEvent(event))
        .expectNextMatches(result -> result.isSuccess())
        .verifyComplete();
    
    // Верификация вызова Kafka
    verify(kafkaTemplate).send(eq("product-events"), eq(event.getId()), anyString());
}

@TestConfiguration
public class KafkaTestConfig {
    
    @Bean
    public EmbeddedKafkaBroker embeddedKafka() {
        return new EmbeddedKafkaBroker(1)
            .kafkaPorts(9092)
            .brokerProperties(Map.of(
                "auto.create.topics.enable", "true",
                "group.initial.rebalance.delay.ms", "0"
            ));
    }
    
    @Bean
    public ReactiveKafkaProducerTemplate<String, String> testProducerTemplate(
            EmbeddedKafkaBroker broker) {
        Map<String, Object> props = KafkaTestUtils.producerProps(broker);
        return new ReactiveKafkaProducerTemplate<>(
            SenderOptions.create(props)
        );
    }
}

// Добавление метрик в поток
flux.name("process-events")
    .metrics()
    .doOnEach(signal -> 
        registry.counter("events.processed").increment());

// Настройка в application.yml
management:
  endpoints:
    web:
      exposure:
        include: health,info,prometheus,metrics
  metrics:
    export:
      prometheus:
        enabled: true

// Включение в main() или через JVM параметры
Hooks.onOperatorDebug();
 
// Или более селективно
Hooks.onOperatorDebug(operatorDebugPredicate);

// Буферизация избыточных данных с риском OutOfMemoryError при слишком большом потоке
flux.onBackpressureBuffer();
 
// Буферизация с ограниченным размером и стратегией переполнения
flux.onBackpressureBuffer(
    10000, // максимальный размер буфера
    () -> log.warn("Буфер переполнен!"),
    BufferOverflowStrategy.DROP_OLDEST // отбрасывать старые элементы
);
 
// Пропуск элементов, когда потребитель не успевает обрабатывать
flux.onBackpressureDrop(droppedItem -> 
    log.warn("Элемент отброшен: {}", droppedItem));
 
// Сохранение только последнего элемента и сброс предыдущих
flux.onBackpressureLatest();

// Получение сообщений с ручным управлением смещениями
receiverOptions
    .subscription(Collections.singleton("topic"))
    .addAssignListener(partitions -> log.info("Partitions assigned: {}", partitions))
    .commitInterval(Duration.ZERO) // Отключаем автоматический коммит по времени
    .commitBatchSize(0); // Отключаем автоматический коммит по количеству
 
reactiveKafkaConsumerTemplate
    .receive()
    .concatMap(record -> {
        return processRecord(record)
            .then(Mono.fromRunnable(() -> {
                // Коммит смещения только после успешной обработки
                record.receiverOffset().acknowledge();
            }))
            .onErrorResume(e -> {
                log.error("Ошибка обработки сообщения", e);
                // Решаем, коммитить ли смещение при ошибке 
                // в зависимости от стратегии обработки ошибок
                if (shouldCommitOnError(e)) {
                    record.receiverOffset().acknowledge();
                }
                return Mono.empty();
            });
    })
    .subscribe();

reactiveKafkaConsumerTemplate
    .receive()
    .bufferTimeout(100, Duration.ofMillis(500)) // Группируем по 100 записей или ждем 500мс
    .flatMap(records -> {
        List<String> payloads = records.stream()
            .map(ReceiverRecord::value)
            .collect(Collectors.toList());
        
        return processBatch(payloads)
            .then(Mono.fromRunnable(() -> {
                // Коммитим последнее смещение в пакете после обработки всего пакета
                records.get(records.size() - 1).receiverOffset().acknowledge();
            }));
    }, 4) // Параллельно обрабатываем максимум 4 пакета
    .subscribe();

@Configuration
public class MetricsConfig {
    
    @Autowired
    private MeterRegistry meterRegistry;
    
    @Bean
    public WebClient webClient() {
        return WebClient.builder()
            .filter(WebClientMetricsFilter
                .named("external-service-client")
                .uriMapper(request -> request.url().getPath())
                .enable(meterRegistry))
            .baseUrl("https://api.external-service.com")
            .build();
    }
    
    @Bean
    public RouterFunction<ServerResponse> metricsRoute() {
        return route(GET("/actuator/metrics/custom"), 
            request -> ServerResponse.ok().body(
                Flux.interval(Duration.ofSeconds(1))
                    .name("custom-stream")
                    .tag("component", "demo")
                    .metrics()
                    .map(i -> "Metric: " + i), 
                String.class
            )
        );
    }
}

@Service
@Slf4j
public class MonitoredKafkaService {
    
    private final ReactiveKafkaConsumerTemplate<String, String> kafkaConsumer;
    private final MeterRegistry registry;
    private final Counter processedMessages;
    private final Counter failedMessages;
    private final Timer processingTime;
    
    public MonitoredKafkaService(
            ReactiveKafkaConsumerTemplate<String, String> kafkaConsumer,
            MeterRegistry registry) {
        this.kafkaConsumer = kafkaConsumer;
        this.registry = registry;
        
        // Создание счетчиков и таймеров
        this.processedMessages = registry.counter("kafka.messages.processed");
        this.failedMessages = registry.counter("kafka.messages.failed");
        this.processingTime = registry.timer("kafka.message.processing.time");
    }
    
    @PostConstruct
    public void startConsumer() {
        kafkaConsumer.receiveAutoAck()
            .doOnNext(record -> log.debug("Получено сообщение: {}", record.key()))
            .flatMap(record -> {
                // Измеряем время обработки каждого сообщения
                return Mono.just(record)
                    .flatMap(this::processRecord)
                    .name("process-message")
                    .tag("topic", record.topic())
                    .tag("partition", String.valueOf(record.partition()))
                    .metrics()
                    .transform(processingTime::record)
                    .doOnSuccess(result -> {
                        processedMessages.increment();
                        Gauge.builder("kafka.last.processed.timestamp", 
                                () -> System.currentTimeMillis())
                            .register(registry);
                    })
                    .doOnError(e -> failedMessages.increment());
            })
            .retry() // Важно для продолжения работы при ошибках
            .subscribe();
    }
    
    private Mono<String> processRecord(ReceiverRecord<String, String> record) {
        // Бизнес-логика обработки
        return Mono.fromCallable(() -> {
            // Логика обработки
            return "Processed: " + record.value();
        });
    }
}

@Service
public class ResilientExternalService {
    
    private final WebClient webClient;
    private final CircuitBreakerFactory circuitBreakerFactory;
    
    public ResilientExternalService(WebClient.Builder webClientBuilder,
                                   CircuitBreakerFactory circuitBreakerFactory) {
        this.webClient = webClientBuilder.baseUrl("https://api.external.com").build();
        this.circuitBreakerFactory = circuitBreakerFactory;
    }
    
    public Mono<ExternalData> getExternalData(String id) {
        CircuitBreaker circuitBreaker = circuitBreakerFactory.create("externalService");
        
        return ReactiveCircuitBreaker.create(circuitBreaker)
            .run(webClient.get()
                .uri("/data/{id}", id)
                .retrieve()
                .bodyToMono(ExternalData.class),
                throwable -> {
                    log.error("Ошибка при вызове внешнего сервиса", throwable);
                    return Mono.just(ExternalData.fallback(id));
                }
            );
    }
}

@Service
public class SwitchableKafkaService {
    
    private final ReactiveKafkaConsumerTemplate<String, String> primaryConsumer;
    private final ReactiveKafkaConsumerTemplate<String, String> backupConsumer;
    private final AtomicBoolean usePrimary = new AtomicBoolean(true);
    
    public SwitchableKafkaService(
            @Qualifier("primaryConsumer") ReactiveKafkaConsumerTemplate<String, String> primaryConsumer,
            @Qualifier("backupConsumer") ReactiveKafkaConsumerTemplate<String, String> backupConsumer) {
        this.primaryConsumer = primaryConsumer;
        this.backupConsumer = backupConsumer;
    }
    
    public Flux<ReceiverRecord<String, String>> consumeMessages() {
        return Flux.defer(() -> {
            if (usePrimary.get()) {
                return primaryConsumer.receive()
                    .doOnError(e -> {
                        log.error("Ошибка основного источника, переключение на резервный", e);
                        usePrimary.set(false);
                    });
            } else {
                return backupConsumer.receive()
                    .doOnError(e -> {
                        log.error("Ошибка резервного источника, переключение на основной", e);
                        usePrimary.set(true);
                    });
            }
        }).retry(); // Повторять при ошибках для автоматического переключения
    }
    
    // Ручное переключение источников
    public void switchToBackup() {
        usePrimary.set(false);
    }
    
    public void switchToPrimary() {
        usePrimary.set(true);
    }
}

@Service
public class EventReplicationService {
    
    private final ReactiveKafkaProducerTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;
    
    public EventReplicationService(ReactiveKafkaProducerTemplate<String, String> kafkaTemplate,
                                  ObjectMapper objectMapper) {
        this.kafkaTemplate = kafkaTemplate;
        this.objectMapper = objectMapper;
    }
    
    public Mono<Void> sendWithReplication(String mainTopic, String backupTopic, String key, Object event) {
        try {
            String eventJson = objectMapper.writeValueAsString(event);
            
            return kafkaTemplate.send(mainTopic, key, eventJson)
                .then(kafkaTemplate.send(backupTopic, key, eventJson))
                .then();
        } catch (JsonProcessingException e) {
            return Mono.error(e);
        }
    }
    
    // Версия с параллельной отправкой в оба топика
    public Mono<Void> sendWithParallelReplication(String mainTopic, String backupTopic, 
                                                String key, Object event) {
        try {
            String eventJson = objectMapper.writeValueAsString(event);
            
            return Mono.zip(
                    kafkaTemplate.send(mainTopic, key, eventJson),
                    kafkaTemplate.send(backupTopic, key, eventJson)
                ).then();
        } catch (JsonProcessingException e) {
            return Mono.error(e);
        }
    }
}

@Service
public class ProductAggregatorService {
    
    private final ProductRepository productRepository;
    private final ReviewService reviewService;
    private final InventoryService inventoryService;
    private final PricingService pricingService;
    
    public Mono<ProductAggregate> getProductDetails(String productId) {
        return Mono.zip(
                productRepository.findById(productId),
                reviewService.getReviewsForProduct(productId).collectList(),
                inventoryService.getInventoryStatus(productId),
                pricingService.getCurrentPricing(productId)
            )
            .map(tuple -> {
                Product product = tuple.getT1();
                List<Review> reviews = tuple.getT2();
                InventoryStatus inventory = tuple.getT3();
                PricingInfo pricing = tuple.getT4();
                
                return new ProductAggregate(product, reviews, inventory, pricing);
            })
            .timeout(Duration.ofSeconds(3))
            .onErrorResume(TimeoutException.class, e -> {
                // Возвращаем частичные данные при таймауте
                return productRepository.findById(productId)
                    .map(product -> ProductAggregate.partial(product));
            });
    }
}

@Service
public class ReactiveCacheService {
    
    private final ConcurrentMap<String, Signal<? extends Object>> cache = new ConcurrentHashMap<>();
    
    public <T> Mono<T> cacheOrCompute(String key, Mono<T> source, Duration ttl) {
        return Mono.defer(() -> {
            // Проверка кэша
            Signal<? extends Object> cachedSignal = cache.get(key);
            
            if (cachedSignal != null) {
                // Кэш содержит сигнал завершения (ошибка или успех)
                if (cachedSignal.isOnError()) {
                    return Mono.error(cachedSignal.getThrowable());
                } else if (cachedSignal.isOnNext()) {
                    return Mono.just((T) cachedSignal.get());
                }
            }
            
            // Кэш не содержит сигнала или содержит пустой сигнал
            return source
                .doOnNext(value -> cache.put(key, Signal.next(value)))
                .doOnError(error -> cache.put(key, Signal.error(error)))
                .doFinally(signalType -> {
                    if (ttl.toMillis() > 0) {
                        Mono.delay(ttl)
                            .doOnNext(l -> cache.remove(key))
                            .subscribe();
                    }
                });
        });
    }
    
    public void invalidateCache(String key) {
        cache.remove(key);
    }
    
    public void invalidateAllCache() {
        cache.clear();
    }
}