从崩溃到重生：微服务架构下Saga分布式事务的终极实战指南

当单体应用拆分为数十个微服务，你突然发现自己陷入了一个分布式事务的噩梦——没有全局锁、没有共享事务、网络随时可能断开。Saga模式，正是斩断这团乱麻的利刃。本文将带你从零构建一个生产级Saga编排引擎，涵盖补偿事务设计、幂等性保障、悬挂与孤儿事务的防御策略。

一、为什么传统事务在微服务中彻底失效

在单体架构中，我们习惯于依赖数据库的ACID事务：扣库存、创建订单、扣款，三步操作包裹在同一个事务中，要么全部成功，要么全部回滚。但当这些操作分散在三个独立服务、三个独立数据库中时，两阶段提交（2PC）就变成了性能杀手。

2PC的致命缺陷：

• 同步阻塞：所有参与者锁定资源等待协调者决策，在高并发下导致连接池耗尽
• 单点故障：协调者宕机导致所有参与者无限期等待
• 脑裂风险：网络分区时部分参与者提交、部分回滚
• 性能灾难：两阶段RTT + 锁定时间，TPS通常衰减60%以上

Saga模式的核心思想是：长事务拆分为一系列本地短事务，每个本地事务成功后发布事件，后续事务失败时触发之前的补偿操作。这不是回滚，而是向前补偿（compensating forward）。

二、Saga的两种实现模式：编排（Choreography）vs 指挥（Orchestration）

2.1 编排模式（Choreography）

各服务自治，通过事件总线传递信号。订单服务发出OrderCreated事件，库存服务监听并扣减，成功后发出StockReserved事件，支付服务监听并扣款。

@EventListener
public void onOrderCreated(OrderCreatedEvent event) {
    try {
        stockRepository.decrease(event.getProductId(), event.getQuantity());
        eventPublisher.publish(new StockReservedEvent(event.getOrderId()));
    } catch (InsufficientStockException e) {
        eventPublisher.publish(new StockReservationFailedEvent(event.getOrderId()));
    }
}

编排的优势：去中心化、松耦合，适合简单流程（2-3个服务）。

编排的噩梦：当流程涉及5+服务时，事件流向变成"回调地狱"，无人能说清全局流程；服务间形成隐性依赖；测试时需要同时启动整个事件链。

2.2 指挥模式（Orchestration）

引入一个Saga协调者作为大脑，它告诉每个服务该做什么，然后根据返回结果决定下一步。

public class CreateOrderSaga {
    
    private final StateMachine<SagaState, SagaEvent> stateMachine;
    
    public CreateOrderSaga() {
        this.stateMachine = StateMachineBuilder
            .newBuilder()
            .initialState(SagaState.CREATED)
            .transition(SagaState.CREATED, SagaState.ORDER_CREATED)
            .onEvent(SagaEvent.CREATE_ORDER)
            .execute(ctx -> orderService.create(ctx))
            .transition(SagaState.ORDER_CREATED, SagaState.STOCK_RESERVED)
            .onEvent(SagaEvent.RESERVE_STOCK)
            .execute(ctx -> stockService.reserve(ctx))
            .onFailure(SagaCompensation.RELEASE_STOCK)
            .transition(SagaState.STOCK_RESERVED, SagaState.PAYMENT_COMPLETED)
            .onEvent(SagaEvent.PROCESS_PAYMENT)
            .execute(ctx -> paymentService.charge(ctx))
            .onFailure(SagaCompensation.REFUND_PAYMENT)
            .transition(SagaState.PAYMENT_COMPLETED, SagaState.COMPLETED)
            .onEvent(SagaEvent.CONFIRM)
            .build();
    }
}

指挥的优势：集中管控、流程可视化、易调试，适合复杂业务流。

指挥的代价：协调者本身成为单点，需要高可用设计。

三、生产级Saga引擎的核心设计要素

3.1 幂等性——补偿安全的基石

补偿操作必须保证幂等，因为网络超时可能导致补偿被重复触发。

BEGIN;
  UPDATE orders 
  SET status = 'CANCELLED', cancelled_at = NOW()
  WHERE order_id = ? AND status != 'CANCELLED';
  SELECT ROW_COUNT() AS affected_rows;
COMMIT;

-- affected_rows=0 表示已经补偿过，无需再执行

3.2 事务日志与持久化

Saga引擎必须将每一步操作及其状态持久化到可靠存储，确保即使进程崩溃，恢复后仍能继续推进或补偿。

CREATE TABLE saga_instance (
    saga_id         VARCHAR(64) PRIMARY KEY,
    saga_type       VARCHAR(128) NOT NULL,
    current_state   VARCHAR(64) NOT NULL,
    payload         JSON,
    created_at      TIMESTAMP DEFAULT NOW(),
    updated_at      TIMESTAMP,
    completed_at    TIMESTAMP,
    version         INT DEFAULT 0,
    INDEX idx_state_time (current_state, updated_at)
);

CREATE TABLE saga_event_log (
    log_id          BIGINT AUTO_INCREMENT PRIMARY KEY,
    saga_id         VARCHAR(64) NOT NULL,
    step_name       VARCHAR(128),
    action          ENUM('EXECUTE', 'COMPENSATE', 'SKIP'),
    status          ENUM('SUCCESS', 'FAILED', 'TIMEOUT', 'SKIPPED'),
    request_payload JSON,
    response_payload JSON,
    error_message   TEXT,
    executed_at     TIMESTAMP,
    INDEX idx_saga (saga_id, log_id)
);

3.3 超时与悬挂事务防御

悬挂事务：某个本地事务因网络超时失败，但实际上已成功提交，后续补偿操作不应执行。

孤儿事务：Saga协调者崩溃后恢复，某些本地事务已完成但未被协调者记录。

防御策略：

• 每个本地事务写入txn_marker标记，补偿时先读标记确认状态
• 引入Saga超时定时器，对长时间无响应的Saga实例进行全局补偿
• 实现空补偿防护：补偿时发现正向操作未执行，直接跳过

四、实战：手写一个Saga编排引擎（基于Spring Boot + StateMachine）

4.1 核心抽象层

public interface SagaStep {
    String getStepName();
    SagaResult execute(SagaContext context);
    SagaResult compensate(SagaContext context);
    default boolean isSkippable() { return false; }
}

public class SagaContext {
    private final String sagaId;
    private final Map<String, Object> parameters = new ConcurrentHashMap<>();
    private final Map<String, SagaResult> stepResults = new LinkedHashMap<>();
    
    public void putResult(String stepName, SagaResult result) {
        stepResults.put(stepName, result);
    }
}

4.2 协调者核心逻辑

@Component
public class sagaOrchestrator {
    
    private final SagaLogRepository logRepository;
    private final ThreadPoolExecutor executor = 
        new ThreadPoolExecutor(8, 32, 60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000));
    
    public SagaResult execute(String sagaType, SagaContext context) {
        List<SagaStep> steps = stepRegistry.getSteps(sagaType);
        int stepIndex = 0;
        SagaResult lastResult = SagaResult.success();
        Set<String> executedSteps = logRepository.findExecutedSteps(context.getSagaId());
        
        try {
            while (stepIndex < steps.size()) {
                SagaStep step = steps.get(stepIndex);
                if (executedSteps.contains(step.getStepName())) {
                    stepIndex++;
                    continue;
                }
                logRepository.logStepStart(context.getSagaId(), step.getStepName());
                lastResult = step.execute(context);
                context.putResult(step.getStepName(), lastResult);
                logRepository.logStepSuccess(context.getSagaId(), step.getStepName(), lastResult);
                if (!lastResult.isSuccess()) {
                    compensate(steps, stepIndex, context);
                    return SagaResult.failed(lastResult.getError());
                }
                stepIndex++;
            }
            logRepository.markSagaCompleted(context.getSagaId());
            return SagaResult.success();
        } catch (Exception e) {
            compensate(steps, stepIndex, context);
            throw new SagaException("Saga execution failed", e);
        }
    }
    
    private void compensate(List<SagaStep> steps, int failedIndex, SagaContext context) {
        for (int i = failedIndex - 1; i >= 0; i--) {
            SagaStep step = steps.get(i);
            try {
                SagaResult result = step.compensate(context);
                logRepository.logCompensation(context.getSagaId(), 
                    step.getStepName(), result);
            } catch (Exception e) {
                alertService.sendCompensationFailureAlert(context.getSagaId(), 
                    step.getStepName(), e);
            }
        }
    }
}

4.3 三步构建订单Saga

@Service
public class OrderSagaStepConfig {
    
    @Bean
    public SagaStep<SagaContext> createOrderStep(OrderService orderService) {
        return SagaStep.<SagaContext>builder()
            .stepName('createOrder')
            .execute(ctx -> {
                CreateOrderRequest req = ctx.get('orderRequest');
                Order order = orderService.createOrder(req);
                ctx.put('orderId', order.getId());
                return SagaResult.success(order);
            })
            .compensate(ctx -> {
                Long orderId = ctx.get('orderId');
                orderService.cancelOrder(orderId);
                return SagaResult.success();
            })
            .build();
    }
    
    @Bean
    public SagaStep<SagaContext> reserveStockStep(StockFeignClient stockClient) {
        return SagaStep.<SagaContext>builder()
            .stepName('reserveStock')
            .execute(ctx -> {
                Long orderId = ctx.get('orderId');
                ReserveStockRequest req = ctx.get('stockRequest');
                return stockClient.reserve(orderId, req);
            })
            .compensate(ctx -> {
                Long orderId = ctx.get('orderId');
                return stockClient.release(orderId);
            })
            .build();
    }
    
    @Bean
    public SagaStep<SagaContext> processPaymentStep(PaymentFeignClient paymentClient) {
        return SagaStep.<SagaContext>builder()
            .stepName('processPayment')
            .execute(ctx -> {
                Long orderId = ctx.get('orderId');
                PaymentRequest req = ctx.get('paymentRequest');
                return paymentClient.charge(orderId, req);
            })
            .compensate(ctx -> {
                Long orderId = ctx.get('orderId');
                return paymentClient.refund(orderId);
            })
            .build();
    }
}

五、高级话题：Saga的性能优化与避坑指南

5.1 并行执行加速

当多个步骤互不依赖时，通过并行执行大幅降低整体延迟。

sagaBuilder
    .parallel(
        branch -> branch.add(stepA).add(stepB),
        branch -> branch.add(stepC)
    )
    .than()
    .add(stepD)
    .add(stepE);

5.2 可观测性——Saga的X光机

saga:
  metrics:
    enabled: true
    prefix: 'saga.orchestrator'
    counters:
      - saga.completed
      - saga.compensated
      - saga.timed_out
    histograms:
      - saga.step.duration
      - saga.total.duration
    alerts:
      - condition: 'compensation.failure.count > 0'
        severity: critical
        channels: [dingtalk, pagerduty]
      - condition: 'saga.timed_out.count > 10 in 5m'
        severity: warning

5.3 常见踩坑清单

六、选型建议：什么时候用Saga，什么时候用TCC

七、总结：Saga不是银弹，而是分布式系统的生存哲学

Saga本质上是接受了分布式系统的不完美——我们无法保证同时成功，但我们保证最终通过补偿达到一致。它教会我们的不仅是一种技术模式，更是一种设计思维：

1. 假设失败：每一步都可能失败，为每一步设计补偿
2. 追求幂等：重复执行的结果应该和单次执行一样
3. 保持简单：复杂的Saga比复杂的2PC更难调试
4. 拥抱最终一致：在一致性和可用性之间找到业务容忍的平衡点

在微服务大行其道的今天，Saga模式已经不只是理论上的名词——它是每一个需要构建可靠分布式系统的工程师必须掌握的核心武器。从今天开始，别再谈分布式事务色变，拿起Saga这个武器，去构建真正可靠的微服务系统。