spring-boot-saga-pattern

Spring Boot Saga Pattern

概要

Saga パターンを使用して、マイクロサービス間の分散トランザクションを実装します。2 フェーズコミットを、ローカルトランザクションと補償アクションの一連の処理に置き換えます。Kafka、RabbitMQ、または Axon Framework を使用した、コレオグラフィ（イベント駆動型）とオーケストレーション（中央集権型コーディネーター）のアプローチをサポートします。

使用する場面

• 複数のマイクロサービスにまたがる分散トランザクションを構築する場合
• 2 フェーズコミット（2PC）を、よりスケーラブルなソリューションに置き換える場合
• サービス障害時のトランザクションロールバックに対応する場合
• マイクロサービスアーキテクチャで結果整合性を確保する場合
• 失敗した操作に対する補償トランザクションを実装する場合
• 複数のサービスにまたがる複雑なビジネスプロセスを調整する場合

トリガーフレーズ：分散トランザクション、Saga パターン、補償トランザクション、マイクロサービスのトランザクション、結果整合性、サービス間のロールバック、オーケストレーションパターン、コレオグラフィパターン

手順

1. トランザクションフローを設計する

操作の順序と、それぞれに対応する補償トランザクションを整理します。

Order → Payment → Inventory → Shipment
  ↓        ↓        ↓          ↓
Cancel  Refund   Release    Cancel

検証：すべての前進ステップに、対応する補償処理があることを確認します。

2. 実装アプローチを選択する

アプローチ	ユースケース	スタック
コレオグラフィ	新規開発、参加者が少ない場合	Spring Cloud Stream + Kafka/RabbitMQ
オーケストレーション	複雑なワークフロー、既存システムの改修	Axon Framework、Eventuate Tram、Camunda

検証：選択する前に、チームの専門知識とシステムの複雑さを検討します。

3. ローカルトランザクションを使用してサービスを実装する

各サービスは、ローカルの ACID トランザクションをアトミックに完了させます。

@Service
@RequiredArgsConstructor
public class OrderService {
    private final OrderRepository orderRepository;
    private final KafkaTemplate<String, Object> kafka;

    @Transactional
    public Order createOrder(CreateOrderCommand cmd) {
        Order order = orderRepository.save(new Order(cmd.orderId(), cmd.items()));
        kafka.send("order.created", new OrderCreatedEvent(order.getId(), order.getItems()));
        return order;
    }
}

検証：イベントが公開される前にローカルトランザクションがコミットされることをテストします。

4. 補償トランザクションを実装する

すべての前進処理には、冪等性のある補償処理が必要です。

@Service
@RequiredArgsConstructor
public class PaymentService {
    private final PaymentRepository paymentRepository;
    private final KafkaTemplate<String, Object> kafka;

    public void processPayment(PaymentRequest request) {
        Payment payment = paymentRepository.save(new Payment(request.orderId(), request.amount()));
        kafka.send("payment.processed", new PaymentProcessedEvent(payment.getId(), request.orderId()));
    }

    @Transactional
    public void refundPayment(String paymentId) {
        paymentRepository.findById(paymentId)
            .ifPresent(p -> {
                p.setStatus(REFUNDED);
                paymentRepository.save(p);
                kafka.send("payment.refunded", new PaymentRefundedEvent(paymentId));
            });
    }
}

検証：補償処理を複数回安全に実行できること（冪等性）を確認します。

5. メッセージブローカーをセットアップする

冪等なコンシューマーを使用するように Kafka を設定します。

@Configuration
@EnableKafka
public class KafkaConfig {
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
            ConsumerFactory<String, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setCommonErrorHandler(new DefaultErrorHandler());
        return factory;
    }
}

検証：トランザクション ID を有効にし、exactly-once セマンティクスを検証します。

6. Saga オーケストレーターを実装する（オーケストレーションのみ）

@Service
@RequiredArgsConstructor
public class OrderSagaOrchestrator {
    private final KafkaTemplate<String, Object> kafka;
    private final SagaStateRepository sagaStateRepo;

    public void startSaga(OrderRequest request) {
        String sagaId = UUID.randomUUID().toString();
        sagaStateRepo.save(new SagaState(sagaId, STARTED, LocalDateTime.now()));
        kafka.send("saga.order.start", new StartOrderSagaCommand(sagaId, request));
    }

    @KafkaListener(topics = "payment.failed")
    public void handlePaymentFailed(PaymentFailedEvent event) {
        kafka.send("order.compensate", new CompensateOrderCommand(event.getSagaId()));
        kafka.send("inventory.compensate", new ReleaseInventoryCommand(event.getSagaId()));
        sagaStateRepo.updateStatus(event.getSagaId(), FAILED);
    }
}

検証：コマンドの送信前に Saga の状態が永続化されることを確認します。各障害経路で補償処理が開始されることを確認します。

7. イベントハンドラーを実装する（コレオグラフィのみ）

@Service
public class OrderEventHandler {
    private final OrderService orderService;
    private final KafkaTemplate<String, Object> kafka;

    @KafkaListener(topics = "payment.processed", groupId = "order-service")
    public void onPaymentProcessed(PaymentProcessedEvent event) {
        try {
            InventoryReservedEvent result = orderService.reserveInventory(event.toInventoryRequest());
            kafka.send("inventory.reserved", result);
        } catch (InsufficientInventoryException e) {
            kafka.send("inventory.insufficient", new InsufficientInventoryEvent(event.getOrderId(), event.getPaymentId()));
        }
    }
}

検証：各イベントハンドラーが次のステップまたは補償処理を正しく開始することをテストします。

8. モニタリングと可観測性を追加する

@Configuration
public class SagaMetricsConfig {
    @Bean
    public MeterRegistry meterRegistry() {
        return new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
    }
}

追跡項目：Saga の実行時間、補償処理の回数、失敗率、停滞した Saga。

検証：想定時間を超過した Saga に対するアラートを設定します。

ベストプラクティス

設計：

• データベース制約または重複排除テーブルを使用して、補償トランザクションを冪等にします
• 偶発的な変更を防ぐため、不変イベント（Java レコード）を使用します
• 復旧に備えて、Saga の状態を永続ストレージに保存します

エラー処理：

• サービス間呼び出しにサーキットブレーカーを実装します
• 再試行上限を超えたメッセージにはデッドレターキューを使用します
• Saga の各ステップに適切なタイムアウトを設定します（デフォルト 30s、設定可能）

モニタリング：

• Saga の状態を追跡します：PENDING、COMPLETED、COMPENSATING、FAILED
• 補償処理の実行時間をモニタリングします
• Saga が SLA の所要時間を超過した場合にアラートを出します

制約と警告

• すべての前進トランザクションには、対応する補償トランザクションが必須です
• 再試行に対応するため、補償トランザクションは冪等でなければなりません
• 障害と復旧に対応するため、Saga の状態は永続化しなければなりません
• Saga の参加者間で同期通信を使用しないでください
• Saga が提供するのは結果整合性であり、強整合性ではありません
• 部分的な障害を含め、すべての障害シナリオをテストします
• 複雑なオーケストレーションには、Axon Framework または Eventuate の使用を検討します
• メッセージブローカーの高可用性を確保します

例

コレオグラフィベースの Saga

// Application.java
@SpringBootApplication
@EnableKafka
@EnableKafkaListeners
public class OrderApplication {
    public static void main(String[] args) {
        SpringApplication.run(OrderApplication.class, args);
    }
}

// Event Classes (immutable)
public record OrderCreatedEvent(String orderId, List<OrderItem> items) {}
public record PaymentProcessedEvent(String paymentId, String orderId) {}
public record InventoryReservedEvent(String reservationId, String orderId) {}
public record PaymentFailedEvent(String orderId, String reason) {}
public record InsufficientInventoryEvent(String orderId, String paymentId) {}

// OrderService with compensation
@Service
@RequiredArgsConstructor
public class OrderService {
    private final OrderRepository orderRepository;
    private final KafkaTemplate<String, Object> kafka;

    @KafkaListener(topics = "payment.failed", groupId = "order-service")
    public void handleCompensation(PaymentFailedEvent event) {
        orderRepository.findByOrderId(event.orderId())
            .ifPresent(order -> {
                order.setStatus(CANCELLED);
                orderRepository.save(order);
            });
    }
}

Axon Framework を使用したオーケストレーションベースの Saga

// Command
@Aggregate
public class OrderAggregate {
    @AggregateIdentifier
    private String orderId;

    @CommandHandler
    public OrderAggregate(CreateOrderCommand cmd) {
        apply(new OrderCreatedEvent(cmd.orderId(), cmd.items()));
    }

    @EventSourcingHandler
    public void on(OrderCreatedEvent event) {
        this.orderId = event.orderId();
    }

    @CommandHandler
    public void handle(CancelOrderCommand cmd) {
        apply(new OrderCancelledEvent(cmd.orderId(), cmd.reason()));
    }
}

参考資料

• Saga パターンの定義
• コレオグラフィーの実装
• オーケストレーションの実装
• 補償トランザクション
• 状態管理
• エラー処理と再試行
• テスト戦略
• 落とし穴と解決策
• 例