Saga 事务模式#

// 补偿操作示例：支付失败的补偿不是"删除支付记录"
public class PaymentSaga {

    // T4: 支付
    public boolean pay(Order order, int amount) {
        paymentService.deduct(order.getUserId(), amount);
        order.setStatus(OrderStatus.PAID);
        orderMapper.update(order);
        return true;
    }

    // c4: 支付失败的补偿 = 退款
    public boolean compensatePay(Order order, int amount) {
        // 补偿不是"删除支付记录"
        // 补偿是"执行一次退款"
        paymentService.refund(order.getUserId(), amount);
        order.setStatus(OrderStatus.PAY_FAILED);
        orderMapper.update(order);
        return true;
    }
}

@Service
public class OrderSagaOrchestrator {

    @Autowired
    private OrderService orderService;
    @Autowired
    private InventoryService inventoryService;
    @Autowired
    private CouponService couponService;
    @Autowired
    private PaymentService paymentService;

    @Autowired
    private SagaStateMapper sagaMapper;

    /**
     * 执行订单履约 Saga
     */
    public boolean execute(OrderDTO order) {
        String sagaId = UUID.randomUUID().toString();
        SagaContext context = new SagaContext(sagaId);

        try {
            // T1: 创建订单
            step(context, "createOrder", () -> orderService.create(order));

            // T2: 预扣库存
            step(context, "reserveInventory", () -> inventoryService.reserve(order.getGoodsId(), order.getCount()));

            // T3: 锁定优惠券
            if (order.getCouponId() != null) {
                step(context, "lockCoupon", () -> couponService.lock(order.getUserId(), order.getCouponId()));
            }

            // T4: 支付
            step(context, "pay", () -> paymentService.pay(order.getUserId(), order.getAmount()));

            // T5-T12: 其他步骤...

            return true;

        } catch (SagaStepException e) {
            // 某个步骤失败，执行补偿链
            compensate(context);
            return false;
        }
    }

    private void step(SagaContext ctx, String stepName, Supplier<Boolean> action) {
        // 记录步骤开始
        ctx.recordStep(stepName, SagaStepStatus.EXECUTING);
        try {
            boolean result = action.get();
            if (result) {
                ctx.recordStep(stepName, SagaStepStatus.COMPLETED);
            } else {
                throw new SagaStepException(stepName + " returned false");
            }
        } catch (Exception e) {
            ctx.recordStep(stepName, SagaStepStatus.FAILED);
            throw new SagaStepException(stepName + " failed", e);
        }
    }

    /**
     * 补偿链：倒序执行已成功步骤的补偿操作
     */
    private void compensate(SagaContext ctx) {
        List<String> completedSteps = ctx.getCompletedSteps(); // 倒序

        for (String stepName : completedSteps) {
            try {
                switch (stepName) {
                    case "pay":
                        paymentService.refund(ctx.getOrder().getUserId(), ctx.getOrder().getAmount());
                        break;
                    case "lockCoupon":
                        couponService.unlock(ctx.getOrder().getUserId(), ctx.getOrder().getCouponId());
                        break;
                    case "reserveInventory":
                        inventoryService.release(ctx.getOrder().getGoodsId(), ctx.getOrder().getCount());
                        break;
                    case "createOrder":
                        orderService.cancel(ctx.getOrder().getId());
                        break;
                }
                ctx.recordCompensation(stepName, SagaCompensationStatus.DONE);
            } catch (Exception e) {
                ctx.recordCompensation(stepName, SagaCompensationStatus.FAILED);
                log.error("补偿 step={} 失败，sagaId={}", stepName, ctx.getSagaId(), e);
                // 补偿失败需要重试或人工介入
                throw e;
            }
        }
    }
}

// 订单服务：发布创建事件
@Service
public class OrderService {

    @Autowired
    private EventPublisher eventPublisher;

    public void createOrder(Order order) {
        orderMapper.insert(order);
        eventPublisher.publish("OrderCreatedEvent", order);
    }

    // 监听库存释放事件
    @Subscribe("InventoryReleasedEvent")
    public void onInventoryReleased(InventoryReleasedEvent event) {
        orderService.updateStatus(event.getOrderId(), OrderStatus.CANCELLED);
    }
}

// 库存服务：监听订单创建，执行业务
@Service
public class InventoryService {

    @Autowired
    private EventPublisher eventPublisher;

    @Subscribe("OrderCreatedEvent")
    public void onOrderCreated(OrderCreatedEvent event) {
        boolean reserved = inventoryService.reserve(event.getGoodsId(), event.getCount());
        if (reserved) {
            eventPublisher.publish("InventoryReservedEvent", event);
        } else {
            eventPublisher.publish("InventoryReserveFailedEvent", event);
        }
    }

    @Subscribe("PaymentFailedEvent")
    public void onPaymentFailed(PaymentFailedEvent event) {
        // 补偿：释放预扣库存
        inventoryService.release(event.getGoodsId(), event.getCount());
        eventPublisher.publish("InventoryReleasedEvent", event);
    }
}