Обмен данными в микросервисной архитектуре

GET /api/users/123 HTTP/1.1
Host: user-service.example.com

syntax = "proto3";
 
service UserService {
  rpc GetUser (UserRequest) returns (User);
}
 
message UserRequest {
  string user_id = 1;
}
 
message User {
  string user_id = 1;
  string name = 2;
  string email = 3;
}

// Пример отправки сообщения в Kafka
ProducerRecord<String, String> record = 
    new ProducerRecord<>("user-updates", userId, userJson);
producer.send(record);

query {
  user(id: "123") {
    name
    email
    avatar
    lastLogin
  }
}

message Payment {
  string transaction_id = 1;
  double amount = 2;
  string currency = 3;
  enum Status {
    PENDING = 0;
    COMPLETED = 1;
    FAILED = 2;
  }
  Status status = 4;
}

// Обновление локальной копии данных при получении события
@KafkaListener(topics = "user-updates")
public void handleUserUpdate(UserUpdatedEvent event) {
    UserView user = userViewRepository.findById(event.getUserId())
        .orElse(new UserView());
    user.setDisplayName(event.getDisplayName());
    user.setAvatarUrl(event.getAvatarUrl());
    user.setLastUpdated(LocalDateTime.now());
    userViewRepository.save(user);
}

// Фрагмент оркестрации саги процесса заказа
public void startOrderProcess(OrderCommand cmd) {
    // Резервируем товар на складе
    kafkaTemplate.send("inventory-commands", 
        new ReserveItemsCommand(cmd.getOrderId(), cmd.getItems()));
}
 
@KafkaListener(topics = "inventory-events")
public void handleInventoryEvent(InventoryEvent event) {
    if (event instanceof ItemsReservedEvent) {
        // Переходим к обработке платежа
        kafkaTemplate.send("payment-commands",
            new ProcessPaymentCommand(event.getOrderId(), event.getAmount()));
    } else if (event instanceof ReservationFailedEvent) {
        // Завершаем сагу с ошибкой
        kafkaTemplate.send("order-events",
            new OrderFailedEvent(event.getOrderId(), "Нет в наличии"));
    }
}
 
@KafkaListener(topics = "payment-events")
public void handlePaymentEvent(PaymentEvent event) {
    if (event instanceof PaymentFailedEvent) {
        // Компенсирующая транзакция: отменяем резервирование
        kafkaTemplate.send("inventory-commands",
            new ReleaseItemsCommand(event.getOrderId()));
    }
}

// Пример CQRS в сервисе заказов
@Service
public class OrderCommandService {
    public void createOrder(CreateOrderCommand cmd) {
        // Валидация и создание событий для изменения состояния
        OrderCreatedEvent event = new OrderCreatedEvent(
            UUID.randomUUID().toString(),
            cmd.getCustomerId(),
            cmd.getItems(),
            LocalDateTime.now()
        );
        eventStore.saveEvent(event);
        kafkaTemplate.send("order-events", event);
    }
}
 
@Service
public class OrderQueryService {
    public List<OrderSummary> getCustomerOrders(String customerId) {
        // Чтение из оптимизированного представления
        return orderViewRepository.findByCustomerId(customerId);
    }
}

+----------------+     +----------------+     +----------------+
| Product Service|     | Order Service  |     | Payment Service|
+----------------+     +----------------+     +----------------+
        |                      |                      |
        v                      v                      v
+----------------+     +----------------+     +----------------+
| Product DB     |     | Order DB       |     | Payment DB     |
+----------------+     +----------------+     +----------------+

// Создание агрегата через события
public class Account {
    private String id;
    private double balance;
    private List<Event> changes = new ArrayList<>();
    
    public void applyDeposit(double amount) {
        apply(new DepositedEvent(id, amount));
    }
    
    public void applyWithdrawal(double amount) {
        if (balance < amount) {
            throw new InsufficientFundsException();
        }
        apply(new WithdrawnEvent(id, amount));
    }
    
    private void apply(Event event) {
        if (event instanceof DepositedEvent) {
            balance += ((DepositedEvent) event).getAmount();
        } else if (event instanceof WithdrawnEvent) {
            balance -= ((WithdrawnEvent) event).getAmount();
        }
        changes.add(event);
    }
}

# Схема сервиса заказов
type Order @key(fields: "id") {
  id: ID!
  created: DateTime!
  customer: Customer
  items: [OrderItem!]!
}
 
# Схема сервиса клиентов
type Customer @key(fields: "id") {
  id: ID!
  name: String!
  email: String!
  orders: [Order!]!
}

// Инвалидация кэша по событию
@KafkaListener(topics = "product-updated")
public void handleProductUpdate(ProductUpdatedEvent event) {
    // Удаляем устаревшие данные из кэша
    cacheManager.getCache("products").evict(event.getProductId());
    
    // Или обновляем кэш новыми данными
    Product product = productRepository.findById(event.getProductId()).orElse(null);
    if (product != null) {
        cacheManager.getCache("products").put(event.getProductId(), product);
    }
}

ContextMap eCommerceMap {
  contains ProductCatalogContext
  contains OrderContext
  contains CustomerContext
  
  ProductCatalogContext [U] <- [D] OrderContext
  CustomerContext [P] <-> [P] OrderContext
}

@Aggregate
public class Order {
    @AggregateIdentifier
    private OrderId id;
    private CustomerId customerId;
    private Money totalAmount;
    private OrderStatus status;
    private Set<OrderLine> orderLines;
    
    public void addProduct(ProductId productId, int quantity, Money price) {
        // Бизнес-логика добавления продукта
    }
    
    public void confirm() {
        if (orderLines.isEmpty()) {
            throw new EmptyOrderException();
        }
        this.status = OrderStatus.CONFIRMED;
        // Генерация события подтверждения заказа
    }
}

@Service
public class UserEventProducer {
    private final KafkaTemplate<String, UserEvent> kafkaTemplate;
    private final String topic = "user-events";
    
    public UserEventProducer(KafkaTemplate<String, UserEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    
    public void publishUserUpdated(String userId, UserDetails details) {
        UserUpdatedEvent event = new UserUpdatedEvent(
            userId,
            details.getDisplayName(),
            details.getEmail(),
            LocalDateTime.now()
        );
        
        // Используем ID пользователя как ключ для обеспечения порядка событий
        kafkaTemplate.send(topic, userId, event)
            .addCallback(
                success -> log.info("Событие опубликовано: {}", event),
                failure -> log.error("Ошибка публикации события: {}", failure.getMessage())
            );
    }
}

@Service
public class UserEventConsumer {
    private final UserViewRepository repository;
    
    @KafkaListener(topics = "user-events")
    public void consume(ConsumerRecord<String, UserEvent> record) {
        UserEvent event = record.value();
        
        if (event instanceof UserUpdatedEvent) {
            UserUpdatedEvent updateEvent = (UserUpdatedEvent) event;
            
            UserView userView = repository.findById(updateEvent.getUserId())
                .orElse(new UserView(updateEvent.getUserId()));
                
            userView.setDisplayName(updateEvent.getDisplayName());
            userView.setEmail(updateEvent.getEmail());
            userView.setLastUpdated(updateEvent.getTimestamp());
            
            repository.save(userView);
            log.info("Обновлено представление пользователя: {}", userView);
        }
    }
}

@Service
public class NotificationService {
    private final RabbitTemplate rabbitTemplate;
    
    public void sendNotification(Notification notification) {
        // Определяем маршрутизацию в зависимости от типа уведомления
        String routingKey = "notification." + notification.getType().toLowerCase();
        
        rabbitTemplate.convertAndSend("notifications-exchange", routingKey, notification);
        log.info("Отправлено уведомление с routing key: {}", routingKey);
    }
}

@Configuration
public class RabbitMQConfig {
    @Bean
    public TopicExchange notificationsExchange() {
        return new TopicExchange("notifications-exchange");
    }
    
    @Bean
    public Queue emailNotificationsQueue() {
        return new Queue("email-notifications");
    }
    
    @Bean
    public Queue pushNotificationsQueue() {
        return new Queue("push-notifications");
    }
    
    @Bean
    public Binding emailBinding(Queue emailNotificationsQueue, TopicExchange exchange) {
        return BindingBuilder.bind(emailNotificationsQueue)
            .to(exchange)
            .with("notification.email");
    }
    
    @Bean
    public Binding pushBinding(Queue pushNotificationsQueue, TopicExchange exchange) {
        return BindingBuilder.bind(pushNotificationsQueue)
            .to(exchange)
            .with("notification.push");
    }
}

@Transactional
public void updateUser(String userId, UserUpdateRequest request) {
    // Обновляем основную модель
    User user = userRepository.findById(userId)
        .orElseThrow(() -> new UserNotFoundException(userId));
    user.setDisplayName(request.getDisplayName());
    user.setEmail(request.getEmail());
    userRepository.save(user);
    
    // Записываем событие в таблицу исходящих сообщений
    OutboxMessage outboxMessage = new OutboxMessage(
        UUID.randomUUID().toString(),
        "user-events",
        userId, // ключ партиции
        objectMapper.writeValueAsString(new UserUpdatedEvent(
            userId,
            request.getDisplayName(),
            request.getEmail(),
            LocalDateTime.now()
        )),
        MessageStatus.PENDING
    );
    outboxRepository.save(outboxMessage);
}

@Scheduled(fixedRate = 5000)
public void processOutboxMessages() {
    List<OutboxMessage> pendingMessages = outboxRepository
        .findByStatus(MessageStatus.PENDING, PageRequest.of(0, 100));
        
    for (OutboxMessage message : pendingMessages) {
        try {
            kafkaTemplate.send(
                message.getTopic(),
                message.getPartitionKey(),
                message.getPayload()
            ).get(10, TimeUnit.SECONDS); // ждем подтверждения
            
            message.setStatus(MessageStatus.PUBLISHED);
            outboxRepository.save(message);
        } catch (Exception e) {
            log.error("Ошибка публикации сообщения {}: {}", 
                      message.getId(), e.getMessage());
            // Можно увеличить счетчик попыток или пометить как проблемное
        }
    }
}

@Service
public class ProductServiceClient {
    private final WebClient webClient;
    private final CircuitBreaker circuitBreaker;
    
    public ProductServiceClient(WebClient.Builder webClientBuilder, CircuitBreakerRegistry registry) {
        this.webClient = webClientBuilder.baseUrl("http://product-service").build();
        this.circuitBreaker = registry.circuitBreaker("productService");
    }
    
    public Mono<ProductDetails> getProductDetails(String productId) {
        return CircuitBreaker.decorateMonoSupplier(circuitBreaker, 
            () -> webClient.get()
                .uri("/products/{id}", productId)
                .retrieve()
                .bodyToMono(ProductDetails.class)
        ).onErrorResume(ex -> {
            log.warn("Ошибка при получении деталей продукта: {}", ex.getMessage());
            return Mono.just(getFallbackProductDetails(productId));
        });
    }
    
    private ProductDetails getFallbackProductDetails(String productId) {
        // Возвращаем кэшированные данные или заглушку
        return new ProductDetails(productId, "Недоступно", "", 0.0);
    }
}

@Service
public class PaymentProcessor {
    private final PaymentRepository paymentRepository;
    private final ProcessedMessageRepository processedMsgRepository;
    
    @Transactional
    public void processPayment(String messageId, PaymentCommand command) {
        // Проверяем, обрабатывали ли мы это сообщение ранее
        if (processedMsgRepository.existsById(messageId)) {
            log.info("Сообщение {} уже обработано, пропускаем", messageId);
            return;
        }
        
        // Обработка платежа
        Payment payment = new Payment();
        payment.setOrderId(command.getOrderId());
        payment.setAmount(command.getAmount());
        payment.setStatus(PaymentStatus.COMPLETED);
        paymentRepository.save(payment);
        
        // Сохраняем ID сообщения, чтобы избежать повторной обработки
        ProcessedMessage processedMsg = new ProcessedMessage(messageId, LocalDateTime.now());
        processedMsgRepository.save(processedMsg);
    }
}

// Антипаттерн: длинная цепочка синхронных вызовов
public OrderDetails getOrderDetails(String orderId) {
    Order order = orderRepository.findById(orderId);
    Customer customer = customerClient.getCustomer(order.getCustomerId());
    List<Product> products = order.getItems().stream()
        .map(item -> productClient.getProduct(item.getProductId()))
        .collect(Collectors.toList());
    DeliveryInfo delivery = deliveryClient.getDeliveryInfo(order.getDeliveryId());
    
    return new OrderDetails(order, customer, products, delivery);
}

public OrderDetails getOrderDetails(String orderId) {
    Order order = orderRepository.findById(orderId);
    
    // Параллельные запросы
    CompletableFuture<Customer> customerFuture = 
        CompletableFuture.supplyAsync(() -> customerClient.getCustomer(order.getCustomerId()));
    
    CompletableFuture<List<Product>> productsFuture = CompletableFuture.supplyAsync(() -> 
        order.getItems().parallelStream()
            .map(item -> productClient.getProduct(item.getProductId()))
            .collect(Collectors.toList()));
    
    CompletableFuture<DeliveryInfo> deliveryFuture = 
        CompletableFuture.supplyAsync(() -> deliveryClient.getDeliveryInfo(order.getDeliveryId()));
    
    // Ожидание завершения всех запросов
    CompletableFuture.allOf(customerFuture, productsFuture, deliveryFuture).join();
    
    return new OrderDetails(
        order, 
        customerFuture.join(), 
        productsFuture.join(), 
        deliveryFuture.join()
    );
}

// CP-система: блокирующая операция для обеспечения согласованности
@Transactional(isolation = Isolation.SERIALIZABLE)
public void bookSeat(String flightId, String seatId, String passengerId) {
  Seat seat = seatRepository.findById(seatId)
      .orElseThrow(() -> new SeatNotFoundException(seatId));
  
  if (seat.isBooked()) {
    throw new SeatAlreadyBookedException(seatId);
  }
  
  seat.setBooked(true);
  seat.setPassengerId(passengerId);
  seat.setBookingTime(LocalDateTime.now());
  seatRepository.save(seat);
}
 
// AP-система: асинхронное обновление рекомендаций
@KafkaListener(topics = "user-purchases")
public void updateRecommendations(PurchaseEvent event) {
  try {
    recommendationService.updateUserPreferences(
        event.getUserId(), 
        event.getProductId(), 
        event.getAmount()
    );
  } catch (Exception e) {
    // Временная ошибка не критична для бизнеса
    log.warn("Не удалось обновить рекомендации: {}", e.getMessage());
    // Сохраняем событие для повторной обработки позже
    retryQueue.offer(event);
  }
}

// Предзагрузка данных о пользователях в кэш
@Scheduled(fixedRate = 300000) // Каждые 5 минут
public void preloadActiveUsers() {
  List<String> activeUserIds = userActivityService.getRecentlyActiveUserIds();
  List<User> users = userRepository.findAllById(activeUserIds);
  
  users.forEach(user -> 
      cacheManager.getCache("users").put(user.getId(), user)
  );
  
  log.info("Предзагружено {} активных пользователей в кэш", users.size());
}

syntax = "proto3";
 
service OrderService {
  rpc GetOrders (GetOrdersRequest) returns (GetOrdersResponse);
}
 
message GetOrdersRequest {
  string customer_id = 1;
  OrderStatus status = 2;
  int32 page = 3;
  int32 page_size = 4;
}
 
message GetOrdersResponse {
  repeated Order orders = 1;
  int32 total_count = 2;
}
 
message Order {
  string id = 1;
  // ... другие поля
}
 
enum OrderStatus {
  ALL = 0;
  PENDING = 1;
  CONFIRMED = 2;
  SHIPPED = 3;
  DELIVERED = 4;
  CANCELLED = 5;
}

public List<Product> getRecommendedProducts(String userId) {
  try {
    // Пытаемся получить персонализированные рекомендации
    return recommendationClient.getPersonalizedRecommendations(userId)
        .timeout(Duration.ofMillis(300)) // Строгий таймаут
        .toList()
        .block();
  } catch (Exception e) {
    log.warn("Не удалось получить персонализированные рекомендации: {}", e.getMessage());
    // Используем заранее подготовленные популярные товары
    return cachingService.getPopularProducts();
  }
}

@GetMapping("/product/{id}")
public ProductResponse getProduct(@PathVariable String id, 
                                  @RequestParam(defaultValue = "basic") String view) {
  Product product = productRepository.findById(id)
      .orElseThrow(() -> new ProductNotFoundException(id));
  
  // Базовая информация о продукте всегда доступна
  ProductResponse response = new ProductResponse(product);
  
  // Расширенная информация добавляется при нормальной нагрузке
  if ("full".equals(view) && !circuitBreaker.isOpen()) {
    try {
      // Обогащаем дополнительными данными
      response.setReviews(reviewClient.getProductReviews(id));
      response.setRelatedProducts(recommendationClient.getSimilarProducts(id));
    } catch (Exception e) {
      // При ошибке возвращаем хотя бы базовую информацию
      log.warn("Не удалось получить расширенную информацию о продукте: {}", e.getMessage());
    }
  }
  
  return response;
}

public void logEvent(String event, LogLevel level) {
  // Проверяем текущую загрузку системы
  SystemHealth health = healthMonitor.getCurrentHealth();
  
  switch (health) {
    case NORMAL:
      // Логируем все события
      logger.log(level, event);
      break;
    case DEGRADED:
      // Логируем только события уровня WARNING и выше
      if (level.ordinal() >= LogLevel.WARNING.ordinal()) {
        logger.log(level, event);
      }
      break;
    case CRITICAL:
      // Логируем только ERROR и CRITICAL
      if (level.ordinal() >= LogLevel.ERROR.ordinal()) {
        logger.log(level, event);
      }
      break;
  }
}

@Component
public class PrioritizedRateLimiter {
  private final Map<Priority, RateLimiter> limiters = Map.of(
      Priority.HIGH, RateLimiter.create(1000), // 1000 запросов в секунду
      Priority.MEDIUM, RateLimiter.create(200),
      Priority.LOW, RateLimiter.create(50)
  );
  
  public boolean tryAcquire(Priority priority) {
    // При нормальной работе все запросы проходят
    if (systemStatus.isHealthy()) {
      return true;
    }
    
    // При деградации применяем лимиты с учётом приоритета
    return limiters.get(priority).tryAcquire();
  }
}