从某电商平台订单超卖事故说起程序员如何用悲观锁配合超时机制与锁顺序策略避免死锁并保障高并发场景下的数据一致性

那惊天动地的一次”超卖”

2023年6月18日凌晨，某知名电商平台发生了一起堪称教科书级别的事故。当时平台正在进行年中大促活动，一款限量发售的限量版球鞋库存只有100双，但后台日志显示实际卖出了347双。这意味着平台少收了247双鞋的钱，却多发了247双鞋出去。

事件发生后的复盘会上，技术总监的脸色比机房里的服务器还绿。问题的根源很简单，却又极其致命：在高并发场景下，多个用户几乎同时读取到了”库存还有”的状态，然后都通过了校验，最终导致了超卖。

这不是什么惊天大案，但却是很多开发者在学习并发控制时都会遇到的经典场景。今天，我们就来聊聊如何用最朴素的悲观锁，配合超时机制和锁顺序策略，彻底解决这个让无数程序员头秃的问题。

---

悲观锁：一个”先占后算”的老实人

悲观锁这个概念，听起来有点抽象，但其实很好理解。想象你在食堂排队打饭，窗口只有一个，你担心如果有人插队你就打不到饭了，所以你决定”保守”一点——在打饭之前先占住窗口，打完饭再走。

在数据库领域，悲观锁就是这么个思路：在操作数据之前，先加锁，确保只有自己能看到和修改这段数据，等事情办完再解锁。

让我用Java代码给你演示一下，一个没有加锁的库存扣减系统长什么样：

/**
 * 没有加锁的库存服务 - 超卖版
 */
@Service
public class StockService {
    
    @Autowired
    private StockMapper stockMapper;
    
    /**
     * 扣减库存 - 这个版本有严重的并发问题
     * @param stockId 库存ID
     * @param quantity 扣减数量
     * @return 是否扣减成功
     */
    @Transactional
    public boolean deductStock(Long stockId, int quantity) {
        // 第一步：查询当前库存
        Stock stock = stockMapper.selectById(stockId);
        
        // 第二步：检查库存是否充足
        if (stock.getRemaining() < quantity) {
            return false;
        }
        
        // 第三步：扣减库存
        // 这里有个问题：如果两个线程同时执行到这里
        // 它们都会通过库存检查，然后都会扣减库存
        // 最终导致库存扣减为负数，也就是超卖
        stock.setRemaining(stock.getRemaining() - quantity);
        stockMapper.updateById(stock);
        
        return true;
    }
}

看到问题了吗？当两个线程几乎同时执行到”检查库存是否充足”这一步时，它们看到的库存都是一样的。如果库存是5，两个线程都发现5>=1，然后都扣减1，结果库存变成了3，但实际上应该只扣减1，变成4。

这就是经典的“读-改-写”竞态条件。在高并发场景下，这种情况会以指数级增长。

引入悲观锁后的改造

/**
 * 使用悲观锁的库存服务 - 改进版
 */
@Service
public class StockServiceWithPessimisticLock {
    
    @Autowired
    private StockMapper stockMapper;
    
    /**
     * 扣减库存 - 使用悲观锁
     * @param stockId 库存ID
     * @param quantity 扣减数量
     * @return 是否扣减成功
     */
    @Transactional
    public boolean deductStock(Long stockId, int quantity) {
        // 使用 SELECT ... FOR UPDATE 获取悲观锁
        // 这会锁定该行记录，直到事务提交
        Stock stock = stockMapper.selectForUpdate(stockId);
        
        // 到这里为止，其他事务无法读取或修改这行记录
        // 因为只有当前事务持有这把锁
        
        if (stock.getRemaining() < quantity) {
            return false;
        }
        
        // 扣减库存
        stock.setRemaining(stock.getRemaining() - quantity);
        stockMapper.updateById(stock);
        
        return true;
    }
}

在MyBatis中，selectForUpdate对应的SQL长这样：

SELECT * FROM stock WHERE id = #{id} FOR UPDATE;

这条SQL会在查询的同时，对该行加上排他锁（Exclusive Lock）。在MySQL的InnoDB引擎中，这是一个行级锁，意味着只会锁住这一行记录，其他行的查询不受影响。

悲观锁的工作原理

让我用一个更直观的方式来解释：

时间线：
Thread-1: [查库存] → [加锁] → [检查] → [扣减] → [提交] → [解锁]
Thread-2: [等待加锁...] → [获取锁] → [查库存] → [检查] → [扣减] → [提交] → [解锁]

关键在于，Thread-2必须等待Thread-1完成并提交后，才能获取锁并执行后续操作。这样就保证了库存数据的一致性。

---

但悲观锁有个问题：太”霸道”了

悲观锁虽然能保证数据一致性，但它有个很大的缺点：性能。

想象一下，如果一千个人同时来买同一款限量球鞋，悲观锁会让他们排成一队，一个一个来。前面的人慢一点，后面的人就得等很久。在高并发场景下，这会导致：

1. 大量请求被阻塞，服务器压力反而更大
2. 用户体验极差，等待时间过长
3. 数据库连接池可能被耗尽

为了解决这个问题，我们需要引入超时机制。

---

超时机制：给悲观锁加个”闹钟”

超时机制的核心思想是：如果等待锁的时间超过了预设值，就放弃操作，返回给用户一个友好的提示。

/**
 * 使用悲观锁 + 超时机制的库存服务
 */
@Service
public class StockServiceWithTimeout {
    
    @Autowired
    private StockMapper stockMapper;
    
    /**
     * 扣减库存 - 带超时机制
     * @param stockId 库存ID
     * @param quantity 扣减数量
     * @param lockWaitTimeout 锁等待超时时间（毫秒）
     * @return 操作结果
     */
    public OperationResult deductStock(Long stockId, int quantity, long lockWaitTimeout) {
        Connection connection = null;
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;
        
        try {
            // 获取数据库连接
            connection = DataSourceUtils.getConnection();
            
            // 设置事务隔离级别为可重复读（防止幻读）
            connection.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);
            
            // 关闭自动提交，手动管理事务
            connection.setAutoCommit(false);
            
            // 设置锁等待超时时间
            // MySQL 5.5.3+ 支持 innodb_lock_wait_timeout
            // 这里通过 SQL 语句设置
            connection.createStatement().execute(
                "SET innodb_lock_wait_timeout = " + (lockWaitTimeout / 1000)
            );
            
            // 执行带锁的查询
            String sql = "SELECT id, remaining FROM stock WHERE id = ? FOR UPDATE";
            preparedStatement = connection.prepareStatement(sql);
            preparedStatement.setLong(1, stockId);
            resultSet = preparedStatement.executeQuery();
            
            if (resultSet.next()) {
                int remaining = resultSet.getInt("remaining");
                
                if (remaining < quantity) {
                    // 库存不足，回滚并返回
                    connection.rollback();
                    return OperationResult.insufficientStock();
                }
                
                // 扣减库存
                int updateSql = "UPDATE stock SET remaining = remaining - ? WHERE id = ? AND remaining >= ?";
                PreparedStatement updateStmt = connection.prepareStatement(updateSql);
                updateStmt.setInt(1, quantity);
                updateStmt.setLong(2, stockId);
                updateStmt.setInt(3, quantity);
                
                int affectedRows = updateStmt.executeUpdate();
                
                if (affectedRows == 0) {
                    // 检查是否因为并发更新导致库存不足
                    connection.rollback();
                    return OperationResult.insufficientStock();
                }
                
                // 提交事务
                connection.commit();
                return OperationResult.success();
            }
            
            // 记录不存在，回滚
            connection.rollback();
            return OperationResult.recordNotFound();
            
        } catch (SQLException e) {
            // 如果是锁等待超时
            if (e.getErrorCode() == 1205) { // MySQL error code for lock wait timeout
                // 1205 is ER_LOCK_WAIT_TIMEOUT
                if (connection != null) {
                    try {
                        connection.rollback();
                    } catch (SQLException ex) {
                        // 忽略回滚异常
                    }
                }
                return OperationResult.lockTimeout();
            }
            
            // 其他异常，回滚
            if (connection != null) {
                try {
                    connection.rollback();
                } catch (SQLException ex) {
                    // 忽略回滚异常
                }
            }
            
            throw new RuntimeException("扣减库存失败", e);
        } finally {
            // 关闭资源
            if (resultSet != null) {
                try { resultSet.close(); } catch (SQLException e) { /* ignore */ }
            }
            if (preparedStatement != null) {
                try { preparedStatement.close(); } catch (SQLException e) { /* ignore */ }
            }
            if (connection != null) {
                try { connection.close(); } catch (SQLException e) { /* ignore */ }
            }
        }
    }
}

/**
 * 操作结果封装类
 */
class OperationResult {
    private final boolean success;
    private final String message;
    private final int errorCode;
    
    private OperationResult(boolean success, String message, int errorCode) {
        this.success = success;
        this.message = message;
        this.errorCode = errorCode;
    }
    
    public static OperationResult success() {
        return new OperationResult(true, "库存扣减成功", 0);
    }
    
    public static OperationResult insufficientStock() {
        return new OperationResult(false, "库存不足", 1);
    }
    
    public static OperationResult lockTimeout() {
        return new OperationResult(false, "系统繁忙，请稍后重试", 2);
    }
    
    public static OperationResult recordNotFound() {
        return new OperationResult(false, "记录不存在", 3);
    }
    
    public boolean isSuccess() { return success; }
    public String getMessage() { return message; }
    public int getErrorCode() { return errorCode; }
}

超时机制带来的改进

有了超时机制，当锁等待时间过长时，请求会快速失败，而不是无限期阻塞。这样可以：

1. 避免数据库连接被长时间占用
2. 给用户更快的反馈：”系统繁忙，请稍后重试”
3. 防止线程池被耗尽

但是，超时机制也不是万能的。如果并发量非常大，大量的请求同时等待锁，即使设置了超时时间，也会导致大量的请求超时失败，用户体验仍然不好。

这时候，我们就需要更高级的解决方案：锁顺序策略。

---

锁顺序策略：给”霸道”加个”规矩”

锁顺序策略解决的是一个更隐蔽但更致命的问题：死锁。

什么是死锁？

死锁是指两个或多个线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。

让我用一个经典的例子来说明：

场景：用户A想从账户1转账到账户2，用户B想从账户2转账到账户1

Thread-A:                          Thread-B:
1. 获取账户1的锁                   1. 获取账户2的锁
2. 尝试获取账户2的锁  ← 等待       2. 尝试获取账户1的锁  ← 等待
                                  （Thread-B在等Thread-A释放账户1的锁）
（Thread-A在等Thread-B释放账户2的锁）

结果：两个线程互相等待，永远无法完成！

这就是死锁。在高并发的电商系统中，如果有多个库存项需要同时扣减，或者多个业务操作需要获取多个锁，死锁的风险就会大大增加。

锁顺序策略的原理

锁顺序策略的核心思想是：给所有的锁定义一个全局唯一的顺序，所有线程都按照这个顺序获取锁。

这样，即使两个线程同时请求多个锁，它们也会按照相同的顺序获取，永远不会出现互相等待的情况。

/**
 * 使用锁顺序策略避免死锁的库存服务
 */
@Service
public class StockServiceWithLockOrdering {
    
    @Autowired
    private StockMapper stockMapper;
    
    /**
     * 扣减多个库存 - 使用锁顺序策略避免死锁
     * @param stockIds 库存ID列表
     * @param quantityMap 每个库存ID对应的扣减数量
     * @return 是否全部扣减成功
     */
    @Transactional
    public boolean deductMultipleStocks(Map<Long, Integer> quantityMap) {
        if (quantityMap == null || quantityMap.isEmpty()) {
            return true;
        }
        
        // 关键步骤：按照ID的升序排列锁
        // 这样可以确保所有线程都以相同的顺序获取锁
        List<Long> sortedStockIds = new ArrayList<>(quantityMap.keySet());
        Collections.sort(sortedStockIds);
        
        try {
            // 按顺序获取每个库存的锁
            for (Long stockId : sortedStockIds) {
                Stock stock = stockMapper.selectForUpdate(stockId);
                
                int quantity = quantityMap.get(stockId);
                if (stock.getRemaining() < quantity) {
                    // 库存不足，回滚
                    throw new InsufficientStockException(
                        "库存ID " + stockId + " 不足，当前库存: " + stock.getRemaining() 
                        + ", 需要: " + quantity
                    );
                }
            }
            
            // 所有锁都已获取，开始扣减库存
            for (Map.Entry<Long, Integer> entry : quantityMap.entrySet()) {
                Long stockId = entry.getKey();
                int quantity = entry.getValue();
                
                // 使用 UPDATE ... WHERE 来保证原子性
                int affectedRows = stockMapper.deductStock(stockId, quantity);
                
                if (affectedRows == 0) {
                    // 更新失败，说明库存可能已被其他事务修改
                    throw new ConcurrentUpdateException(
                        "库存ID " + stockId + " 并发更新失败"
                    );
                }
            }
            
            return true;
            
        } catch (Exception e) {
            // 任何异常都会触发事务回滚
            throw e;
        }
    }
}

为什么锁顺序能避免死锁？

让我们用之前的死锁例子来说明：

死锁场景（无锁顺序）：
Thread-A: 获取账户1的锁 → 获取账户2的锁（等待）
Thread-B: 获取账户2的锁 → 获取账户1的锁（等待）
结果：死锁

使用锁顺序后：
Thread-A: 获取账户1的锁 → 获取账户2的锁（如果线程B先获取了账户1，这里会等待）
Thread-B: 获取账户1的锁 → 获取账户2的锁（如果线程A先获取了账户1，这里会等待）
结果：不会死锁，因为两个线程都按照相同的顺序（账户1 → 账户2）获取锁

关键在于：只要所有线程都按照相同的顺序获取锁，就不可能出现循环等待，也就不会发生死锁。

---

悲观锁 + 超时机制 + 锁顺序：三位一体的完整方案

现在，我们把前面讲的所有技术组合起来，形成一个完整的解决方案：

/**
 * 完整的并发库存扣减服务
 * 结合悲观锁、超时机制和锁顺序策略
 */
@Service
public class StockServiceComplete {
    
    @Autowired
    private StockMapper stockMapper;
    
    @Value("${stock.lock.wait.timeout:3000}")
    private long lockWaitTimeout;
    
    /**
     * 扣减库存 - 完整版本
     * @param stockId 库存ID
     * @param quantity 扣减数量
     * @return 操作结果
     */
    public OperationResult deductStock(Long stockId, int quantity) {
        return deductStock(stockId, quantity, lockWaitTimeout);
    }
    
    /**
     * 扣减库存 - 带超时时间
     * @param stockId 库存ID
     * @param quantity 扣减数量
     * @param lockWaitTimeout 锁等待超时时间（毫秒）
     * @return 操作结果
     */
    public OperationResult deductStock(Long stockId, int quantity, long lockWaitTimeout) {
        Connection connection = null;
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;
        
        try {
            connection = DataSourceUtils.getConnection();
            
            // 设置事务隔离级别
            connection.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);
            connection.setAutoCommit(false);
            
            // 设置锁等待超时时间（秒）
            int timeoutSeconds = (int) Math.max(1, lockWaitTimeout / 1000);
            connection.createStatement().execute(
                "SET innodb_lock_wait_timeout = " + timeoutSeconds
            );
            
            // 执行带锁的查询
            String sql = "SELECT id, remaining, version FROM stock WHERE id = ? FOR UPDATE";
            preparedStatement = connection.prepareStatement(sql);
            preparedStatement.setLong(1, stockId);
            resultSet = preparedStatement.executeQuery();
            
            if (resultSet.next()) {
                int remaining = resultSet.getInt("remaining");
                int version = resultSet.getInt("version");
                
                if (remaining < quantity) {
                    connection.rollback();
                    return OperationResult.insufficientStock();
                }
                
                // 使用乐观锁（版本号）来防止并发更新
                // 这样可以避免锁持有时间过长
                String updateSql = "UPDATE stock SET remaining = remaining - ?, version = version + 1 " +
                                   "WHERE id = ? AND remaining >= ? AND version = ?";
                PreparedStatement updateStmt = connection.prepareStatement(updateSql);
                updateStmt.setInt(1, quantity);
                updateStmt.setLong(2, stockId);
                updateStmt.setInt(3, quantity);
                updateStmt.setInt(4, version);
                
                int affectedRows = updateStmt.executeUpdate();
                
                if (affectedRows == 0) {
                    // 更新失败，可能是并发冲突
                    connection.rollback();
                    return OperationResult.insufficientStock();
                }
                
                connection.commit();
                return OperationResult.success();
            }
            
            connection.rollback();
            return OperationResult.recordNotFound();
            
        } catch (SQLException e) {
            if (e.getErrorCode() == 1205) { // ER_LOCK_WAIT_TIMEOUT
                if (connection != null) {
                    try { connection.rollback(); } catch (SQLException ex) { /* ignore */ }
                }
                return OperationResult.lockTimeout();
            }
            
            if (connection != null) {
                try { connection.rollback(); } catch (SQLException ex) { /* ignore */ }
            }
            
            throw new RuntimeException("扣减库存失败", e);
        } finally {
            closeResources(resultSet, preparedStatement, connection);
        }
    }
    
    /**
     * 扣减多个库存 - 使用锁顺序策略
     * @param stockMap 库存ID到扣减数量的映射
     * @return 操作结果
     */
    public Map<Long, OperationResult> deductMultipleStocks(Map<Long, Integer> stockMap) {
        if (stockMap == null || stockMap.isEmpty()) {
            return Collections.emptyMap();
        }
        
        // 锁顺序：按库存ID升序排列
        List<Long> sortedStockIds = new ArrayList<>(stockMap.keySet());
        Collections.sort(sortedStockIds);
        
        Map<Long, OperationResult> results = new HashMap<>();
        Connection connection = null;
        
        try {
            connection = DataSourceUtils.getConnection();
            connection.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);
            connection.setAutoCommit(false);
            
            // 按顺序获取所有锁
            for (Long stockId : sortedStockIds) {
                OperationResult result = deductStockInternal(stockId, stockMap.get(stockId), connection);
                if (!result.isSuccess()) {
                    connection.rollback();
                    // 填充失败结果
                    results.put(stockId, result);
                    return results;
                }
                results.put(stockId, OperationResult.success());
            }
            
            connection.commit();
            return results;
            
        } catch (Exception e) {
            if (connection != null) {
                try { connection.rollback(); } catch (SQLException ex) { /* ignore */ }
            }
            throw new RuntimeException("批量扣减库存失败", e);
        } finally {
            if (connection != null) {
                try { connection.close(); } catch (SQLException e) { /* ignore */ }
            }
        }
    }
    
    /**
     * 内部扣减方法 - 复用连接，不提交事务
     */
    private OperationResult deductStockInternal(Long stockId, int quantity, Connection connection) {
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;
        
        try {
            // 设置锁等待超时
            int timeoutSeconds = (int) Math.max(1, lockWaitTimeout / 1000);
            connection.createStatement().execute(
                "SET innodb_lock_wait_timeout = " + timeoutSeconds
            );
            
            // 带锁查询
            String sql = "SELECT id, remaining, version FROM stock WHERE id = ? FOR UPDATE";
            preparedStatement = connection.prepareStatement(sql);
            preparedStatement.setLong(1, stockId);
            resultSet = preparedStatement.executeQuery();
            
            if (resultSet.next()) {
                int remaining = resultSet.getInt("remaining");
                int version = resultSet.getInt("version");
                
                if (remaining < quantity) {
                    return OperationResult.insufficientStock();
                }
                
                // 更新库存
                String updateSql = "UPDATE stock SET remaining = remaining - ?, version = version + 1 " +
                                   "WHERE id = ? AND remaining >= ? AND version = ?";
                PreparedStatement updateStmt = connection.prepareStatement(updateSql);
                updateStmt.setInt(1, quantity);
                updateStmt.setLong(2, stockId);
                updateStmt.setInt(3, quantity);
                updateStmt.setInt(4, version);
                
                int affectedRows = updateStmt.executeUpdate();
                
                if (affectedRows == 0) {
                    return OperationResult.insufficientStock();
                }
                
                return OperationResult.success();
            }
            
            return OperationResult.recordNotFound();
            
        } catch (SQLException e) {
            if (e.getErrorCode() == 1205) {
                return OperationResult.lockTimeout();
            }
            throw new RuntimeException("扣减库存失败", e);
        } finally {
            closeResources(resultSet, preparedStatement, null);
        }
    }
    
    private void closeResources(ResultSet resultSet, PreparedStatement preparedStatement, Connection connection) {
        if (resultSet != null) {
            try { resultSet.close(); } catch (SQLException e) { /* ignore */ }
        }
        if (preparedStatement != null) {
            try { preparedStatement.close(); } catch (SQLException e) { /* ignore */ }
        }
    }
}

---

性能优化：悲观锁不是万能的

虽然悲观锁能很好地保证数据一致性，但在极高并发场景下，它的性能瓶颈依然很明显。因为所有的请求都被串行化了，只有一个请求能拿到锁并执行，其他的都在等待。

为了进一步提升性能，我们可以考虑以下优化策略：

1. 缩短锁持有时间

/**
 * 优化：使用 SELECT ... FOR UPDATE SKIP LOCKED
 * 跳过已被锁定的行，避免等待
 */
@Service
public class StockServiceSkipLocked {
    
    public OperationResult deductStock(Long stockId, int quantity) {
        Connection connection = null;
        try {
            connection = DataSourceUtils.getConnection();
            connection.setAutoCommit(false);
            
            // 使用 SKIP LOCKED 跳过已被锁定的行
            // 这样即使有其他事务持有锁，当前请求也不会等待
            String sql = "SELECT id, remaining FROM stock WHERE id = ? FOR UPDATE SKIP LOCKED";
            PreparedStatement preparedStatement = connection.prepareStatement(sql);
            preparedStatement.setLong(1, stockId);
            ResultSet resultSet = preparedStatement.executeQuery();
            
            if (resultSet.next()) {
                int remaining = resultSet.getInt("remaining");
                
                if (remaining < quantity) {
                    connection.rollback();
                    return OperationResult.insufficientStock();
                }
                
                int updateRows = connection.createStatement().executeUpdate(
                    "UPDATE stock SET remaining = remaining - ? WHERE id = ? AND remaining >= ?"
                );
                
                if (updateRows == 0) {
                    connection.rollback();
                    return OperationResult.insufficientStock();
                }
                
                connection.commit();
                return OperationResult.success();
            }
            
            // 行已被其他事务锁定，返回锁超时
            connection.rollback();
            return OperationResult.lockTimeout();
            
        } catch (Exception e) {
            if (connection != null) {
                try { connection.rollback(); } catch (SQLException ex) { /* ignore */ }
            }
            throw new RuntimeException("扣减库存失败", e);
        } finally {
            if (connection != null) {
                try { connection.close(); } catch (SQLException e) { /* ignore */ }
            }
        }
    }
}

2. 使用数据库行锁代替表锁

确保你的查询条件能命中索引，这样才能使用行锁而不是表锁：

-- 错误的查询：没有索引，会锁住整个表
SELECT * FROM stock WHERE category_id = 123 FOR UPDATE;

-- 正确的查询：命中主键索引，只锁住需要的行
SELECT * FROM stock WHERE id = 1 FOR UPDATE;

3. 结合乐观锁进一步提高并发能力

对于某些场景，乐观锁可能更适合：

/**
 * 乐观锁 + 悲观锁的混合策略
 * 先用乐观锁尝试，失败后再降级到悲观锁
 */
@Service
public class StockServiceHybrid {
    
    @Autowired
    private StockMapper stockMapper;
    
    /**
     * 先尝试乐观锁，失败后降级到悲观锁
     */
    public OperationResult deductStock(Long stockId, int quantity) {
        // 第一次尝试：乐观锁
        OperationResult optimisticResult = tryOptimisticLock(stockId, quantity);
        if (optimisticResult.isSuccess()) {
            return optimisticResult;
        }
        
        // 如果乐观锁失败，降级到悲观锁
        return tryPessimisticLock(stockId, quantity);
    }
    
    private OperationResult tryOptimisticLock(Long stockId, int quantity) {
        Stock stock = stockMapper.selectById(stockId);
        if (stock == null) {
            return OperationResult.recordNotFound();
        }
        
        if (stock.getRemaining() < quantity) {
            return OperationResult.insufficientStock();
        }
        
        // 使用 CAS 操作更新库存
        int affectedRows = stockMapper.deductStockWithVersion(
            stockId, quantity, stock.getVersion()
        );
        
        if (affectedRows > 0) {
            return OperationResult.success();
        }
        
        // CAS 失败，可能有并发冲突，返回 null 表示需要降级
        return OperationResult.insufficientStock();
    }
    
    private OperationResult tryPessimisticLock(Long stockId, int quantity) {
        // 使用悲观锁
        return pessimisticDeductStock(stockId, quantity);
    }
}

---

实战案例：解决超卖问题的完整流程

让我们回到最初的那个超卖事故，看看如何用我们学到的知识来解决它：

/**
 * 订单服务 - 解决超卖问题
 */
@Service
public class OrderService {
    
    @Autowired
    private StockServiceComplete stockService;
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private OrderItemMapper orderItemMapper;
    
    /**
     * 创建订单
     * @param userId 用户ID
     * @param items 订单商品列表
     * @return 订单ID
     */
    @Transactional
    public Long createOrder(Long userId, List<OrderItemDTO> items) {
        // 1. 校验商品和库存
        Map<Long, Integer> stockMap = new HashMap<>();
        for (OrderItemDTO item : items) {
            // 获取商品信息
            Product product = productMapper.selectById(item.getProductId());
            if (product == null) {
                throw new BusinessException("商品不存在");
            }
            
            // 检查库存
            OperationResult result = stockService.deductStock(
                product.getStockId(), 
                item.getQuantity()
            );
            
            if (!result.isSuccess()) {
                throw new BusinessException("库存不足: " + result.getMessage());
            }
            
            stockMap.put(product.getStockId(), item.getQuantity());
        }
        
        // 2. 创建订单
        Order order = new Order();
        order.setUserId(userId);
        order.setTotalAmount(calculateTotalAmount(items));
        order.setCreateTime(new Date());
        orderMapper.insert(order);
        
        // 3. 创建订单明细
        for (OrderItemDTO item : items) {
            OrderItem orderItem = new OrderItem();
            orderItem.setOrderId(order.getId());
            orderItem.setProductId(item.getProductId());
            orderItem.setQuantity(item.getQuantity());
            orderItemMapper.insert(orderItem);
        }
        
        return order.getId();
    }
}

---

总结：如何选择合适的方案

在实际开发中，没有银弹。悲观锁、超时机制和锁顺序策略各有优劣，需要根据具体的业务场景来选择：

场景	推荐方案	原因
高并发、低并发冲突	乐观锁	性能更好，无锁开销
高并发、高并发冲突	悲观锁 + 超时	保证一致性，避免无限等待
多资源并发操作	悲观锁 + 锁顺序	避免死锁
极低并发	乐观锁或悲观锁均可	性能差异不大
超高并发（秒杀）	缓存 + 异步扣减	减轻数据库压力

最后，记住一句话：数据一致性永远是第一位的，性能优化要在保证一致性的前提下进行。

希望这篇文章能帮你彻底理解悲观锁、超时机制和锁顺序策略，再也不怕并发场景下的超卖问题了。