分布式事务解决方案

@Service
public class OrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private InventoryFeignClient inventoryClient;
    
    @Autowired
    private PaymentFeignClient paymentClient;
    
    // 2PC 协调者
    public void createOrderWith2PC(OrderRequest request) {
        // 阶段一：Prepare
        boolean orderPrepared = orderRepository.prepare(request);
        boolean inventoryPrepared = inventoryClient.prepare(request.getSkuId(), request.getQuantity());
        boolean paymentPrepared = paymentClient.prepare(request.getUserId(), request.getAmount());
        
        // 阶段二：Commit 或 Rollback
        if (orderPrepared && inventoryPrepared && paymentPrepared) {
            orderRepository.commit();
            inventoryClient.commit();
            paymentClient.commit();
        } else {
            orderRepository.rollback();
            inventoryClient.rollback();
            paymentClient.rollback();
            throw new DistributedTransactionException("事务回滚");
        }
    }
}

@Service
public class OrderTccService {
    
    @Autowired
    private OrderTccAction orderTccAction;
    
    @Autowired
    private InventoryTccAction inventoryTccAction;
    
    @GlobalTransactional  // Seata 注解
    public void createOrder(OrderRequest request) {
        // Try 阶段
        boolean orderTry = orderTccAction.tryCreateOrder(request);
        boolean inventoryTry = inventoryTccAction.tryDeduct(request.getSkuId(), request.getQuantity());
        
        if (!orderTry || !inventoryTry) {
            throw new BusinessException("资源不足");
        }
        
        // Confirm 由框架自动调用（或业务代码显式调用）
        // 如果 Confirm 失败，不断重试
    }
}

@Component
public class InventoryTccAction {
    
    @Autowired
    private InventoryRepository inventoryRepository;
    
    @Autowired
    private InventoryFreezeRepository freezeRepository;
    
    // Try：冻结库存
    public boolean tryDeduct(String skuId, int quantity) {
        Inventory inventory = inventoryRepository.findBySkuId(skuId);
        
        // 检查库存
        if (inventory.getAvailable() < quantity) {
            return false;
        }
        
        // 冻结库存
        inventory.setAvailable(inventory.getAvailable() - quantity);
        inventory.setFrozen(inventory.getFrozen() + quantity);
        inventoryRepository.save(inventory);
        
        // 记录冻结记录（用于幂等和回滚）
        freezeRepository.save(new InventoryFreeze(skuId, quantity));
        
        return true;
    }
    
    // Confirm：确认扣减
    public boolean confirmDeduct(String skuId, int quantity) {
        Inventory inventory = inventoryRepository.findBySkuId(skuId);
        
        // 扣减冻结库存
        inventory.setFrozen(inventory.getFrozen() - quantity);
        inventoryRepository.save(inventory);
        
        // 删除冻结记录
        freezeRepository.deleteBySkuId(skuId);
        
        return true;
    }
    
    // Cancel：释放冻结
    public boolean cancelDeduct(String skuId, int quantity) {
        Inventory inventory = inventoryRepository.findBySkuId(skuId);
        
        // 释放冻结库存
        inventory.setAvailable(inventory.getAvailable() + quantity);
        inventory.setFrozen(inventory.getFrozen() - quantity);
        inventoryRepository.save(inventory);
        
        // 删除冻结记录
        freezeRepository.deleteBySkuId(skuId);
        
        return true;
    }
}

// Confirm 和 Cancel 必须保证幂等（可能重复调用）
public boolean confirmDeduct(String skuId, int quantity) {
    // 检查是否已处理
    if (freezeRepository.findBySkuId(skuId) == null) {
        return true;  // 已处理，直接返回
    }
    // 执行业务
}

// Try 超时未执行，Cancel 先执行
public boolean cancelDeduct(String skuId, int quantity) {
    // 检查是否有冻结记录
    if (freezeRepository.findBySkuId(skuId) == null) {
        // 空回滚，记录回滚状态，防止后续 Try 执行
        idempotentRepository.recordCancel(skuId);
        return true;
    }
    // 正常回滚
}

// Cancel 先执行，Try 后到达
public boolean tryDeduct(String skuId, int quantity) {
    // 检查是否已回滚
    if (idempotentRepository.isCancelled(skuId)) {
        return false;  // 拒绝执行
    }
    // 正常执行
}

@Service
public class OrderServiceWithMessageTable {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private MessageRepository messageRepository;
    
    @Transactional
    public void createOrder(OrderRequest request) {
        // 1. 创建订单
        Order order = new Order();
        order.setUserId(request.getUserId());
        order.setAmount(request.getAmount());
        order.setStatus(OrderStatus.CREATED);
        orderRepository.save(order);
        
        // 2. 记录消息（同一事务）
        Message message = new Message();
        message.setTopic("order_created");
        message.setBody(JsonUtils.toJson(new OrderCreatedEvent(order.getId())));
        message.setStatus(MessageStatus.PENDING);
        messageRepository.save(message);
        
        // 事务提交后，订单和消息同时成功或同时失败
    }
}

@Component
public class MessageSender {
    
    @Autowired
    private MessageRepository messageRepository;
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    // 定时扫描，每 5 秒执行一次
    @Scheduled(fixedDelay = 5000)
    public void sendPendingMessages() {
        List<Message> pendingMessages = messageRepository
            .findByStatusAndRetryLessThan(MessageStatus.PENDING, 3);
        
        for (Message message : pendingMessages) {
            try {
                SendResult result = rocketMQTemplate.syncSend(
                    message.getTopic(), 
                    message.getBody()
                );
                
                if (result.getSendStatus() == SendStatus.SEND_OK) {
                    message.setStatus(MessageStatus.SENT);
                    messageRepository.save(message);
                }
            } catch (Exception e) {
                message.setRetryCount(message.getRetryCount() + 1);
                messageRepository.save(message);
            }
        }
    }
}

@Component
@RocketMQMessageListener(topic = "order_created", consumerGroup = "inventory_consumer")
public class InventoryConsumer implements RocketMQListener<OrderCreatedEvent> {
    
    @Autowired
    private InventoryService inventoryService;
    
    @Override
    public void onMessage(OrderCreatedEvent event) {
        // 幂等检查
        if (inventoryService.isProcessed(event.getOrderId())) {
            return;
        }
        
        // 扣减库存
        inventoryService.deduct(event.getOrderId());
        
        // 记录已处理
        inventoryService.markProcessed(event.getOrderId());
    }
}

@Service
public class SagaOrchestrator {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private PaymentService paymentService;
    
    public void executeOrderSaga(OrderRequest request) {
        String sagaId = generateSagaId();
        
        try {
            // Step 1: 创建订单
            Long orderId = orderService.createOrder(request);
            recordSagaStep(sagaId, "CREATE_ORDER", orderId);
            
            // Step 2: 扣减库存
            boolean inventoryResult = inventoryService.deduct(request.getSkuId(), request.getQuantity());
            if (!inventoryResult) {
                throw new InventoryInsufficientException();
            }
            recordSagaStep(sagaId, "DEDUCT_INVENTORY", null);
            
            // Step 3: 支付
            boolean paymentResult = paymentService.pay(request.getUserId(), request.getAmount());
            if (!paymentResult) {
                throw new PaymentFailedException();
            }
            recordSagaStep(sagaId, "PAY", null);
            
            // Saga 完成
            completeSaga(sagaId);
            
        } catch (Exception e) {
            // 执行补偿
            compensateSaga(sagaId);
        }
    }
    
    private void compensateSaga(String sagaId) {
        List<SagaStep> steps = getSagaSteps(sagaId);
        
        // 反向执行补偿
        for (int i = steps.size() - 1; i >= 0; i--) {
            SagaStep step = steps.get(i);
            
            switch (step.getStepName()) {
                case "PAY":
                    paymentService.refund(step.getContext());
                    break;
                case "DEDUCT_INVENTORY":
                    inventoryService.rollbackDeduct(step.getContext());
                    break;
                case "CREATE_ORDER":
                    orderService.cancelOrder(step.getContext());
                    break;
            }
        }
    }
}

// 启动类
@SpringBootApplication
@EnableAutoDataSourceProxy  // Seata 数据源代理
public class OrderApplication {
    public static void main(String[] args) {
        SpringApplication.run(OrderApplication.class, args);
    }
}

// 业务代码
@Service
public class OrderService {
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private InventoryFeignClient inventoryFeignClient;
    
    @Autowired
    private AccountFeignClient accountFeignClient;
    
    @GlobalTransactional(name = "create-order", rollbackFor = Exception.class)
    public void createOrder(Order order) {
        // 1. 创建订单
        orderMapper.insert(order);
        
        // 2. 扣减库存
        inventoryFeignClient.deduct(order.getSkuId(), order.getCount());
        
        // 3. 扣减账户余额
        accountFeignClient.debit(order.getUserId(), order.getMoney());
        
        // 任意步骤异常，自动回滚
    }
}