Приведи пример асинхронной коммуникации

Асинхронная коммуникация в Java: практические примеры

Асинхронная коммуникация — это паттерн, где отправитель не ждёт ответа получателя. Это критически важно для масштабируемых систем, которые должны обрабатывать высокие нагрузки без блокировки потоков.

Что такое асинхронная коммуникация?

Синхронная (блокирующая):

Клиент → Сервер (ждёт) → Ответ

Асинхронная (неблокирующая):

Клиент → Message Broker → Сервер (получает когда готов)
Клиент продолжает работу

Пример 1: Message Queue (RabbitMQ)

Это классический пример асинхронной коммуникации для обработки задач в фоне:

// Sender: отправляет сообщение в очередь
@Service
public class OrderService {
    private final RabbitTemplate rabbitTemplate;
    
    public OrderService(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public void createOrder(OrderRequest request) {
        // Сохраняем заказ
        Order order = Order.builder()
            .id(UUID.randomUUID())
            .customerId(request.getCustomerId())
            .status("PENDING")
            .createdAt(LocalDateTime.now())
            .build();
        
        orderRepository.save(order);
        
        // Отправляем событие асинхронно БЕЗ ожидания
        // Клиент сразу получает response, а обработка идёт в фоне
        rabbitTemplate.convertAndSend(
            "orders.exchange",
            "order.created",
            new OrderCreatedEvent(order.getId(), order.getCustomerId())
        );
        
        // Ответ клиенту приходит ТУТ, а не после обработки
    }
}

// Receiver: слушает очередь и обрабатывает
@Component
public class OrderProcessingListener {
    private final EmailService emailService;
    private final InventoryService inventoryService;
    
    @RabbitListener(queues = "orders.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        try {
            // Долгая операция: отправка письма
            emailService.sendOrderConfirmation(event.getOrderId());
            
            // Долгая операция: изменение инвентаря
            inventoryService.reserveItems(event.getOrderId());
            
            // Это всё выполняется асинхронно, не блокируя клиента
        } catch (Exception e) {
            // Обработка ошибок, retry логика
            log.error("Failed to process order: {}", event.getOrderId(), e);
        }
    }
}

Преимущества:

• Клиент получает быстрый ответ (200 OK сразу)
• Обработка идёт параллельно в другом процессе
• Масштабируется: можно добавить N workers
• Декаплинг: сервисы не знают друг о друге

Пример 2: Callback / Webhook

Внешний сервис уведомляет нас, когда он закончил работу:

// Инициатор: платёжный сервис
@Service
public class PaymentService {
    private final RestTemplate restTemplate;
    
    public void initiatePayment(PaymentRequest request) {
        // Отправляем запрос на платёж с callback URL
        String callbackUrl = "https://ourapp.com/api/v1/webhooks/payment-completed";
        
        PaymentInitiation payment = PaymentInitiation.builder()
            .amount(request.getAmount())
            .currency("USD")
            .callbackUrl(callbackUrl)  // Платёжный сервис отправит сюда результат
            .metadata(request.getOrderId())
            .build();
        
        // Отправляем запрос и СРАЗУ возвращаем
        // Ответ придёт позже на callbackUrl
        restTemplate.postForObject(
            "https://payment-provider.com/api/payments",
            payment,
            PaymentResponse.class
        );
    }
}

// Receiver: получает callback после завершения платежа
@RestController
@RequestMapping("/api/v1/webhooks")
public class WebhookController {
    private final PaymentConfirmationService confirmationService;
    
    @PostMapping("/payment-completed")
    public ResponseEntity<Void> handlePaymentCompleted(
        @RequestBody PaymentWebhook webhook,
        @RequestHeader("X-Signature") String signature
    ) {
        // Проверяем подпись для безопасности
        if (!verifySignature(webhook, signature)) {
            return ResponseEntity.status(HttpStatus.UNAUTHORIZED).build();
        }
        
        // Обрабатываем результат платежа
        if (webhook.getStatus().equals("SUCCESS")) {
            confirmationService.confirmPayment(webhook.getOrderId());
        } else {
            confirmationService.failPayment(webhook.getOrderId());
        }
        
        return ResponseEntity.ok().build();
    }
}

Пример 3: Event-driven Architecture (Spring Events)

Внутренняя асинхронная коммуникация между компонентами:

// Event
public class UserRegisteredEvent extends ApplicationEvent {
    private final String userId;
    private final String email;
    
    public UserRegisteredEvent(Object source, String userId, String email) {
        super(source);
        this.userId = userId;
        this.email = email;
    }
}

// Publisher: отправляет событие
@Service
public class AuthService {
    private final ApplicationEventPublisher eventPublisher;
    
    public void registerUser(UserRegistrationRequest request) {
        User user = User.builder()
            .email(request.getEmail())
            .password(passwordEncoder.encode(request.getPassword()))
            .createdAt(LocalDateTime.now())
            .build();
        
        userRepository.save(user);
        
        // Отправляем событие БЕЗ ожидания обработчиков
        eventPublisher.publishEvent(
            new UserRegisteredEvent(this, user.getId(), user.getEmail())
        );
    }
}

// Listener 1: отправляет приветствие
@Component
public class WelcomeEmailListener {
    private final EmailService emailService;
    
    @EventListener
    @Async  // Выполняется в отдельном потоке (нужна @EnableAsync)
    public void sendWelcomeEmail(UserRegisteredEvent event) {
        emailService.sendWelcomeEmail(event.getEmail());
    }
}

// Listener 2: добавляет пользователя в рассылку
@Component
public class MailingListListener {
    private final MailingListService mailingListService;
    
    @EventListener
    @Async
    public void addToMailingList(UserRegisteredEvent event) {
        mailingListService.subscribe(event.getEmail());
    }
}

Пример 4: Reactive (Project Reactor / WebFlux)

Высокопроизводительная асинхронная обработка:

@RestController
@RequestMapping("/api/v1/data")
public class ReactiveController {
    private final DataService dataService;
    
    @GetMapping("/{id}")
    public Mono<DataResponse> getData(@PathVariable String id) {
        return dataService.fetchDataAsync(id)
            .map(data -> DataResponse.builder()
                .id(data.getId())
                .content(data.getContent())
                .timestamp(LocalDateTime.now())
                .build()
            )
            .doOnError(error -> log.error("Error fetching data", error))
            .timeout(Duration.ofSeconds(5));
    }
    
    @GetMapping("/batch")
    public Flux<DataResponse> getBatchData() {
        // Параллельная обработка множества элементов
        return Flux.range(1, 100)
            .flatMap(id -> dataService.fetchDataAsync(String.valueOf(id)))
            .map(data -> DataResponse.from(data))
            .parallel()
            .runOn(Schedulers.parallel())
            .sequential();
    }
}

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

Важные моменты

• Гарантии доставки: Message Queue обеспечивает, что сообщение будет обработано
• Retry логика: важна для обработки временных сбоев
• Идемпотентность: операция должна быть безопасной при повторном выполнении
• Monitoring: нужно отслеживать длину очереди и время обработки