Как создать асинхронную микросервисную архитектуру?

Асинхронная микросервисная архитектура в Java

Асинхронная архитектура — это парадигма, где сервисы не ждут синхронных ответов друг от друга, а взаимодействуют через очереди сообщений, события и callback'и. Это обеспечивает масштабируемость и отказоустойчивость.

Архитектурные паттерны

1. Event-Driven Architecture (событийная архитектура)

// Модель события
public class OrderCreatedEvent {
    private String orderId;
    private String customerId;
    private BigDecimal amount;
    private LocalDateTime timestamp;
    
    public OrderCreatedEvent(String orderId, String customerId, BigDecimal amount) {
        this.orderId = orderId;
        this.customerId = customerId;
        this.amount = amount;
        this.timestamp = LocalDateTime.now();
    }
    
    // getters
}

// Издатель события (Order Service)
@Service
public class OrderService {
    private final ApplicationEventPublisher eventPublisher;
    
    public OrderService(ApplicationEventPublisher eventPublisher) {
        this.eventPublisher = eventPublisher;
    }
    
    @Transactional
    public void createOrder(OrderRequest request) {
        // Создаем заказ
        Order order = new Order(request);
        // Сохраняем в БД
        // ...
        
        // Публикуем событие (асинхронно!)
        eventPublisher.publishEvent(
            new OrderCreatedEvent(order.getId(), request.getCustomerId(), request.getAmount())
        );
    }
}

// Слушатель события (Email Service)
@Component
public class EmailNotificationListener {
    private final EmailService emailService;
    
    public EmailNotificationListener(EmailService emailService) {
        this.emailService = emailService;
    }
    
    @EventListener
    @Async
    public void handleOrderCreated(OrderCreatedEvent event) {
        // Асинхронная отправка письма
        emailService.sendOrderConfirmation(event.getCustomerId());
    }
}

// Слушатель события (Payment Service)
@Component
public class PaymentProcessListener {
    private final PaymentService paymentService;
    
    public PaymentProcessListener(PaymentService paymentService) {
        this.paymentService = paymentService;
    }
    
    @EventListener
    @Async
    public void handleOrderCreated(OrderCreatedEvent event) {
        // Асинхронная обработка платежа
        paymentService.processPayment(event.getOrderId(), event.getAmount());
    }
}

// Конфигурация для асинхронности
@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
    
    @Bean(name = "taskExecutor")
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("async-");
        executor.initialize();
        return executor;
    }
}

Message Queue паттерн (RabbitMQ, Kafka)

Вариант 1: RabbitMQ

// Конфигурация RabbitMQ
@Configuration
public class RabbitMQConfig {
    
    public static final String ORDER_QUEUE = "order.queue";
    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_ROUTING_KEY = "order.created";
    
    @Bean
    public Queue orderQueue() {
        return new Queue(ORDER_QUEUE, true); // durable queue
    }
    
    @Bean
    public TopicExchange orderExchange() {
        return new TopicExchange(ORDER_EXCHANGE, true, false);
    }
    
    @Bean
    public Binding orderBinding(Queue orderQueue, TopicExchange orderExchange) {
        return BindingBuilder.bind(orderQueue)
            .to(orderExchange)
            .with(ORDER_ROUTING_KEY);
    }
}

// Издатель (Producer)
@Service
public class OrderProducerService {
    private final RabbitTemplate rabbitTemplate;
    
    public OrderProducerService(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public void publishOrderCreatedEvent(Order order) {
        OrderEventMessage message = new OrderEventMessage(
            order.getId(),
            order.getCustomerId(),
            order.getAmount()
        );
        
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            message
        );
    }
}

// Потребитель (Consumer) - Email Service
@Service
public class EmailServiceConsumer {
    private final EmailService emailService;
    
    public EmailServiceConsumer(EmailService emailService) {
        this.emailService = emailService;
    }
    
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void processOrderEvent(OrderEventMessage message) {
        try {
            emailService.sendOrderConfirmation(message.getCustomerId());
        } catch (Exception e) {
            // Обработка ошибки
            // Dead Letter Queue
        }
    }
}

Вариант 2: Apache Kafka

// Конфигурация Kafka Producer
@Configuration
public class KafkaProducerConfig {
    
    @Bean
    public ProducerFactory<String, OrderEventMessage> 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, OrderEventMessage> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

// Kafka Producer
@Service
public class OrderKafkaProducer {
    private final KafkaTemplate<String, OrderEventMessage> kafkaTemplate;
    private static final String TOPIC = "order-events";
    
    public OrderKafkaProducer(KafkaTemplate<String, OrderEventMessage> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    
    public void publishOrderEvent(Order order) {
        OrderEventMessage message = new OrderEventMessage(
            order.getId(),
            order.getCustomerId(),
            order.getAmount()
        );
        
        kafkaTemplate.send(TOPIC, order.getId(), message)
            .addCallback(
                result -> {
                    System.out.println("Sent message=[" + message + "]");
                    System.out.println("with offset=[" + result.getRecordMetadata().offset() + "]");
                },
                ex -> System.out.println("Failed to send message=[" + message + "]", ex)
            );
    }
}

// Kafka Consumer
@Service
public class OrderKafkaConsumer {
    private final EmailService emailService;
    
    public OrderKafkaConsumer(EmailService emailService) {
        this.emailService = emailService;
    }
    
    @KafkaListener(topics = "order-events", groupId = "email-service-group")
    public void consumeOrderEvent(OrderEventMessage message) {
        try {
            emailService.sendOrderConfirmation(message.getCustomerId());
        } catch (Exception e) {
            // Логирование и обработка ошибки
            // Можно использовать @KafkaListener с errorHandler
        }
    }
}

// Consumer с обработкой ошибок
@Configuration
public class KafkaConsumerConfig {
    
    @Bean
    public ConsumerFactory<String, OrderEventMessage> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, OrderEventMessage.class.getName());
        return new DefaultConsumerFactory<>(props);
    }
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEventMessage> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, OrderEventMessage> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

Паттерн Request-Reply (асинхронный RPC)

// Клиент отправляет запрос и получает результат асинхронно
@Service
public class OrderServiceClient {
    private final RabbitTemplate rabbitTemplate;
    
    public void createOrderAsync(OrderRequest request) {
        // Создаем уникальный correlation ID для ответа
        String correlationId = UUID.randomUUID().toString();
        
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            new OrderRequest(request, correlationId),
            message -> {
                message.getMessageProperties().setCorrelationId(correlationId);
                message.getMessageProperties().setReplyTo("order-response-queue");
                return message;
            }
        );
    }
    
    @RabbitListener(queues = "order-response-queue")
    public void handleOrderResponse(OrderResponse response, 
                                   @Header("amqp_correlationId") String correlationId) {
        System.out.println("Order created: " + response.getOrderId());
    }
}

Saga паттерн для распределённых транзакций

// Orestra-based Saga для управления транзакциями между сервисами
@Component
public class OrderSaga {
    private final OrderService orderService;
    private final PaymentService paymentService;
    private final InventoryService inventoryService;
    
    @Transactional
    public void executeOrderSaga(OrderRequest request) {
        try {
            // Шаг 1: Создаем заказ
            Order order = orderService.createOrder(request);
            
            // Шаг 2: Зарезервируем товар
            inventoryService.reserveInventory(order.getId(), request.getItems());
            
            // Шаг 3: Обрабатываем платёж
            paymentService.processPayment(order.getId(), order.getAmount());
            
            // Успех
            orderService.markOrderAsConfirmed(order.getId());
            
        } catch (PaymentException e) {
            // Компенсирующие транзакции (откатываем шаги в обратном порядке)
            inventoryService.releaseInventory(order.getId());
            orderService.cancelOrder(order.getId());
            throw e;
        } catch (InventoryException e) {
            orderService.cancelOrder(order.getId());
            throw e;
        }
    }
}

Best Practices для асинхронной архитектуры

1. Используй идемпотентность — сообщение может быть обработано дважды

@Service
public class IdempotentOrderProcessor {
    private final Set<String> processedEvents = new ConcurrentHashSet<>();
    
    public void processOrder(OrderEventMessage message) {
        if (!processedEvents.add(message.getEventId())) {
            return; // Уже обработано
        }
        // Обработка...
    }
}

1. Реализуй Dead Letter Queue для неудачных сообщений
2. Используй Circuit Breaker паттерн для отказоустойчивости
3. Мониторь задержку между событиями
4. Версионируй события для обратной совместимости

Сравнение подходов

Асинхронная архитектура — это ключ к масштабируемым и отказоустойчивым микросервисам в Java.