跳至主要內容

SAGA 模式：长事务的优雅解决方案

老马啸西风2025/9/1大约 9 分钟DisTxdis-tx

SAGA 模式：长事务的优雅解决方案

在分布式系统中，处理长时间运行的业务流程是一个常见且具有挑战性的问题。传统的ACID事务在处理这类场景时往往表现不佳，要么因为锁持有时间过长影响系统性能，要么因为超时导致事务失败。SAGA模式作为一种长事务的解决方案，为这类问题提供了优雅的解决思路。

SAGA模式概述

SAGA的核心概念

SAGA模式最初由Hector Garcia-Molina和Kenneth Salem在1987年提出，它将一个长事务分解为多个短事务，每个短事务都对应一个补偿事务。当所有短事务都成功执行时，整个SAGA事务成功；如果任何一个短事务失败，则通过执行相应的补偿事务来回滚已执行的操作。

SAGA的基本特征

  1. 长事务分解：将一个长事务分解为多个短事务
  2. 补偿机制：每个短事务都有对应的补偿事务
  3. 最终一致性：通过补偿机制保证最终一致性
  4. 异步执行：短事务可以异步执行，提高系统性能

顺序执行与补偿事务

顺序执行机制

在SAGA模式中，事务按照预定义的顺序依次执行，每个步骤都是一个独立的本地事务。

执行流程

  1. 事务分解：将长事务分解为多个短事务T1, T2, ..., Tn
  2. 顺序执行：按照T1 → T2 → ... → Tn的顺序执行
  3. 状态记录：记录每个事务的执行状态
  4. 成功完成：所有事务执行成功，SAGA事务完成

顺序执行示例

以电商订单处理为例，一个完整的订单流程可能包括以下步骤：

public class OrderSaga {
    
    private List<SagaStep> steps = Arrays.asList(
        new ReserveInventoryStep(),
        new CreateOrderStep(),
        new ProcessPaymentStep(),
        new UpdateInventoryStep(),
        new NotifyUserStep()
    );
    
    public void execute(OrderRequest request) {
        List<ExecutedStep> executedSteps = new ArrayList<>();
        
        try {
            // 顺序执行每个步骤
            for (SagaStep step : steps) {
                ExecutedStep executedStep = step.execute(request);
                executedSteps.add(executedStep);
            }
            
            // 所有步骤执行成功
            markSagaAsCompleted();
        } catch (Exception e) {
            // 执行补偿操作
            compensate(executedSteps);
            throw new SagaExecutionException("Saga execution failed", e);
        }
    }
    
    private void compensate(List<ExecutedStep> executedSteps) {
        // 逆序执行补偿操作
        for (int i = executedSteps.size() - 1; i >= 0; i--) {
            try {
                executedSteps.get(i).compensate();
            } catch (Exception e) {
                // 记录补偿失败，但继续执行其他补偿操作
                log.error("Compensation failed for step: " + executedSteps.get(i).getStepName(), e);
            }
        }
    }
}

补偿事务设计

补偿事务是SAGA模式的核心，它负责撤销已执行的业务操作。

补偿事务的特点

  1. 可逆性：必须能够完全撤销对应的业务操作
  2. 幂等性：支持重复执行，不会产生副作用
  3. 独立性：每个补偿事务独立执行，不依赖其他补偿事务

补偿事务实现示例

public class ReserveInventoryStep implements SagaStep {
    
    @Override
    public ExecutedStep execute(OrderRequest request) {
        // 执行库存预留
        String reservationId = inventoryService.reserve(request.getProductId(), request.getQuantity());
        
        // 返回已执行步骤信息，包含补偿操作
        return new ExecutedStep() {
            @Override
            public String getStepName() {
                return "ReserveInventory";
            }
            
            @Override
            public void compensate() {
                // 执行补偿：释放预留的库存
                inventoryService.release(reservationId);
            }
        };
    }
}

public class ProcessPaymentStep implements SagaStep {
    
    @Override
    public ExecutedStep execute(OrderRequest request) {
        // 执行支付处理
        String paymentId = paymentService.process(request.getPaymentInfo());
        
        // 返回已执行步骤信息，包含补偿操作
        return new ExecutedStep() {
            @Override
            public String getStepName() {
                return "ProcessPayment";
            }
            
            @Override
            public void compensate() {
                // 执行补偿：退款
                paymentService.refund(paymentId);
            }
        };
    }
}

编排式 vs 协作式 SAGA

编排式SAGA（Orchestration）

在编排式SAGA中，存在一个中央协调器（Orchestrator）来控制整个事务的执行流程。

架构特点

  1. 中央控制：由协调器控制事务的执行顺序
  2. 状态管理：协调器维护事务的执行状态
  3. 决策逻辑：协调器决定何时执行下一步或触发补偿

编排式SAGA实现

@Component
public class OrderSagaOrchestrator {
    
    @Autowired
    private SagaStateRepository stateRepository;
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private PaymentService paymentService;
    
    public void startSaga(OrderRequest request) {
        // 初始化SAGA状态
        SagaState state = new SagaState();
        state.setSagaId(UUID.randomUUID().toString());
        state.setStatus(SagaStatus.RUNNING);
        state.setRequest(request);
        stateRepository.save(state);
        
        try {
            // 步骤1：预留库存
            String reservationId = inventoryService.reserve(request.getProductId(), request.getQuantity());
            state.setReservationId(reservationId);
            state.setCurrentStep(1);
            stateRepository.update(state);
            
            // 步骤2：创建订单
            String orderId = orderService.createOrder(request);
            state.setOrderId(orderId);
            state.setCurrentStep(2);
            stateRepository.update(state);
            
            // 步骤3：处理支付
            String paymentId = paymentService.process(request.getPaymentInfo());
            state.setPaymentId(paymentId);
            state.setCurrentStep(3);
            stateRepository.update(state);
            
            // SAGA执行成功
            state.setStatus(SagaStatus.COMPLETED);
            stateRepository.update(state);
        } catch (Exception e) {
            // 执行补偿
            compensate(state);
            throw new SagaExecutionException("Saga execution failed", e);
        }
    }
    
    private void compensate(SagaState state) {
        state.setStatus(SagaStatus.COMPENSATING);
        stateRepository.update(state);
        
        try {
            // 根据当前步骤执行相应的补偿操作
            switch (state.getCurrentStep()) {
                case 3:
                    // 补偿支付
                    paymentService.refund(state.getPaymentId());
                case 2:
                    // 补偿订单创建
                    orderService.cancelOrder(state.getOrderId());
                case 1:
                    // 补偿库存预留
                    inventoryService.release(state.getReservationId());
            }
            
            state.setStatus(SagaStatus.COMPENSATED);
            stateRepository.update(state);
        } catch (Exception e) {
            state.setStatus(SagaStatus.COMPENSATION_FAILED);
            stateRepository.update(state);
            throw new SagaCompensationException("Saga compensation failed", e);
        }
    }
}

协作式SAGA（Choreography）

在协作式SAGA中，没有中央协调器，各个服务通过事件驱动的方式协作完成事务。

架构特点

  1. 去中心化：没有中央协调器，各服务自主决策
  2. 事件驱动：通过事件来触发下一步操作
  3. 松耦合：服务之间通过事件进行通信，耦合度较低

协作式SAGA实现

@Service
public class OrderService {
    
    @Autowired
    private EventBus eventBus;
    
    public void handleOrderCreationRequested(OrderCreationRequestedEvent event) {
        try {
            // 创建订单
            Order order = createOrder(event.getRequest());
            
            // 发布订单创建成功事件
            OrderCreatedEvent createdEvent = new OrderCreatedEvent();
            createdEvent.setOrderId(order.getId());
            createdEvent.setRequest(event.getRequest());
            eventBus.publish(createdEvent);
        } catch (Exception e) {
            // 发布订单创建失败事件
            OrderCreationFailedEvent failedEvent = new OrderCreationFailedEvent();
            failedEvent.setRequest(event.getRequest());
            failedEvent.setError(e.getMessage());
            eventBus.publish(failedEvent);
        }
    }
    
    public void handleOrderCreationFailed(OrderCreationFailedEvent event) {
        // 处理订单创建失败，可能需要触发其他补偿操作
        log.info("Order creation failed, triggering compensation");
    }
}

@Service
public class PaymentService {
    
    @Autowired
    private EventBus eventBus;
    
    public void handleOrderCreated(OrderCreatedEvent event) {
        try {
            // 处理支付
            String paymentId = processPayment(event.getRequest().getPaymentInfo());
            
            // 发布支付成功事件
            PaymentProcessedEvent processedEvent = new PaymentProcessedEvent();
            processedEvent.setPaymentId(paymentId);
            processedEvent.setOrderId(event.getOrderId());
            eventBus.publish(processedEvent);
        } catch (Exception e) {
            // 发布支付失败事件
            PaymentFailedEvent failedEvent = new PaymentFailedEvent();
            failedEvent.setOrderId(event.getOrderId());
            failedEvent.setError(e.getMessage());
            eventBus.publish(failedEvent);
        }
    }
    
    public void handlePaymentFailed(PaymentFailedEvent event) {
        // 处理支付失败，触发补偿操作
        log.info("Payment failed, triggering compensation for order: " + event.getOrderId());
        // 发布补偿事件
        PaymentCompensationEvent compensationEvent = new PaymentCompensationEvent();
        compensationEvent.setOrderId(event.getOrderId());
        eventBus.publish(compensationEvent);
    }
}

与微服务调用链结合

SAGA在微服务架构中的应用

在微服务架构中，SAGA模式能够很好地解决跨服务事务的问题。每个微服务负责自己的业务逻辑和补偿逻辑，通过事件驱动的方式协作完成复杂的业务流程。

微服务SAGA架构

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Inventory     │    │     Order       │    │    Payment      │
│    Service      │    │    Service      │    │    Service      │
│                 │    │                 │    │                 │
│ reserve()       │    │ createOrder()   │    │ process()       │
│ release()       │    │ cancelOrder()   │    │ refund()        │
└─────────────────┘    └─────────────────┘    └─────────────────┘
         │                       │                       │
         ▼                       ▼                       ▼
┌─────────────────────────────────────────────────────────────────┐
│                        Event Bus                                │
└─────────────────────────────────────────────────────────────────┘

分布式追踪集成

在SAGA模式中，分布式追踪对于监控和调试非常重要。

@Component
public class TracedSagaOrchestrator {
    
    @Autowired
    private Tracer tracer;
    
    public void executeSaga(OrderRequest request) {
        // 创建SAGA追踪跨度
        Span sagaSpan = tracer.buildSpan("saga-execution")
            .withTag("saga-type", "order-processing")
            .withTag("order-id", request.getOrderId())
            .start();
            
        try (Scope scope = tracer.activateSpan(sagaSpan)) {
            // 执行各个步骤并创建子跨度
            executeInventoryStep(request, sagaSpan);
            executeOrderStep(request, sagaSpan);
            executePaymentStep(request, sagaSpan);
            
            sagaSpan.finish();
        } catch (Exception e) {
            sagaSpan.setTag("error", true);
            sagaSpan.log(ImmutableMap.of("event", "error", "message", e.getMessage()));
            sagaSpan.finish();
            throw e;
        }
    }
    
    private void executeInventoryStep(OrderRequest request, Span parentSpan) {
        Span stepSpan = tracer.buildSpan("reserve-inventory")
            .asChildOf(parentSpan)
            .withTag("product-id", request.getProductId())
            .withTag("quantity", request.getQuantity())
            .start();
            
        try (Scope scope = tracer.activateSpan(stepSpan)) {
            inventoryService.reserve(request.getProductId(), request.getQuantity());
            stepSpan.finish();
        } catch (Exception e) {
            stepSpan.setTag("error", true);
            stepSpan.log(ImmutableMap.of("event", "error", "message", e.getMessage()));
            stepSpan.finish();
            throw e;
        }
    }
}

SAGA模式的优缺点

优点

  1. 支持长事务：能够处理长时间运行的业务流程
  2. 高可用性：不需要长时间持有锁，系统可用性高
  3. 灵活性：可以灵活地添加或修改业务步骤
  4. 可扩展性：易于水平扩展，适合微服务架构

缺点

  1. 复杂性：需要为每个业务操作设计补偿逻辑
  2. 最终一致性：只能保证最终一致性，不适合强一致性要求的场景
  3. 调试困难：分布式环境下调试和排查问题较为困难
  4. 幂等性要求：所有操作都需要实现幂等性

实际应用案例

电商订单处理

在电商系统中，一个完整的订单处理流程可能涉及多个服务：

public class ECommerceOrderSaga {
    
    private static final List<SagaStep> ORDER_STEPS = Arrays.asList(
        new ValidateOrderStep(),
        new ReserveInventoryStep(),
        new CreateOrderStep(),
        new ProcessPaymentStep(),
        new UpdateInventoryStep(),
        new ShipOrderStep(),
        new NotifyCustomerStep()
    );
    
    public void processOrder(OrderRequest request) {
        SagaContext context = new SagaContext();
        context.setRequest(request);
        context.setSagaId(UUID.randomUUID().toString());
        
        List<ExecutedStep> executedSteps = new ArrayList<>();
        
        try {
            for (SagaStep step : ORDER_STEPS) {
                ExecutedStep executedStep = step.execute(context);
                executedSteps.add(executedStep);
                context.addExecutedStep(executedStep);
            }
            
            // 订单处理成功
            context.setStatus(SagaStatus.COMPLETED);
            sagaRepository.update(context);
        } catch (Exception e) {
            // 执行补偿
            compensate(executedSteps, context);
            throw new OrderProcessingException("Order processing failed", e);
        }
    }
    
    private void compensate(List<ExecutedStep> executedSteps, SagaContext context) {
        context.setStatus(SagaStatus.COMPENSATING);
        sagaRepository.update(context);
        
        // 逆序执行补偿操作
        for (int i = executedSteps.size() - 1; i >= 0; i--) {
            try {
                executedSteps.get(i).compensate();
            } catch (Exception e) {
                log.error("Compensation failed", e);
                // 记录补偿失败，但继续执行其他补偿操作
            }
        }
        
        context.setStatus(SagaStatus.COMPENSATED);
        sagaRepository.update(context);
    }
}

金融转账场景

在金融系统中，跨行转账可能涉及多个银行和清算系统：

public class CrossBankTransferSaga {
    
    public void executeTransfer(TransferRequest request) {
        TransferContext context = new TransferContext();
        context.setRequest(request);
        context.setTransferId(UUID.randomUUID().toString());
        
        List<TransferStep> executedSteps = new ArrayList<>();
        
        try {
            // 步骤1：验证转账请求
            validateTransfer(request);
            executedSteps.add(new ValidationStep());
            
            // 步骤2：冻结源账户资金
            freezeSourceAccount(request);
            executedSteps.add(new FreezeAccountStep());
            
            // 步骤3：通知目标银行
            notifyTargetBank(request);
            executedSteps.add(new NotifyBankStep());
            
            // 步骤4：执行转账
            executeTransfer(request);
            executedSteps.add(new ExecuteTransferStep());
            
            // 步骤5：解冻源账户资金
            unfreezeSourceAccount(request);
            executedSteps.add(new UnfreezeAccountStep());
            
            // 转账成功
            context.setStatus(TransferStatus.COMPLETED);
        } catch (Exception e) {
            // 执行补偿
            compensate(executedSteps, context);
            throw new TransferException("Transfer failed", e);
        }
    }
}

SAGA模式的最佳实践

1. 合理设计补偿逻辑

补偿逻辑是SAGA模式成功的关键，需要仔细设计：

public class PaymentService {
    
    public void processPayment(PaymentRequest request) {
        // 执行支付处理
        String paymentId = paymentGateway.process(request);
        
        // 记录支付记录，用于补偿
        PaymentRecord record = new PaymentRecord();
        record.setPaymentId(paymentId);
        record.setAmount(request.getAmount());
        record.setStatus(PaymentStatus.PROCESSED);
        paymentRepository.save(record);
    }
    
    public void refund(String paymentId) {
        // 幂等性检查
        PaymentRecord record = paymentRepository.findByPaymentId(paymentId);
        if (record == null || record.getStatus() == PaymentStatus.REFUNDED) {
            return;
        }
        
        // 执行退款
        paymentGateway.refund(paymentId);
        
        // 更新状态
        record.setStatus(PaymentStatus.REFUNDED);
        paymentRepository.update(record);
    }
}

2. 实现幂等性

所有SAGA操作都需要实现幂等性：

public class OrderService {
    
    public Order createOrder(String requestId, OrderRequest request) {
        // 幂等性检查
        Order existingOrder = orderRepository.findByRequestId(requestId);
        if (existingOrder != null) {
            return existingOrder;
        }
        
        // 创建订单
        Order order = new Order();
        order.setRequestId(requestId);
        order.setDetails(request);
        order.setStatus(OrderStatus.CREATED);
        
        return orderRepository.save(order);
    }
}

3. 状态持久化

SAGA状态需要持久化，以便在系统故障后能够恢复：

@Entity
@Table(name = "saga_state")
public class SagaState {
    
    @Id
    private String sagaId;
    
    private String sagaType;
    
    @Enumerated(EnumType.STRING)
    private SagaStatus status;
    
    @Lob
    private String requestData;
    
    @Lob
    private String responseData;
    
    private int currentStep;
    
    private Date createTime;
    
    private Date updateTime;
    
    // getters and setters
}

4. 监控和告警

完善的监控和告警机制对于SAGA模式的运维至关重要：

@Component
public class SagaMonitor {
    
    private final MeterRegistry meterRegistry;
    
    public void recordSagaExecution(String sagaType, boolean success) {
        Counter.builder("saga.execution")
            .tag("saga_type", sagaType)
            .tag("success", String.valueOf(success))
            .register(meterRegistry)
            .increment();
    }
    
    public void recordCompensation(String sagaType) {
        Counter.builder("saga.compensation")
            .tag("saga_type", sagaType)
            .register(meterRegistry)
            .increment();
    }
    
    @EventListener
    public void handleSagaFailed(SagaFailedEvent event) {
        // 发送告警
        alertService.sendAlert("SAGA execution failed: " + event.getSagaType());
    }
}

总结

SAGA模式是一种处理长事务的有效解决方案，特别适用于微服务架构中的复杂业务流程。通过将长事务分解为多个短事务，并为每个短事务设计补偿逻辑，SAGA模式能够在保证最终一致性的同时，提供较高的系统可用性和性能。

然而，SAGA模式也有其局限性，特别是在补偿逻辑设计和幂等性实现方面需要特别注意。在实际应用中，需要根据具体的业务场景来决定是否采用SAGA模式，并遵循最佳实践来确保系统的稳定性和可靠性。

在后续章节中，我们将继续探讨其他分布式事务模式和框架，帮助你在实际项目中正确选择和应用这些技术。

我是老马，期待与你的下次重逢
Copyright © 2025 老马啸西风