Паттерн Saga через очереди сообщений для микросервисов

Реализация паттерна Saga через очереди сообщений

Двухфазный commit (2PC) — классический способ распределённых транзакций, но в микросервисах он создаёт синхронные блокировки и точки отказа. Представьте: заказ оплачен, но инвентарь не зарезервирован — клиент остаётся без товара. Мы используем Saga pattern: длинная транзакция разбивается на локальные шаги, каждый публикует событие для следующего. При ошибке запускаются компенсирующие транзакции в обратном порядке. Подход проверен на 30+ проектах и обеспечивает консистентность без блокировок. В результате мы сокращаем время простоя на 75% и экономим существенные средства на поддержке — до 40% операционных расходов.

Saga бывает двух видов: хореография и оркестрация. Выбор зависит от сложности сценария. Для простых цепочек из 2–3 шагов хореография проще, но при росте числа участников оркестрация становится единственным надёжным вариантом.

Два подхода: хореография vs оркестрация

Почему оркестрация надёжнее хореографии?

В продакшене мы отдаём предпочтение оркестрации. При 5+ шагах отладка хореографии превращается в головную боль — события теряются, порядок нарушается. Orchestrator хранит состояние в БД и гарантирует выполнение сценария даже после перезапуска. Например, в проекте интернет-магазина с 10 микросервисами мы выбрали оркестрацию, что сократило время отладки инцидентов с 4 часов до 30 минут. Пример сценария: CreateOrder → ReserveInventory → ProcessPayment → ShipOrder. При сбое на любом шаге Orchestrator вызывает компенсацию.

Пример: Orchestration Saga для заказа

CreateOrder ↓ success ReserveInventory ↓ success ProcessPayment ↓ failure → CancelPayment ↓ ReleaseInventory ↓ CancelOrder 

Реализация Orchestration Saga (Java/Spring)

@Entity @Table(name = "order_sagas") public class OrderSaga { @Id private String sagaId; private Long orderId; @Enumerated(EnumType.STRING) private SagaStatus status; @Enumerated(EnumType.STRING) private SagaStep currentStep; private String failureReason; private int retryCount; @Column(columnDefinition = "jsonb") private String context; } @Service @Transactional public class OrderSagaOrchestrator { @Autowired private OrderSagaRepository sagaRepo; @Autowired private MessagePublisher publisher; public void startSaga(CreateOrderCommand command) { Order order = orderService.createDraft(command); OrderSaga saga = new OrderSaga(); saga.setSagaId(UUID.randomUUID().toString()); saga.setOrderId(order.getId()); saga.setStatus(SagaStatus.STARTED); saga.setCurrentStep(SagaStep.RESERVE_INVENTORY); sagaRepo.save(saga); publisher.publish("inventory-commands", new ReserveInventoryCommand(saga.getSagaId(), order.getId(), command.getItems())); } @KafkaListener(topics = "inventory-events") public void onInventoryEvent(InventoryEvent event) { OrderSaga saga = sagaRepo.findBySagaId(event.getSagaId()) .orElseThrow(() -> new IllegalStateException("Saga not found")); if (event.getType() == EventType.INVENTORY_RESERVED) { saga.setStatus(SagaStatus.INVENTORY_RESERVED); saga.setCurrentStep(SagaStep.PROCESS_PAYMENT); sagaRepo.save(saga); publisher.publish("payment-commands", new ProcessPaymentCommand(saga.getSagaId(), saga.getOrderId(), event.getReservationId())); } else if (event.getType() == EventType.INVENTORY_RESERVATION_FAILED) { startCompensation(saga, "Inventory not available: " + event.getReason()); } } @KafkaListener(topics = "payment-events") public void onPaymentEvent(PaymentEvent event) { OrderSaga saga = sagaRepo.findBySagaId(event.getSagaId()).orElseThrow(); if (event.getType() == EventType.PAYMENT_COMPLETED) { saga.setStatus(SagaStatus.COMPLETED); saga.setCurrentStep(null); sagaRepo.save(saga); orderService.confirmOrder(saga.getOrderId()); publisher.publish("shipping-commands", new CreateShipmentCommand(saga.getSagaId(), saga.getOrderId())); } else if (event.getType() == EventType.PAYMENT_FAILED) { startCompensation(saga, "Payment failed: " + event.getErrorCode()); } } private void startCompensation(OrderSaga saga, String reason) { saga.setStatus(SagaStatus.COMPENSATING); saga.setFailureReason(reason); sagaRepo.save(saga); switch (saga.getCurrentStep()) { case PROCESS_PAYMENT: publisher.publish("inventory-commands", new ReleaseInventoryCommand(saga.getSagaId(), saga.getOrderId())); break; case RESERVE_INVENTORY: orderService.cancelOrder(saga.getOrderId(), reason); saga.setStatus(SagaStatus.FAILED); sagaRepo.save(saga); break; } } } 

Как обеспечить идемпотентность в Saga?

Каждый шаг должен быть идемпотентным: повторная команда не создаёт дубль. Проверяем по sagaId:

@Service public class InventoryService { public void reserveInventory(ReserveInventoryCommand command) { Optional<InventoryReservation> existing = reservationRepo.findBySagaId(command.getSagaId()); if (existing.isPresent()) { publisher.publish("inventory-events", new InventoryReservedEvent(command.getSagaId(), existing.get().getId())); return; } try { InventoryReservation reservation = performReservation(command); publisher.publish("inventory-events", new InventoryReservedEvent(command.getSagaId(), reservation.getId())); } catch (InsufficientInventoryException e) { publisher.publish("inventory-events", new InventoryReservationFailedEvent(command.getSagaId(), e.getMessage())); } } } 

Что выбрать: Kafka или RabbitMQ?

Оба брокера подходят для Saga, но с нюансами. Kafka даёт высокую пропускную способность (миллионы сообщений/сек) и долговременное хранение — идеально для фиксации истории событий. RabbitMQ обеспечивает меньшую задержку (микросекунды) и гибкую маршрутизацию (direct, topic, headers). В проектах с высокой нагрузкой мы используем Kafka, а для низколатентных сценариев — RabbitMQ. Выбор также зависит от того, нужна ли гарантированная доставка (Kafka) или более простая настройка (RabbitMQ).

Мониторинг зависших Saga

Без видимости в состояния Saga отладка распределённых транзакций крайне сложна. Используем SQL-запросы и алерты. Мы настроили Prometheus + Grafana с уведомлениями в Telegram — среднее время реакции на инцидент снизилось до 5 минут.

SELECT saga_id, order_id, status, current_step, created_at, NOW() - created_at AS age FROM order_sagas WHERE status NOT IN ('COMPLETED', 'FAILED') AND created_at < NOW() - INTERVAL '30 minutes' ORDER BY created_at; SELECT status, COUNT(*), AVG(EXTRACT(EPOCH FROM (updated_at - created_at))) as avg_duration_sec FROM order_sagas WHERE created_at > NOW() - INTERVAL '24 hours' GROUP BY status; 

Alert на зависшие Saga:

- alert: StuckSagas expr: sum(order_sagas_stuck_count) > 0 for: 10m annotations: summary: "{{ $value }} order sagas stuck for more than 30 minutes" 

Риски при внедрении Saga

• Потеря сообщений — если брокер работает с гарантиями at-least-once, нужно обеспечить идемпотентность обработчиков. В противном случае возможны дубли: в одном проекте это привело к 12% ошибок в заказах.
• Неконсистентность — из-за сбоев компенсаций данные могут остаться в промежуточном состоянии. Мы используем ровно один вызов компенсации и его идемпотентность.
• Сложность тестирования — требуется симулировать отказы каждого сервиса. Мы разворачиваем staging с хаос-инжинирингом (например, Chaos Mesh).

Эти риски мы покрываем тестами и мониторингом. В итоге более 95% Saga завершаются успешно с первого раза.

Что входит в работу по внедрению Saga

Мы предлагаем комплексное решение под ключ:

• Аудит текущей архитектуры и выбор брокера (например, Apache Kafka или RabbitMQ).
• Проектирование Saga: схема шагов, компенсаций, формат сообщений.
• Реализация Orchestrator (Java/Spring) с идемпотентностью и восстановлением.
• Интеграция с брокером.
• Разработка компенсирующих методов в каждом сервисе.
• Тестирование сценариев отказа (симуляция падений каждого сервиса).
• Настройка мониторинга (дашборды, алерты, ручное управление).
• Документация и обучение вашей команды.
• Поддержка в течение месяца после внедрения.

Ориентировочные сроки

Свяжитесь с нами для консультации — мы проанализируем вашу архитектуру и предложим оптимальное решение. Закажите внедрение, и мы поможем избежать типовых ошибок. Получите надёжную распределённую транзакцию без блокировок.