Может ли монолит взаимодействовать с Kafka?

Может ли монолит взаимодействовать с Kafka

Да, монолит может взаимодействовать с Kafka. Это правильный подход для асинхронной коммуникации, даже в монолитной архитектуре. Kafka часто используется как event bus или message broker внутри монолита.

Почему монолит + Kafka?

Преимущества:

• Асинхронность: Операции не блокируют друг друга
• Масштабируемость: Легко добавить параллельные обработчики
• Отделение логики: Разделение concerns (domain → event publishing)
• Миграция: Переход от монолита к микросервисам упрощен
• Надежность: События не теряются даже при падении приложения

Архитектура: Монолит + Kafka

┌──────────────────────────────────────────┐
│         Monolith Application             │
│                                          │
│  ┌─────────────────────────────────┐   │
│  │   Business Logic                │   │
│  │   (Users, Orders, Payments)     │   │
│  └─────────────┬───────────────────┘   │
│                │                        │
│  ┌─────────────▼──────────────────┐   │
│  │  Event Publisher (Kafka)       │   │
│  │  - UserCreatedEvent            │   │
│  │  - OrderPlacedEvent            │   │
│  │  - PaymentProcessedEvent       │   │
│  └─────────────┬──────────────────┘   │
│                │                        │
└────────────────┼────────────────────────┘
                 │
              Kafka Broker
                 │
      ┌──────────┴──────────┐
      │                     │
      ▼                     ▼
  Listener 1            Listener 2
  (Send Email)       (Update Analytics)

Практическая реализация

1. Подключение зависимостей (Spring Boot)

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>3.0.0</version>
</dependency>

2. Event класс

public class UserCreatedEvent {
    private Long userId;
    private String email;
    private String username;
    private LocalDateTime createdAt;
    
    // Constructor, getters, setters
    public UserCreatedEvent(Long userId, String email, String username) {
        this.userId = userId;
        this.email = email;
        this.username = username;
        this.createdAt = LocalDateTime.now();
    }
}

3. Event Publisher (отправитель)

@Service
public class UserService {
    private final UserRepository userRepository;
    private final KafkaTemplate<String, UserCreatedEvent> kafkaTemplate;
    
    private static final String TOPIC = "user-events";
    
    @Autowired
    public UserService(UserRepository userRepository,
                       KafkaTemplate<String, UserCreatedEvent> kafkaTemplate) {
        this.userRepository = userRepository;
        this.kafkaTemplate = kafkaTemplate;
    }
    
    @Transactional
    public User createUser(CreateUserRequest request) {
        // 1. Создаем пользователя
        User user = new User(request.getEmail(), request.getUsername());
        User saved = userRepository.save(user);
        
        // 2. Публикуем событие в Kafka
        UserCreatedEvent event = new UserCreatedEvent(
            saved.getId(),
            saved.getEmail(),
            saved.getUsername()
        );
        
        kafkaTemplate.send(TOPIC, event.getUserId().toString(), event);
        
        return saved;
    }
}

4. Event Listener (получатель)

@Service
public class UserEventListener {
    private final EmailService emailService;
    private final AnalyticsService analyticsService;
    
    @Autowired
    public UserEventListener(EmailService emailService,
                             AnalyticsService analyticsService) {
        this.emailService = emailService;
        this.analyticsService = analyticsService;
    }
    
    @KafkaListener(
        topics = "user-events",
        groupId = "user-email-group",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void onUserCreated(UserCreatedEvent event) {
        try {
            // Отправляем приветственное письмо
            emailService.sendWelcomeEmail(
                event.getEmail(),
                event.getUsername()
            );
        } catch (Exception e) {
            // Логируем ошибку, Kafka retry сработает
            System.err.println("Ошибка при отправке письма: " + e.getMessage());
            throw new RuntimeException(e);
        }
    }
    
    @KafkaListener(
        topics = "user-events",
        groupId = "user-analytics-group"
    )
    public void trackUserCreation(UserCreatedEvent event) {
        // Отслеживаем созданного пользователя
        analyticsService.trackEvent("user.created", Map.of(
            "userId", event.getUserId(),
            "email", event.getEmail(),
            "timestamp", event.getCreatedAt()
        ));
    }
}

5. Конфигурация Kafka

@Configuration
public class KafkaConfig {
    
    // Producer Configuration
    @Bean
    public ProducerFactory<String, UserCreatedEvent> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
                       "localhost:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
                       StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
                       JsonSerializer.class);
        configProps.put(ProducerConfig.ACKS_CONFIG, "all");  // Надежность
        configProps.put(ProducerConfig.RETRIES_CONFIG, 3);   // Повторы
        
        return new DefaultProducerFactory<>(configProps);
    }
    
    @Bean
    public KafkaTemplate<String, UserCreatedEvent> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
    
    // Consumer Configuration
    @Bean
    public ConsumerFactory<String, UserCreatedEvent> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 
                       "localhost:9092");
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, 
                       "user-service-group");
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
                       StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
                       JsonDeserializer.class);
        configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, 
                       "earliest");  // Начать с начала если нет offset
        configProps.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        
        return new DefaultConsumerFactory<>(configProps);
    }
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, UserCreatedEvent> 
           kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, UserCreatedEvent> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);  // 3 параллельных потока
        
        return factory;
    }
}

Properties конфигурация

# application.properties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.group-id=my-monolith
spring.kafka.consumer.auto-offset-reset=earliest

Сложный пример: Saga Pattern

@Service
public class OrderService {
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    private final OrderRepository orderRepository;
    
    @Transactional
    public void placeOrder(PlaceOrderRequest request) {
        // 1. Сохраняем заказ в статусе PENDING
        Order order = new Order(
            request.getUserId(),
            request.getItems(),
            OrderStatus.PENDING
        );
        Order saved = orderRepository.save(order);
        
        // 2. Публикуем событие в Kafka
        OrderPlacedEvent event = new OrderPlacedEvent(
            saved.getId(),
            saved.getUserId(),
            saved.getTotalAmount()
        );
        kafkaTemplate.send("orders", event.getOrderId().toString(), event);
    }
}

// Слушатель 1: Проверка платежа
@Service
public class PaymentListener {
    @KafkaListener(topics = "orders", groupId = "payment-group")
    public void onOrderPlaced(OrderPlacedEvent event) {
        try {
            paymentService.processPayment(event.getOrderId(), event.getAmount());
            kafkaTemplate.send("payments", 
                new PaymentSuccessEvent(event.getOrderId()));
        } catch (PaymentException e) {
            kafkaTemplate.send("payments", 
                new PaymentFailedEvent(event.getOrderId()));
        }
    }
}

// Слушатель 2: Обновление инвентаря
@Service
public class InventoryListener {
    @KafkaListener(topics = "payments", groupId = "inventory-group")
    public void onPaymentSuccess(PaymentSuccessEvent event) {
        inventoryService.reserveItems(event.getOrderId());
    }
}

// Слушатель 3: Отправка уведомления
@Service
public class NotificationListener {
    @KafkaListener(topics = "payments", groupId = "notification-group")
    public void onPaymentSuccess(PaymentSuccessEvent event) {
        emailService.sendOrderConfirmation(event.getOrderId());
    }
}

Обработка ошибок и retry

@Service
public class RobustUserEventListener {
    
    @KafkaListener(topics = "user-events", groupId = "user-group")
    public void onUserCreated(UserCreatedEvent event) {
        try {
            emailService.sendWelcomeEmail(event.getEmail(), event.getUsername());
        } catch (Exception e) {
            // Kafka автоматически переотправит (по конфигу)
            throw new RuntimeException("Failed to send email", e);
        }
    }
    
    // Альтернатива: Dead Letter Topic
    @KafkaListener(
        topics = "user-events.DLT",  // Dead Letter Topic
        groupId = "user-dlt-group"
    )
    public void handleFailure(UserCreatedEvent event) {
        // Логируем события, которые не удалось обработать
        System.err.println("Failed to process event: " + event);
        // Отправляем администратору
        alertAdmins("Event processing failed for: " + event.getEmail());
    }
}

Преимущества монолита + Kafka

Когда использовать Kafka в монолите?

• Email отправки (асинхронные письма)
• Аналитика (tracking events)
• Уведомления (push, SMS)
• Интеграции (внешние API)
• Лог-события (audit trail)
• Долгие операции (report generation)

Вывод

Да, монолит может и должен использовать Kafka для асинхронной коммуникации. Это не только возможно, но и рекомендуется для:

1. Улучшения производительности (асинхронность)
2. Повышения надежности (события не теряются)
3. Облегчения перехода на микросервисы
4. Разделения ответственности (separation of concerns)

Кафка в монолите — это отличная инвестиция в архитектуру, которая окупается по мере роста проекта.