Приведи пример взаимодействия между микросервисами с помощью очередей

Взаимодействие микросервисов через очереди

Асинхронная коммуникация через очереди (Message Queues) — это один из ключевых паттернов в микросервисной архитектуре. Вместо прямых HTTP запросов, сервисы отправляют сообщения в очередь, и потребитель обрабатывает их асинхронно.

Архитектура

Producer (Заказы)         Message Broker (RabbitMQ)      Consumer (Email)
      ↓                           ↓                              ↓
  Order Service  ───send───>  Queue: orders  ───consume───>  Email Service
      ↓                           ↓                              ↓
  Payment Service ───send───>  Queue: payments ───consume───>  Payment Service
      ↓                           ↓                              ↓
  Inventory Service ───send───>  Queue: inventory ───consume───>  Inventory Service

Сценарий: E-Commerce платформа

Когда пользователь делает заказ, нужно:

1. Сохранить заказ в БД
2. Обработать платёж
3. Зарезервировать товар
4. Отправить email
5. Уведомить доставку

Синхронный подход (плохо):

Если Email сервис упадёт, весь заказ будет заблокирован.

Асинхронный подход через очереди (хорошо):

Заказ создаётся, затем разные сервисы обрабатывают события асинхронно.

Реализация с RabbitMQ

Зависимости

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

Конфигурация RabbitMQ

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Bean;

@Configuration
public class RabbitMQConfig {
    
    // Очереди
    public static final String ORDER_QUEUE = "order.queue";
    public static final String PAYMENT_QUEUE = "payment.queue";
    public static final String EMAIL_QUEUE = "email.queue";
    public static final String INVENTORY_QUEUE = "inventory.queue";
    
    // Exchange (маршрутизатор)
    public static final String ORDER_EXCHANGE = "order.exchange";
    
    // Routing keys
    public static final String ORDER_ROUTING_KEY = "order.created";
    public static final String PAYMENT_ROUTING_KEY = "payment.processed";
    
    // Создаём очереди
    @Bean
    public Queue orderQueue() {
        return new Queue(ORDER_QUEUE, true); // true = persistent
    }
    
    @Bean
    public Queue paymentQueue() {
        return new Queue(PAYMENT_QUEUE, true);
    }
    
    @Bean
    public Queue emailQueue() {
        return new Queue(EMAIL_QUEUE, true);
    }
    
    @Bean
    public Queue inventoryQueue() {
        return new Queue(INVENTORY_QUEUE, true);
    }
    
    // Создаём Topic Exchange (для маршрутизации по ключам)
    @Bean
    public TopicExchange orderExchange() {
        return new TopicExchange(ORDER_EXCHANGE, true, false);
    }
    
    // Связываем Exchange с Queue (Binding)
    @Bean
    public Binding orderBinding(Queue orderQueue, TopicExchange orderExchange) {
        return BindingBuilder.bind(orderQueue)
            .to(orderExchange)
            .with(ORDER_ROUTING_KEY);
    }
    
    @Bean
    public Binding paymentBinding(Queue paymentQueue, TopicExchange orderExchange) {
        return BindingBuilder.bind(paymentQueue)
            .to(orderExchange)
            .with(PAYMENT_ROUTING_KEY);
    }
}

Event класс (сообщение)

import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.UUID;

public class OrderCreatedEvent implements Serializable {
    private static final long serialVersionUID = 1L;
    
    private String orderId;           // UUID заказа
    private String userId;            // ID пользователя
    private String email;             // Email для отправки
    private double totalAmount;       // Сумма
    private LocalDateTime createdAt;  // Время создания
    private List<OrderItem> items;    // Товары в заказе
    
    // Конструкторы
    public OrderCreatedEvent() {}
    
    public OrderCreatedEvent(String orderId, String userId, String email, 
                            double totalAmount, List<OrderItem> items) {
        this.orderId = orderId;
        this.userId = userId;
        this.email = email;
        this.totalAmount = totalAmount;
        this.items = items;
        this.createdAt = LocalDateTime.now();
    }
    
    // Getters и setters
    public String getOrderId() { return orderId; }
    public String getUserId() { return userId; }
    public String getEmail() { return email; }
    public double getTotalAmount() { return totalAmount; }
    public List<OrderItem> getItems() { return items; }
    public LocalDateTime getCreatedAt() { return createdAt; }
}

public class OrderItem {
    private String productId;
    private int quantity;
    private double price;
    
    // Конструкторы, getters, setters
}

Producer: Order Service (отправляет события)

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.UUID;

@Service
public class OrderService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;  // Для отправки сообщений
    
    @Autowired
    private OrderRepository orderRepository;
    
    public Order createOrder(CreateOrderRequest request) {
        // 1. Сохраняем заказ в БД
        Order order = new Order();
        order.setId(UUID.randomUUID().toString());
        order.setUserId(request.getUserId());
        order.setEmail(request.getEmail());
        order.setTotalAmount(request.getTotalAmount());
        order.setItems(request.getItems());
        order.setStatus("PENDING");  // Заказ в ожидании
        
        order = orderRepository.save(order);
        
        // 2. Публикуем событие в очередь
        OrderCreatedEvent event = new OrderCreatedEvent(
            order.getId(),
            order.getUserId(),
            order.getEmail(),
            order.getTotalAmount(),
            order.getItems()
        );
        
        // Отправляем в exchange с routing key
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            event
        );
        
        System.out.println("Order " + order.getId() + " created and event published");
        
        return order;
    }
}

Consumer 1: Payment Service (обрабатывает платежи)

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class PaymentService {
    
    @Autowired
    private PaymentRepository paymentRepository;
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    // Слушаем очередь payment.queue
    @RabbitListener(queues = RabbitMQConfig.PAYMENT_QUEUE)
    public void processPayment(OrderCreatedEvent event) {
        try {
            System.out.println("Processing payment for order: " + event.getOrderId());
            
            // Обработка платежа
            Payment payment = new Payment();
            payment.setOrderId(event.getOrderId());
            payment.setAmount(event.getTotalAmount());
            payment.setStatus("PROCESSING");
            
            // Заглушка: имитация успешного платежа
            Thread.sleep(1000); // Имитируем задержку обработки
            
            payment.setStatus("SUCCESS");
            paymentRepository.save(payment);
            
            System.out.println("Payment SUCCESS for order: " + event.getOrderId());
            
            // Отправляем событие Payment Processed
            PaymentProcessedEvent paymentEvent = new PaymentProcessedEvent(
                event.getOrderId(),
                "SUCCESS"
            );
            
            rabbitTemplate.convertAndSend(
                RabbitMQConfig.ORDER_EXCHANGE,
                RabbitMQConfig.PAYMENT_ROUTING_KEY,
                paymentEvent
            );
            
        } catch (Exception e) {
            System.err.println("Payment FAILED for order: " + event.getOrderId());
            e.printStackTrace();
        }
    }
}

Consumer 2: Email Service (отправляет уведомления)

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.mail.SimpleMailMessage;
import org.springframework.mail.javamail.JavaMailSender;
import org.springframework.stereotype.Service;

@Service
public class EmailService {
    
    @Autowired
    private JavaMailSender mailSender;
    
    // Слушаем очередь email.queue
    @RabbitListener(queues = RabbitMQConfig.EMAIL_QUEUE)
    public void sendOrderConfirmation(OrderCreatedEvent event) {
        try {
            System.out.println("Sending email to: " + event.getEmail());
            
            SimpleMailMessage message = new SimpleMailMessage();
            message.setTo(event.getEmail());
            message.setSubject("Order Confirmation - Order #" + event.getOrderId());
            
            StringBuilder body = new StringBuilder();
            body.append("Hello,\n\n");
            body.append("Your order #").append(event.getOrderId()).append(" has been created.\n");
            body.append("Total: $").append(event.getTotalAmount()).append("\n\n");
            body.append("Items:\n");
            for (OrderItem item : event.getItems()) {
                body.append("- ").append(item.getProductId())
                    .append(" x").append(item.getQuantity())
                    .append(" = $").append(item.getPrice()).append("\n");
            }
            body.append("\nThank you!");
            
            message.setText(body.toString());
            
            // Отправляем email
            mailSender.send(message);
            
            System.out.println("Email sent successfully to: " + event.getEmail());
            
        } catch (Exception e) {
            System.err.println("Failed to send email: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

Consumer 3: Inventory Service (резервирует товары)

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class InventoryService {
    
    @Autowired
    private InventoryRepository inventoryRepository;
    
    @RabbitListener(queues = RabbitMQConfig.INVENTORY_QUEUE)
    public void reserveInventory(OrderCreatedEvent event) {
        try {
            System.out.println("Reserving inventory for order: " + event.getOrderId());
            
            for (OrderItem item : event.getItems()) {
                // Получаем товар и уменьшаем количество
                Inventory inventory = inventoryRepository.findByProductId(item.getProductId())
                    .orElseThrow(() -> new RuntimeException("Product not found"));
                
                if (inventory.getQuantityAvailable() < item.getQuantity()) {
                    throw new RuntimeException("Not enough inventory for: " + item.getProductId());
                }
                
                inventory.setQuantityAvailable(inventory.getQuantityAvailable() - item.getQuantity());
                inventoryRepository.save(inventory);
                
                System.out.println("Reserved " + item.getQuantity() + " units of " 
                    + item.getProductId());
            }
            
            System.out.println("Inventory reserved for order: " + event.getOrderId());
            
        } catch (Exception e) {
            System.err.println("Inventory reservation FAILED: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

REST Controller для создания заказа

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/v1/orders")
public class OrderController {
    
    @Autowired
    private OrderService orderService;
    
    @PostMapping
    public ResponseEntity<?> createOrder(@RequestBody CreateOrderRequest request) {
        try {
            Order order = orderService.createOrder(request);
            return ResponseEntity.ok(order);
        } catch (Exception e) {
            return ResponseEntity.badRequest().body("Error: " + e.getMessage());
        }
    }
}

public class CreateOrderRequest {
    private String userId;
    private String email;
    private double totalAmount;
    private List<OrderItem> items;
    
    // Getters и setters
}

Полный поток обработки

Сценарий: User создаёт заказ

1. POST /api/v1/orders с данными заказа
   ↓
2. OrderService.createOrder()
   ├─ Сохраняет Order в БД (статус PENDING)
   └─ Отправляет OrderCreatedEvent в RabbitMQ
   ↓
3. RabbitMQ маршрутизирует событие в очереди:
   ├─ payment.queue    → PaymentService слушает
   ├─ email.queue      → EmailService слушает
   └─ inventory.queue  → InventoryService слушает
   ↓
4. Асинхронная обработка (параллельно):
   ├─ PaymentService обрабатывает платёж (может занять 1-3 сек)
   ├─ EmailService отправляет письмо (может занять 0.5-2 сек)
   └─ InventoryService резервирует товар (может занять 0.1-1 сек)
   ↓
5. Когда все готово:
   ├─ Order статус становится CONFIRMED
   ├─ User получает email подтверждение
   ├─ Товар зарезервирован
   └─ Платёж обработан
   ↓
6. Если Email или Inventory упадут, Order всё равно создан
   (они будут обработаны когда восстановятся)

Преимущества асинхронной архитектуры

1. Слабая связанность — сервисы не зависят друг от друга
2. Масштабируемость — можно добавить больше потребителей
3. Отказоустойчивость — если Email упадёт, заказ всё равно обработан
4. Производительность — клиент получает ответ быстро (Order создан)
5. Переиспользуемость — разные сервисы могут слушать одно событие
6. Аудит — все события сохраняются в очереди

Недостатки

1. Усложнение архитектуры — нужен Message Broker
2. Отладка сложнее — асинхронный код требует более тщательного тестирования
3. Консистентность данных — может быть eventual consistency
4. Операционная сложность — нужно мониторить очереди

Этот паттерн используется в большинстве high-load систем (Amazon, Netflix, Uber, Airbnb).