Seata 分布式事务

// Feign 客户端：调用库存服务
@FeignClient(name = "storage-service", path = "/storage")
public interface StorageFeignClient {

    @PostMapping("/deduct")
    Result deduct(@RequestParam("productId") Long productId,
                  @RequestParam("count") Integer count);
}

// Feign 客户端：调用账户服务
@FeignClient(name = "account-service", path = "/account")
public interface AccountFeignClient {

    @PostMapping("/debit")
    Result debit(@RequestParam("userId") Long userId,
                 @RequestParam("amount") BigDecimal amount);
}

@Service
public class OrderService {

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private StorageFeignClient storageClient;

    @Autowired
    private AccountFeignClient accountClient;

    /**
     * 下单核心方法——@GlobalTransactional 是全局事务入口。
     * 任何一步抛出异常，前面已提交的本地事务会被 Seata 自动回滚。
     */
    @GlobalTransactional(name = "create-order", rollbackFor = Exception.class)
    public void createOrder(Order order) {
        // 分支事务 1：订单库——创建订单
        orderMapper.insert(order);

        // 分支事务 2：远程调用库存服务——扣减库存
        Result storageResult = storageClient.deduct(
            order.getProductId(), order.getCount());
        if (!storageResult.isSuccess()) {
            throw new RuntimeException("库存扣减失败");
        }

        // 分支事务 3：远程调用账户服务——扣减余额
        Result accountResult = accountClient.debit(
            order.getUserId(), order.getAmount());
        if (!accountResult.isSuccess()) {
            throw new RuntimeException("余额扣减失败");
        }
    }
}

@RestController
@RequestMapping("/storage")
public class StorageController {

    @Autowired
    private StorageMapper storageMapper;

    @PostMapping("/deduct")
    @Transactional  // 本地事务
    public Result deduct(@RequestParam Long productId,
                         @RequestParam Integer count) {
        // Seata 代理会在此处自动：
        // 1. 查询前镜像
        // 2. 执行 SQL
        // 3. 查询后镜像
        // 4. 保存 UNDO LOG 到 undo_log 表
        // 5. 向 TC 注册分支事务
        int rows = storageMapper.deduct(productId, count);
        if (rows <= 0) {
            throw new RuntimeException("库存不足");
        }
        return Result.success();
    }
}

@RestController
@RequestMapping("/account")
public class AccountController {

    @Autowired
    private AccountMapper accountMapper;

    @PostMapping("/debit")
    @Transactional
    public Result debit(@RequestParam Long userId,
                        @RequestParam BigDecimal amount) {
        int rows = accountMapper.debit(userId, amount);
        if (rows <= 0) {
            throw new RuntimeException("余额不足");
        }
        return Result.success();
    }
}

// 测试代码：在 AccountController.debit() 中注入异常
@PostMapping("/debit")
@Transactional
public Result debit(@RequestParam Long userId,
                    @RequestParam BigDecimal amount) {
    int rows = accountMapper.debit(userId, amount);
    if (rows <= 0) {
        throw new RuntimeException("余额不足");
    }
    // === 故意抛出异常，模拟故障 ===
    throw new RuntimeException("模拟账户服务故障，测试全局回滚");
}

// TCC 接口：声明三阶段行为
@LocalTCC
public interface AccountTccAction {

    @TwoPhaseBusinessAction(
        name = "accountTccAction",
        commitMethod = "commit",
        rollbackMethod = "rollback"
    )
    boolean prepareDebit(
        @BusinessActionContextParameter(paramName = "userId") Long userId,
        @BusinessActionContextParameter(paramName = "amount") BigDecimal amount
    );

    boolean commit(BusinessActionContext context);

    boolean rollback(BusinessActionContext context);
}

@Component
public class AccountTccActionImpl implements AccountTccAction {

    @Autowired
    private AccountMapper accountMapper;

    /**
     * Try 阶段：冻结余额。
     * available_amount - N，frozen_amount + N。
     * 此时用户余额尚未真正减少，只是被"锁定"。
     */
    @Override
    @Transactional
    public boolean prepareDebit(Long userId, BigDecimal amount) {
        // 防悬挂：检查是否存在 Cancel 记录（Cancel 先于 Try 到达的情况）
        if (accountMapper.existsTccRecord(userId, "CANCEL")) {
            return true; // Cancel 已先行到达，拒绝执行 Try
        }

        // 幂等：检查是否已经 Try 过
        if (accountMapper.existsTccRecord(userId, "TRY")) {
            return true;
        }

        // 插入 TCC 控制记录（用于防悬挂和幂等）
        accountMapper.insertTccRecord(userId, amount, "TRY");

        // 冻结余额
        int rows = accountMapper.freeze(userId, amount);
        if (rows == 0) {
            throw new RuntimeException("余额不足");
        }
        return true;
    }

    /**
     * Confirm 阶段：确认扣款。
     * frozen_amount - N，真正扣减冻结的余额。
     */
    @Override
    @Transactional
    public boolean commit(BusinessActionContext context) {
        Long userId = Long.valueOf(context.getActionContext("userId").toString());
        BigDecimal amount = new BigDecimal(context.getActionContext("amount").toString());

        // 幂等：已确认过则直接返回
        if ("CONFIRM".equals(accountMapper.getTccRecordStatus(userId))) {
            return true;
        }

        accountMapper.commitDebit(userId, amount);
        accountMapper.updateTccRecordStatus(userId, "CONFIRM");
        return true;
    }

    /**
     * Cancel 阶段：解冻余额。
     * frozen_amount - N，available_amount + N。
     */
    @Override
    @Transactional
    public boolean rollback(BusinessActionContext context) {
        Long userId = Long.valueOf(context.getActionContext("userId").toString());
        BigDecimal amount = new BigDecimal(context.getActionContext("amount").toString());

        String status = accountMapper.getTccRecordStatus(userId);
        if (status == null) {
            // 空回滚：Try 从未执行，Cancel 却到达（网络超时后 TC 补发 Cancel）
            accountMapper.insertTccRecord(userId, BigDecimal.ZERO, "CANCEL_EMPTY");
            return true;
        }
        if ("CANCEL".equals(status)) {
            return true; // 幂等：已回滚过
        }

        accountMapper.unfreeze(userId, amount);
        accountMapper.updateTccRecordStatus(userId, "CANCEL");
        return true;
    }
}

@PostMapping("/deduct")
@Transactional
public Result deduct(@RequestParam Long productId,
                     @RequestParam Integer count) {
    // 先获取数据库行锁
    Integer currentStock = storageMapper.lockAndGetStock(productId);
    if (currentStock < count) {
        throw new RuntimeException("库存不足");
    }
    // 再执行扣减
    storageMapper.deduct(productId, count);
    return Result.success();
}

@GlobalTransactional(name = "batch-import", timeoutMills = 300000) // 5 分钟
public void batchImport(List<Record> records) { ... }

// Try: 防悬挂
if (tccMapper.existsRecord(bizId, "CANCEL")) {
    return true; // Cancel 已先行到达，拒绝执行 Try
}

// Cancel: 防空回滚
if (tccMapper.getStatus(bizId) == null) {
    tccMapper.insertRecord(bizId, "CANCEL_EMPTY");
    return true;
}

// Try / Confirm / Cancel: 幂等
if (tccMapper.getStatus(bizId).equals(currentPhase)) {
    return true;
}

// 错误示例
@Transactional  // 本地事务
@GlobalTransactional  // 全局事务
public void createOrder(Order order) {
    orderMapper.insert(order);      // 受两个事务管理
    accountClient.debit(...);        // 只受全局事务管理
}
// 当 debit 失败时，@Transactional 已提交或已在独立事务中，
// 可能出现 insert 未被正确回滚的情况

// 正确做法
@Service
public class OrderService {
    @GlobalTransactional  // TM：只标注全局事务
    public void createOrder(Order order) { ... }
}

@RestController
public class StorageController {
    @PostMapping("/deduct")
    @Transactional  // RM：只标注本地事务
    public Result deduct(...) { ... }
}