Для чего нужно использовать Kafka для обращения к сервису?

Для чего нужно использовать Kafka для обращения к сервису

Kafka — это распределённая система обмена сообщениями (message broker), которая позволяет микросервисам общаться асинхронно через события. Это решает проблемы синхронного взаимодействия между сервисами.

Основная проблема синхронного взаимодействия

Синхронное взаимодействие (REST API):

Сервис A → HTTP запрос → Сервис B
          (ждёт ответа)

Проблемы:
- Если B недоступен, A падает
- Если B медленный, A блокируется
- Тесная связанность сервисов
- Сложно масштабировать

Асинхронное взаимодействие через Kafka

Кафка как посредник:

Сервис A          Kafka         Сервис B
  |                |               |
  |-- Отправить → |               |
  |   событие    |               |
  |              |-- Доставить → |
  | (продолжить) |   событие    | (обработать)
  |              |               |

Зачем Kafka нужна

1. Развязывание сервисов (Decoupling)

// БЕЗ Kafka: тесная связанность
@RestController
public class OrderController {
    @Autowired
    private PaymentService paymentService;  // Прямая зависимость!
    @Autowired
    private InventoryService inventoryService;  // Прямая зависимость!
    @Autowired
    private NotificationService notificationService;  // Прямая зависимость!
    
    @PostMapping("/orders")
    public Order createOrder(@RequestBody CreateOrderRequest request) {
        // Синхронно вызываем три сервиса
        paymentService.processPayment(request.getAmount());  // Могла упасть
        inventoryService.reserve(request.getItems());  // Могла упасть
        notificationService.sendEmail(request.getEmail());  // Могла упасть
        
        return new Order();
    }
}

// С Kafka: слабая связанность
@RestController
public class OrderController {
    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;
    
    @PostMapping("/orders")
    public Order createOrder(@RequestBody CreateOrderRequest request) {
        // Отправляем событие в Kafka
        OrderEvent event = new OrderEvent(request);
        kafkaTemplate.send("orders-topic", event);
        
        // Контроллер СРАЗУ возвращает ответ
        return new Order();
        // PaymentService, InventoryService слушают события
        // и обрабатывают независимо
    }
}

2. Асинхронная обработка

// Микросервис платежей слушает события
@Service
public class PaymentService {
    @KafkaListener(topics = "orders-topic")
    public void handleOrderEvent(OrderEvent event) {
        System.out.println("Processing payment for order: " + event.getOrderId());
        
        try {
            // Обработка платежа (может быть медленно)
            processPayment(event.getAmount());
            System.out.println("Payment processed");
        } catch (Exception e) {
            // Ошибка = попытка ещё раз (retry)
            // Kafka гарантирует доставку
        }
    }
}

// Микросервис инвентаря слушает те же события
@Service
public class InventoryService {
    @KafkaListener(topics = "orders-topic")
    public void handleOrderEvent(OrderEvent event) {
        System.out.println("Reserving inventory for order: " + event.getOrderId());
        reserveItems(event.getItems());
    }
}

// Микросервис уведомлений
@Service
public class NotificationService {
    @KafkaListener(topics = "orders-topic")
    public void handleOrderEvent(OrderEvent event) {
        System.out.println("Sending notification");
        sendEmail(event.getEmail());
    }
}

3. Гарантированная доставка

// Kafka гарантирует, что сообщение будет доставлено
// даже если консьюмер был недоступен

// Сообщение в очереди:
[Event1] → Kafka Topic
[Event2] →
[Event3] →

// Когда сервис вернулся в сеть:
@KafkaListener(topics = "orders-topic", groupId = "payment-service")
public void handleOrderEvent(OrderEvent event) {
    // Получит все события, которые были в очереди
    // Гарантированная доставка и обработка
}

4. Масштабируемость

// Несколько инстансов одного сервиса
// Kafka распределяет события между ними

// Сервис A (инстанс 1)
@KafkaListener(topics = "orders-topic", groupId = "order-processor")
public void process(OrderEvent event) {
    processOrder(event);
}

// Сервис A (инстанс 2)
@KafkaListener(topics = "orders-topic", groupId = "order-processor")
public void process(OrderEvent event) {
    processOrder(event);
}

// Сервис A (инстанс 3)
@KafkaListener(topics = "orders-topic", groupId = "order-processor")
public void process(OrderEvent event) {
    processOrder(event);
}

// Kafka автоматически распределяет события
// Никакой дополнительной конфигурации не нужно

Полный пример: обработка заказов

// Event
public class OrderCreatedEvent {
    private String orderId;
    private String userId;
    private BigDecimal amount;
    private List<String> items;
    private String email;
    
    // Getters and setters
}

// Producer (создание события)
@Service
public class OrderService {
    @Autowired
    private OrderRepository orderRepository;
    @Autowired
    private KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;
    
    public Order createOrder(CreateOrderRequest request) {
        // 1. Сохранить заказ
        Order order = new Order();
        order.setUserId(request.getUserId());
        order.setAmount(request.getAmount());
        order.setItems(request.getItems());
        Order saved = orderRepository.save(order);
        
        // 2. Отправить событие
        OrderCreatedEvent event = new OrderCreatedEvent();
        event.setOrderId(saved.getId());
        event.setUserId(request.getUserId());
        event.setAmount(request.getAmount());
        event.setItems(request.getItems());
        event.setEmail(request.getEmail());
        
        kafkaTemplate.send("order-events", event);
        
        return saved;
    }
}

// Consumer 1: Payment Service
@Service
public class PaymentProcessor {
    @KafkaListener(topics = "order-events", groupId = "payment-group")
    public void processPayment(OrderCreatedEvent event) {
        System.out.println("Processing payment for order: " + event.getOrderId());
        // Обработка платежа...
    }
}

// Consumer 2: Inventory Service
@Service
public class InventoryProcessor {
    @KafkaListener(topics = "order-events", groupId = "inventory-group")
    public void reserveInventory(OrderCreatedEvent event) {
        System.out.println("Reserving items for order: " + event.getOrderId());
        // Резервирование товаров...
    }
}

// Consumer 3: Notification Service
@Service
public class NotificationProcessor {
    @KafkaListener(topics = "order-events", groupId = "notification-group")
    public void sendNotification(OrderCreatedEvent event) {
        System.out.println("Sending notification to: " + event.getEmail());
        // Отправка email...
    }
}

Конфигурация Spring Boot + Kafka

# application.properties
spring.kafka.bootstrap-servers=localhost:9092

# Producer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

# Consumer
spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-group
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.properties.spring.json.trusted.packages=*

Сравнение: REST API vs Kafka

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

1. Асинхронные события
   - Создание заказа → обработка платежа
   - Регистрация → отправка письма

2. Развязывание сервисов
   - Сервис A не зависит от доступности B

3. Гарантированная доставка
   - Все события должны быть обработаны

4. Обработка больших объёмов
   - Миллионы событий в секунду

5. Event sourcing
   - История всех событий в системе

Итоговый ответ

Kafka нужна для:

1. Развязывания микросервисов — слабая связанность
2. Асинхронной обработки — неблокирующее взаимодействие
3. Гарантированной доставки — отказоустойчивость
4. Масштабируемости — легко масштабировать
5. Event-driven архитектуры — система на основе событий

Вместо прямого вызова REST API между микросервисами, они отправляют события в Kafka. Это обеспечивает гибкость, надёжность и масштабируемость современных распределённых систем.