Шаблоны и протоколы для создания устойчивых микросервисов

HystrixCommand.Setter config = HystrixCommand.Setter
    .withGroupKey(HystrixCommandGroupKey.Factory.asKey("ServiceGroup"))
    .andCommandKey(HystrixCommandKey.Factory.asKey("GetUserData"))
    .andCommandPropertiesDefaults(
        HystrixCommandProperties.Setter()
            .withCircuitBreakerEnabled(true)
            .withCircuitBreakerRequestVolumeThreshold(20)
            .withCircuitBreakerErrorThresholdPercentage(50)
            .withCircuitBreakerSleepWindowInMilliseconds(5000)
    );

CircuitBreakerConfig config = CircuitBreakerConfig.custom()
    .failureRateThreshold(50)
    .slowCallRateThreshold(50)
    .slowCallDurationThreshold(Duration.ofSeconds(2))
    .permittedNumberOfCallsInHalfOpenState(10)
    .slidingWindowSize(100)
    .recordExceptions(IOException.class, TimeoutException.class)
    .build();

public class AccountAggregate {
    private BigDecimal balance;
    private List<DomainEvent> uncommittedEvents = new ArrayList<>();
    
    public static AccountAggregate fromHistory(List<DomainEvent> events) {
        AccountAggregate account = new AccountAggregate();
        events.forEach(account::apply);
        return account;
    }
    
    public void deposit(BigDecimal amount) {
        DomainEvent event = new DepositEvent(UUID.randomUUID(), amount);
        apply(event);
        uncommittedEvents.add(event);
    }
    
    private void apply(DomainEvent event) {
        if (event instanceof DepositEvent) {
            this.balance = this.balance.add(((DepositEvent) event).getAmount());
        }
        // Другие типы событий...
    }
}

@Service
public class OrderService {
    private final EventBus eventBus;
    
    @Transactional
    public void createOrder(Order order) {
        // Сохранение заказа локально
        orderRepository.save(order);
        
        // Публикация события для других сервисов
        eventBus.publish(new OrderCreatedEvent(order.getId(), order.getItems()));
    }
    
    @EventListener
    public void onPaymentFailed(PaymentFailedEvent event) {
        // Компенсирующая транзакция: отмена заказа
        Order order = orderRepository.findById(event.getOrderId());
        order.setStatus(OrderStatus.CANCELLED);
        orderRepository.save(order);
    }
}

@Service
public class OrderOrchestrator {
    private final OrderService orderService;
    private final InventoryService inventoryService;
    private final PaymentService paymentService;
    
    @Transactional
    public void processOrder(Order order) {
        try {
            // Шаг 1: Создание заказа
            orderService.createOrder(order);
            
            // Шаг 2: Резервирование товаров
            inventoryService.reserveItems(order.getItems());
            
            // Шаг 3: Обработка платежа
            paymentService.processPayment(order.getId(), order.getTotalAmount());
            
            // Шаг 4: Подтверждение заказа
            orderService.confirmOrder(order.getId());
        } catch (InventoryException e) {
            // Компенсирующая логика при ошибке резервирования
            orderService.cancelOrder(order.getId());
        } catch (PaymentException e) {
            // Компенсирующая логика при ошибке платежа
            inventoryService.releaseItems(order.getItems());
            orderService.cancelOrder(order.getId());
        }
    }
}

@RestController
@RequestMapping("/api/v1/users")
public class UserControllerV1 {
    @GetMapping("/{id}")
    public UserResponseV1 getUser(@PathVariable Long id) {
        // Реализация первой версии
    }
}
 
@RestController
@RequestMapping("/api/v2/users")
public class UserControllerV2 {
    @GetMapping("/{id}")
    public UserResponseV2 getUser(@PathVariable Long id) {
        // Реализация второй версии с расширенной функциональностью
    }
}

syntax = "proto3";
 
service UserService {
  rpc GetUser (UserRequest) returns (UserResponse);
  rpc ListUsers (UsersRequest) returns (stream UserResponse);
}
 
message UserRequest {
  int32 user_id = 1;
}
 
message UserResponse {
  int32 user_id = 1;
  string name = 2;
  string email = 3;
}

@ServerEndpoint("/notifications/{userId}")
public class NotificationWebSocket {
 
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        // Регистрация сессии для пользователя
    }
 
    @OnMessage
    public void onMessage(String message, Session session) {
        // Обработка входящих сообщений
    }
    
    // Отправка уведомления конкретному пользователю
    public static void sendNotification(String userId, String message) {
        // Логика отправки
    }
}

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection connection = factory.newConnection();
     Channel channel = connection.createChannel()) {
    // Объявление обменника
    channel.exchangeDeclare("orders", "topic");
    // Публикация сообщения
    channel.basicPublish("orders", "order.created", 
                        null, orderJson.getBytes());
}

// Пример подключения и публикации сообщения в MQTT с использованием Eclipse Paho
MqttClient client = new MqttClient("tcp://broker.hivemq.com:1883", MqttClient.generateClientId());
client.connect();
 
MqttMessage message = new MqttMessage("23.5".getBytes());
message.setQos(1); // Уровень гарантии доставки
client.publish("sensors/temperature/living-room", message);

// Пример обработки потока данных с использованием Kafka Streams
StreamsBuilder builder = new StreamsBuilder();
 
// Чтение из входного топика
KStream<String, Order> orderStream = builder.stream("new-orders");
 
// Фильтрация заказов стоимостью выше определённого порога
KStream<String, Order> largeOrders = orderStream
    .filter((key, order) -> order.getTotalAmount() > 1000);
 
// Запись результатов в выходной топик
largeOrders.to("large-orders");

@RestController
public class OrderController {
    @Autowired
    private Tracer tracer;
    
    @PostMapping("/orders")
    public ResponseEntity<Order> createOrder(@RequestBody Order order) {
        Span span = tracer.buildSpan("create-order-process").start();
        try (Scope scope = tracer.scopeManager().activate(span)) {
            // Бизнес-логика обработки заказа
            // ...
            return ResponseEntity.ok(order);
        } finally {
            span.finish();
        }
    }
}

@Service
public class ProductInventoryService {
    @Transactional
    public void updateInventory(String productId, int quantity) {
        // Обновляем инвентарь в базе данных
        inventoryRepository.updateQuantity(productId, quantity);
        
        // Публикуем событие для других сервисов
        InventoryUpdatedEvent event = new InventoryUpdatedEvent(productId, quantity);
        eventPublisher.publish(event);
    }
}

@Transactional
public void createOrder(Order order) {
    // Сохраняем заказ
    Order savedOrder = orderRepository.save(order);
    
    // В рамках той же транзакции сохраняем событие в outbox
    OutboxEvent event = new OutboxEvent(
        UUID.randomUUID(), 
        "OrderCreated", 
        objectMapper.writeValueAsString(savedOrder)
    );
    outboxRepository.save(event);
}
 
// В отдельном потоке/процессе
@Scheduled(fixedRate = 5000)
public void processOutbox() {
    List<OutboxEvent> events = outboxRepository.findUnprocessed(100);
    for (OutboxEvent event : events) {
        eventBus.publish(event.getType(), event.getPayload());
        event.markProcessed();
        outboxRepository.save(event);
    }
}

@Transactional
public void processPayment(PaymentEvent event) {
    // Проверяем, не обработано ли уже это сообщение
    if (processedEventsRepository.existsById(event.getEventId())) {
        log.info("Event {} already processed, skipping", event.getEventId());
        return;
    }
    
    // Обработка платежа
    Payment payment = paymentRepository.findById(event.getPaymentId())
        .orElseThrow(() -> new NotFoundException("Payment not found"));
    
    if (payment.getStatus() == PaymentStatus.PENDING) {
        payment.setStatus(PaymentStatus.COMPLETED);
        paymentRepository.save(payment);
    }
    
    // Сохраняем идентификатор обработанного события
    processedEventsRepository.save(new ProcessedEvent(event.getEventId()));
}

@Service
public class ProductService {
  @Autowired
  private ProductRepository repository;
  
  @Cacheable(value = "products", key = "#id", unless = "#result == null")
  public Product getProduct(Long id) {
      return repository.findById(id).orElse(null);
  }
  
  @CachePut(value = "products", key = "#product.id")
  public Product updateProduct(Product product) {
      return repository.save(product);
  }
  
  @CacheEvict(value = "products", key = "#id")
  public void deleteProduct(Long id) {
      repository.deleteById(id);
  }
}

@Component
public class RateLimiterInterceptor implements HandlerInterceptor {
  private final RateLimiter rateLimiter = RateLimiter.create(100.0); // 100 запросов в секунду
  
  @Override
  public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
      if (!rateLimiter.tryAcquire()) {
          response.setStatus(429); // Too Many Requests
          return false;
      }
      return true;
  }
}

public Response getProductWithRecommendations(Long productId) {
  Product product = productService.getProduct(productId);
  
  List<Product> recommendations;
  try {
      // Пытаемся получить рекомендации с таймаутом
      recommendations = recommendationService.getRecommendations(productId, 300);
  } catch (TimeoutException e) {
      // При таймауте возвращаем пустой список рекомендаций
      log.warn("Recommendation service timed out, skipping recommendations");
      recommendations = Collections.emptyList();
  }
  
  return new Response(product, recommendations);
}

@KafkaListener(topics = "order-processing")
public void processOrder(OrderEvent orderEvent) {
  try {
      // Обработка заказа
      orderProcessingService.process(orderEvent.getOrderId());
  } catch (Exception e) {
      // Логирование ошибки
      log.error("Error processing order", e);
      
      // Повторная отправка в очередь с задержкой
      // при транзиентных ошибках
      if (isTransientError(e)) {
          kafkaTemplate.send("order-processing-retry", orderEvent);
      } else {
          // Для постоянных ошибок - отправка в очередь DLQ
          kafkaTemplate.send("order-processing-failed", orderEvent);
      }
  }
}

@Scheduled(fixedRate = 60000)
public void scheduledTask() {
  RLock lock = redissonClient.getLock("scheduledTask");
  
  try {
      // Пытаемся получить блокировку на 30 секунд
      boolean acquired = lock.tryLock(5, 30, TimeUnit.SECONDS);
      if (acquired) {
          try {
              // Выполняем задачу, зная что никакой другой экземпляр
              // сервиса её не выполняет параллельно
              processScheduledTask();
          } finally {
              lock.unlock();
          }
      }
  } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
  }
}

// Пример структуры проекта, отражающей бизнес-домены
com.company.orders        // Микросервис заказов
com.company.catalog       // Микросервис каталога товаров
com.company.users         // Микросервис пользователей
com.company.payments      // Микросервис платежей

// Пример маршрутизации в Spring Cloud Gateway
@Bean
public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
    return builder.routes()
        .route("user-service", r -> r.path("/users/**")
            .uri("lb://user-service"))
        .route("order-service", r -> r.path("/orders/**")
            .uri("lb://order-service"))
        .route("product-service", r -> r.path("/products/[B]")
            .uri("lb://product-service"))
        .build();
}

// Пример использования разных типов БД для разных микросервисов
 
// Сервис каталога: MongoDB для хранения структурированных данных с гибкой схемой
@Document(collection = "products")
public class Product {
    @Id
    private String id;
    private String name;
    private Map<String, Object> attributes; // Гибкие атрибуты продукта
}
 
// Сервис заказов: PostgreSQL для транзакционных данных
@Entity
@Table(name = "orders")
public class Order {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @ManyToOne
    @JoinColumn(name = "customer_id")
    private Customer customer;
    
    @OneToMany(cascade = CascadeType.ALL)
    @JoinColumn(name = "order_id")
    private List<OrderItem> items;
}

// Пример реализации Circuit Breaker с Resilience4j
@Bean
public CircuitBreaker orderServiceCircuitBreaker() {
    CircuitBreakerConfig config = CircuitBreakerConfig.custom()
        .failureRateThreshold(50)      // 50% ошибок
        .slidingWindowSize(10)         // на последних 10 запросах
        .minimumNumberOfCalls(5)       // минимум 5 запросов для принятия решения
        .waitDurationInOpenState(Duration.ofSeconds(30)) // 30 секунд в открытом состоянии
        .permittedNumberOfCallsInHalfOpenState(3)  // 3 тестовых запроса в полуоткрытом состоянии
        .automaticTransitionFromOpenToHalfOpenEnabled(true)
        .build();
    
    return CircuitBreaker.of("orderService", config);
}

// Пример добавления трассировки запросов с Spring Cloud Sleuth
@RestController
public class OrderController {
    private static final Logger log = LoggerFactory.getLogger(OrderController.class);
    
    @Autowired
    private OrderService orderService;
    
    @PostMapping("/orders")
    public ResponseEntity<Order> createOrder(@RequestBody OrderRequest request) {
        log.info("Received order creation request: {}", request.getOrderReference());
        Order order = orderService.createOrder(request);
        log.info("Order created successfully: {}", order.getId());
        return ResponseEntity.ok(order);
    }
}

// Пример защиты API с OAuth2 в Spring Security
@Configuration
@EnableWebSecurity
public class SecurityConfig extends WebSecurityConfigurerAdapter {
    
    @Override
    protected void configure(HttpSecurity http) throws Exception {
        http
            .authorizeRequests()
                .antMatchers("/api/public/[B]").permitAll()
                .antMatchers("/api/admin/[/B]").hasRole("ADMIN")
                .anyRequest().authenticated()
            .and()
            .oauth2ResourceServer()
                .jwt();
    }
}

// Пример проверки разрешений с использованием Spring Security и JWT
@PreAuthorize("hasAuthority('EDIT_ORDERS') and @orderSecurity.isOrderOwner(authentication, #orderId)")
public Order updateOrder(Long orderId, OrderUpdateRequest request) {
    // Бизнес-логика обновления заказа
}
 
@Component
public class OrderSecurity {
    @Autowired
    private OrderRepository orderRepository;
    
    public boolean isOrderOwner(Authentication authentication, Long orderId) {
        String currentUserId = ((JwtAuthenticationToken) authentication).getToken()
            .getClaimAsString("sub");
        
        Order order = orderRepository.findById(orderId).orElse(null);
        return order != null && order.getCustomerId().equals(currentUserId);
    }
}

// Пример аудита действий с использованием AOP
@Aspect
@Component
public class AuditAspect {
    private static final Logger auditLogger = LoggerFactory.getLogger("AUDIT");
    
    @Autowired
    private ObjectMapper objectMapper;
    
    @Around("@annotation(com.company.audit.Auditable)")
    public Object auditMethod(ProceedingJoinPoint joinPoint) throws Throwable {
        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
        String username = auth != null ? auth.getName() : "anonymous";
        
        String methodName = joinPoint.getSignature().getName();
        String className = joinPoint.getSignature().getDeclaringTypeName();
        
        // Маскирование чувствительных параметров
        Object[] sanitizedArgs = sanitizeArgs(joinPoint.getArgs());
        
        auditLogger.info("User '{}' called method: {}.{} with args: {}", 
                username, className, methodName, objectMapper.writeValueAsString(sanitizedArgs));
        
        Object result = joinPoint.proceed();
        
        auditLogger.info("Method {}.{} completed for user '{}'", 
                className, methodName, username);
        
        return result;
    }
    
    private Object[] sanitizeArgs(Object[] args) {
        // Логика маскирования чувствительных данных
        // ...
        return args;
    }
}

// Пример структуры монорепозитория, отражающей команды и их сервисы
/
├── platform/                  // Платформенная команда
│   ├── api-gateway/           // API Gateway
│   ├── service-registry/      // Сервис обнаружения
│   └── auth-service/          // Сервис аутентификации
│
├── catalog-team/              // Команда каталога товаров
│   ├── product-service/       // Микросервис продуктов
│   └── category-service/      // Микросервис категорий
│
├── order-team/                // Команда заказов
│   ├── order-service/         // Микросервис заказов
│   └── shipment-service/      // Микросервис доставки
│
└── user-team/                 // Команда пользователей
    ├── user-service/          // Микросервис пользователей
    └── preference-service/    // Микросервис предпочтений