Обеспечивалась ли какая-либо транзакционность сообщений в текущем проекте

Транзакционность сообщений в микросервисной архитектуре

Это важный вопрос, потому что асинхронная обработка сообщений одна из самых сложных проблем в распределённых системах. Расскажу о моем опыте реализации различных уровней гарантий доставки и обработки.

Уровни гарантий доставки (Delivery Guarantees)

В текущих проектах мы реализовали:

1. At-Least-Once (по крайней мере один раз):

   • Сообщение обязательно будет обработано
   • Но может быть обработано несколько раз
   • Требует идемпотентной обработки

2. At-Most-Once (максимум один раз):

   • Сообщение либо будет обработано, либо потеряно
   • Может быть недостаточно для критичных операций

3. Exactly-Once (ровно один раз):

   • Идеальный сценарий, но сложный в реализации
   • Требует координации между producer и consumer

Пример с Kafka и Spring Boot

// Producer: гарантия того, что сообщение отправлено
@Configuration
public class KafkaProducerConfig {
    @Bean
    public ProducerFactory<String, OrderEvent> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.ACKS_CONFIG, "all");  // Wait for all replicas
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);  // Exactly-once
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-producer-1");
        return new DefaultProducerFactory<>(props);
    }
}

// Consumer: обработка с гарантией
@KafkaListener(topics = "orders", groupId = "order-processor")
public void processOrder(OrderEvent event) {
    try {
        orderService.processOrder(event);
    } catch (Exception e) {
        // Сообщение вернётся в очередь (Kafka retry policy)
        throw e;
    }
}

Идемпотентная обработка

Ключевой паттерн для At-Least-Once:

public class IdempotentOrderProcessor {
    
    @Transactional
    public void processOrder(OrderEvent event) {
        // Проверяем, не обработано ли это сообщение ранее
        ProcessedEvent processed = processedEventRepo.findByEventId(event.getId());
        if (processed != null) {
            return;  // Уже обработано, пропускаем
        }
        
        // Основная логика
        Order order = new Order(event);
        orderRepository.save(order);
        
        // Записываем факт обработки
        processedEventRepo.save(new ProcessedEvent(event.getId()));
    }
}

Distributed Transactions (Saga Pattern)

Для операций, охватывающих несколько сервисов:

// Хореография: каждый сервис слушает события предыдущего
@KafkaListener(topics = "payment.completed")
public void handlePaymentCompleted(PaymentEvent event) {
    shipmentService.createShipment(event.getOrderId());
    kafkaTemplate.send("shipment.created", new ShipmentEvent(...));
}

// Оркестрация: центральный сервис координирует процесс
@Service
public class OrderSaga {
    @Transactional
    public void executeOrderSaga(OrderCommand command) {
        // Шаг 1: создать заказ
        Order order = orderService.create(command);
        
        // Шаг 2: зарезервировать платёж
        boolean paymentReserved = paymentService.reserve(order.getId());
        if (!paymentReserved) {
            orderService.cancel(order.getId());
            return;
        }
        
        // Шаг 3: создать отправку
        boolean shipmentCreated = shipmentService.create(order.getId());
        if (!shipmentCreated) {
            paymentService.cancelReservation(order.getId());
            orderService.cancel(order.getId());
        }
    }
}

Handling Dead Letter Queues (DLQ)

Для сообщений, которые не удалось обработать:

@KafkaListener(topics = "orders-dlq")
public void handleDlq(OrderEvent event) {
    log.error("Message failed processing: {}", event.getId());
    
    // Либо уведомляем администраторов
    alertService.sendAlert("Order processing failed", event);
    
    // Либо сохраняем для ручной обработки
    failedMessageRepository.save(new FailedMessage(event));
}

Выводы

Ключевые практики:

• Используем At-Least-Once гарантии (проще и надёжнее)
• Всегда реализуем идемпотентность
• Для сложных операций применяем Saga pattern
• Отслеживаем обработанные события в БД
• Имеем DLQ для неудачных сообщений
• Логируем и мониторим всё с помощью ELK стека

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