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を有効にし、正確に1回のセマンティクスを確認します。

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ステップごとに適切なタイムアウトを設定する（デフォルト30秒、構成可能）

モニタリング:

• saga状態を追跡：PENDING、COMPLETED、COMPENSATING、FAILED
• 補償実行時間を監視する
• sagaがSLA期間を超えた場合はアラートを出す

制約と警告

• すべての前進トランザクションに対応する補償トランザクションが必須
• 補償トランザクションはリトライシナリオに対応するためべき等である必須
• saga状態は障害とリカバリに対応するため永続化が必須
• sagaの参加者間での同期通信は決して使用しない
• sagaは結果整合性を提供し、強い整合性は提供しない
• すべての障害シナリオ（部分障害を含む）をテストする
• 複雑なオーケストレーションにはAxon FrameworkまたはEventicateを検討する
• メッセージブローカーの高可用性を確保する

例

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

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

// イベントクラス（イミュータブル）
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
@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 Pattern Definition
• Choreography Implementation
• Orchestration Implementation
• Compensating Transactions
• State Management
• Error Handling and Retry
• Testing Strategies
• Pitfalls and Solutions
• Examples