Saga 事务#

public class SagaOrchestrator {

    private final SagaSteps steps;
    private final Map<String, Object> context = new HashMap<>();

    public SagaOrchestrator(SagaSteps steps) {
        this.steps = steps;
    }

    /**
     * 执行 Saga 流程
     *
     * @return 执行结果
     */
    public SagaResult execute() {
        List<SagaStep> executedSteps = new ArrayList<>();

        try {
            // 按顺序执行每个步骤
            for (SagaStep step : steps.getSteps()) {
                StepResult result = executeStep(step);

                if (!result.isSuccess()) {
                    // 执行失败，启动补偿
                    return compensateAndReturn(executedSteps, result);
                }

                executedSteps.add(step);
            }

            // 所有步骤都执行成功
            return SagaResult.success(context);

        } catch (Exception e) {
            // 异常情况，执行补偿
            return compensateAndReturn(executedSteps, StepResult.failure(e.getMessage()));
        }
    }

    private StepResult executeStep(SagaStep step) {
        log.info("执行 Saga 步骤: {}", step.getName());

        try {
            Object result = step.getAction().execute(context);
            step.setStatus(StepStatus.EXECUTED);
            step.setResult(result);

            log.info("步骤 {} 执行成功", step.getName());
            return StepResult.success(result);

        } catch (Exception e) {
            log.error("步骤 {} 执行失败: {}", step.getName(), e.getMessage());
            step.setStatus(StepStatus.FAILED);
            return StepResult.failure(e.getMessage());
        }
    }

    private SagaResult compensateAndReturn(List<SagaStep> executedSteps, StepResult failureReason) {
        log.warn("Saga 执行失败，开始补偿，已执行步骤数: {}", executedSteps.size());

        List<String> compensatedSteps = new ArrayList<>();
        List<String> failedCompensations = new ArrayList<>();

        // 逆序执行补偿
        for (int i = executedSteps.size() - 1; i >= 0; i--) {
            SagaStep step = executedSteps.get(i);

            if (step.getCompensation() == null) {
                log.warn("步骤 {} 没有配置补偿，跳过", step.getName());
                continue;
            }

            try {
                log.info("执行补偿: {}", step.getName());
                step.getCompensation().execute(context);
                step.setStatus(StepStatus.COMPENSATED);
                compensatedSteps.add(step.getName());

            } catch (Exception e) {
                log.error("补偿 {} 失败: {}", step.getName(), e.getMessage());
                step.setStatus(StepStatus.COMPENSATION_FAILED);
                failedCompensations.add(step.getName());
            }
        }

        return SagaResult.builder()
            .success(false)
            .failedStep(failureReason.getFailedStep())
            .failureReason(failureReason.getReason())
            .compensatedSteps(compensatedSteps)
            .failedCompensations(failedCompensations)
            .build();
    }
}

@Data
@Builder
public class SagaStep {

    private String name;                    // 步骤名称
    private SagaAction action;              // 正向操作
    private SagaCompensation compensation;   // 补偿操作
    private int retryCount = 0;              // 重试次数
    private int maxRetries = 3;              // 最大重试次数

    // 执行状态
    private StepStatus status;
    private Object result;
    private String errorMessage;

    public enum StepStatus {
        PENDING,
        EXECUTING,
        EXECUTED,
        FAILED,
        COMPENSATING,
        COMPENSATED,
        COMPENSATION_FAILED
    }
}

public class SagaSteps {

    private final List<SagaStep> steps = new ArrayList<>();

    public SagaSteps addStep(String name, SagaAction action, SagaCompensation compensation) {
        steps.add(SagaStep.builder()
            .name(name)
            .action(action)
            .compensation(compensation)
            .build());
        return this;
    }

    public SagaSteps addStep(SagaStep step) {
        steps.add(step);
        return this;
    }

    public List<SagaStep> getSteps() {
        return Collections.unmodifiableList(steps);
    }

    // 静态工厂方法：创建订单 Saga
    public static SagaSteps createOrderSaga() {
        return new SagaSteps()
            .addStep("创建订单",
                context -> {
                    Long userId = (Long) context.get("userId");
                    Order order = orderService.createOrder(userId);
                    context.put("orderId", order.getId());
                    return order;
                },
                context -> {
                    Long orderId = (Long) context.get("orderId");
                    orderService.cancelOrder(orderId);
                })

            .addStep("冻结库存",
                context -> {
                    Long skuId = (Long) context.get("skuId");
                    Integer quantity = (Integer) context.get("quantity");
                    inventoryService.freezeStock(skuId, quantity);
                    return null;
                },
                context -> {
                    Long skuId = (Long) context.get("skuId");
                    Integer quantity = (Integer) context.get("quantity");
                    inventoryService.unfreezeStock(skuId, quantity);
                })

            .addStep("预占优惠",
                context -> {
                    Long userId = (Long) context.get("userId");
                    String couponId = (String) context.get("couponId");
                    promotionService.reserveCoupon(userId, couponId);
                    return null;
                },
                context -> {
                    Long userId = (Long) context.get("userId");
                    String couponId = (String) context.get("couponId");
                    promotionService.releaseCoupon(userId, couponId);
                })

            .addStep("扣减余额",
                context -> {
                    Long userId = (Long) context.get("userId");
                    BigDecimal amount = (BigDecimal) context.get("amount");
                    accountService.deductBalance(userId, amount);
                    return null;
                },
                context -> {
                    Long userId = (Long) context.get("userId");
                    BigDecimal amount = (BigDecimal) context.get("amount");
                    accountService.refundBalance(userId, amount);
                })

            .addStep("调用物流",
                context -> {
                    Long orderId = (Long) context.get("orderId");
                    String address = (String) context.get("address");
                    String logisticsNo = logisticsService.createShipment(orderId, address);
                    context.put("logisticsNo", logisticsNo);
                    return logisticsNo;
                },
                context -> {
                    String logisticsNo = (String) context.get("logisticsNo");
                    logisticsService.cancelShipment(logisticsNo);
                });
    }
}

@Data
@Builder
public class SagaResult {

    private boolean success;                    // 是否全部成功
    private String failedStep;                  // 失败的步骤
    private String failureReason;               // 失败原因
    private List<String> compensatedSteps;       // 已补偿的步骤
    private List<String> failedCompensations;   // 补偿失败的步骤
    private Map<String, Object> context;        // 执行上下文

    public static SagaResult success(Map<String, Object> context) {
        return SagaResult.builder()
            .success(true)
            .context(context)
            .compensatedSteps(Collections.emptyList())
            .failedCompensations(Collections.emptyList())
            .build();
    }

    public boolean hasCompensationFailure() {
        return failedCompensations != null && !failedCompensations.isEmpty();
    }
}

@Service
public class OrderService {

    @Autowired
    private OrderRepository orderRepo;

    @Autowired
    private InventoryService inventoryService;

    @Autowired
    private AccountService accountService;

    @Autowired
    private PromotionService promotionService;

    @Autowired
    private LogisticsService logisticsService;

    /**
     * 创建订单 - 使用 Saga 模式
     */
    public Order createOrderWithSaga(Long userId, Long skuId, Integer quantity,
                                      BigDecimal amount, String couponId, String address) {

        // 1. 准备 Saga 上下文
        Map<String, Object> context = new HashMap<>();
        context.put("userId", userId);
        context.put("skuId", skuId);
        context.put("quantity", quantity);
        context.put("amount", amount);
        context.put("couponId", couponId);
        context.put("address", address);

        // 2. 创建并执行 Saga
        SagaSteps sagaSteps = SagaSteps.createOrderSaga();
        SagaOrchestrator orchestrator = new SagaOrchestrator(sagaSteps);

        SagaResult result = orchestrator.execute();

        // 3. 处理结果
        if (result.isSuccess()) {
            Long orderId = (Long) context.get("orderId");
            return orderRepo.findById(orderId).orElseThrow();
        }

        // 4. 处理失败情况
        log.error("Saga 执行失败: failedStep={}, reason={}, compensated={}",
            result.getFailedStep(), result.getFailureReason(), result.getCompensatedSteps());

        if (result.hasCompensationFailure()) {
            // 补偿失败，需要告警和人工介入
            alertService.sendAlert("Saga补偿失败",
                "failedStep=" + result.getFailedStep() +
                ", failedCompensations=" + result.getFailedCompensations());
        }

        throw new OrderCreationException("订单创建失败: " + result.getFailureReason());
    }
}

// 错误示例：补偿逻辑与正向操作不一致
SagaStep step = SagaStep.builder()
    .name("扣减库存")
    .action(ctx -> inventoryService.decreaseStock(skuId, 10))  // 扣减 10
    .compensation(ctx -> inventoryService.increaseStock(skuId, 5)) // 补偿只返还 5
    .build();

// 错误示例：补偿失败直接抛异常
compensation.execute(context); // 抛异常，不处理

// 错误示例：正向操作没有幂等处理
action.execute(ctx -> inventoryService.decreaseStock(skuId, quantity));

// 错误示例：补偿操作没有幂等处理
compensation.execute(ctx -> inventoryService.increaseStock(skuId, quantity));