spring-boot-saga-pattern

Spring Boot Saga 模式

概述

使用 Saga 模式在微服务之间实现分布式事务。用一系列本地事务和补偿操作取代两阶段提交。支持基于 Kafka、RabbitMQ 或 Axon Framework 的编舞（事件驱动）和编排（集中式协调器）方式。

适用场景

• 构建跨多个微服务的分布式事务
• 用更具扩展性的解决方案取代两阶段提交（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，并验证恰好一次语义。

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 record），防止意外修改
• 将 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 模式定义
• 编舞式实现
• 编排式实现
• 补偿事务
• 状态管理
• 错误处理与重试
• 测试策略
• 陷阱与解决方案
• 示例