电商防止超卖终极方案：让库存管理滴水不漏！[特殊字符]

标题：超卖？不存在的！五大方案让你高枕无忧
副标题：从数据库锁到Redis原子操作，防超卖全攻略

🎬 开篇：一个惨痛的案例

makefile

体验AI代码助手

代码解读

复制代码

双11零点，某电商平台： 00:00:00 - 秒杀活动开始！商品库存：100件 00:00:05 - 系统显示：已售罄！数据库库存：-23件 💀 实际订单：123个 00:01:00 - 客服电话被打爆：客户A：我买到了，为什么取消我订单？😠 客户B：明明还有库存，为什么买不了？😤 损失： - 赔偿金：10万+ - 用户流失：1000+ - 品牌损失：无法估量 CEO：这个锅，谁来背？💀

🤔 超卖问题的本质

想象10个人同时抢1件商品：

没有库存控制：10个人都看到"有货"，都下单成功（超卖9件！）
有库存控制：只有1个人抢到，其他9个人提示"已售罄"（完美！）

核心问题：并发场景下的库存扣减不是原子操作！

📚 知识地图

体验AI代码助手

代码解读

复制代码

防止超卖的五大方案 ├── 🔐 方案1：悲观锁（数据库锁） ├── 🎯 方案2：乐观锁（版本号） ├── ⚡ 方案3：Redis原子操作（推荐！） ├── 📦 方案4：消息队列削峰 └── 🎪 方案5：分布式锁+预扣库存

🔐 方案1：悲观锁 - "数据库行锁"

🌰 生活中的例子

银行取款机：

没有锁：两个人同时取钱，余额可能出错
有锁：第一个人操作时，第二个人只能等待

💻 技术实现

实现1：SELECT ... FOR UPDATE

java

体验AI代码助手

代码解读

复制代码

/** * 悲观锁方案：使用SELECT FOR UPDATE */ @Service public class PessimisticLockStockService { @Autowired private ProductMapper productMapper; @Autowired private OrderService orderService; /** * 购买商品（悲观锁） */ @Transactional(rollbackFor = Exception.class) public boolean buyProduct(Long productId, Integer quantity) { // 1. 🔒 查询并锁定商品（行锁） Product product = productMapper.selectForUpdate(productId); // 2. 检查库存 if (product.getStock() < quantity) { log.warn("库存不足：商品ID={}, 剩余库存={}, 购买数量={}", productId, product.getStock(), quantity); return false; } // 3. 扣减库存 int updated = productMapper.updateStock(productId, quantity); if (updated == 0) { log.error("扣减库存失败：商品ID={}", productId); return false; } // 4. 创建订单 orderService.createOrder(productId, quantity); log.info("购买成功：商品ID={}, 数量={}", productId, quantity); return true; } } /** * MyBatis Mapper */ @Mapper public interface ProductMapper { /** * 查询并锁定（关键！） */ @Select("SELECT * FROM product WHERE id = #{id} FOR UPDATE") Product selectForUpdate(@Param("id") Long id); /** * 扣减库存 */ @Update("UPDATE product SET stock = stock - #{quantity} " + "WHERE id = #{id}") int updateStock(@Param("id") Long id, @Param("quantity") Integer quantity); } /** * 原理说明： * * SELECT ... FOR UPDATE 会对查询到的行加排他锁（X锁） * * 事务A：SELECT ... FOR UPDATE ✅ 获取锁 * 事务B：SELECT ... FOR UPDATE ⏳ 等待... * 事务C：SELECT ... FOR UPDATE ⏳ 等待... * * 事务A：扣减库存 -> COMMIT 🔓 释放锁 * 事务B： ✅ 获取锁 * 事务C： ⏳ 等待... * * 优点： * ✅ 强一致性，绝不超卖 * ✅ 实现简单 * * 缺点： * ❌ 性能差（串行执行） * ❌ 容易死锁 * ❌ 不适合高并发场景 */

实现2：UPDATE直接加锁

java

体验AI代码助手

代码解读

复制代码

/** * 更简洁的悲观锁方案 */ @Service public class SimplePessimisticLockService { @Autowired private ProductMapper productMapper; @Transactional(rollbackFor = Exception.class) public boolean buyProduct(Long productId, Integer quantity) { // ⚡ 直接扣减库存（MySQL会自动加行锁） int updated = productMapper.decreaseStock(productId, quantity); if (updated == 0) { log.warn("扣减库存失败，可能库存不足"); return false; } // 创建订单 orderService.createOrder(productId, quantity); return true; } } @Mapper public interface ProductMapper { /** * 扣减库存（带库存检查） */ @Update("UPDATE product " + "SET stock = stock - #{quantity} " + "WHERE id = #{id} " + "AND stock >= #{quantity}") // ⚡ 关键：检查库存是否足够 int decreaseStock(@Param("id") Long id, @Param("quantity") Integer quantity); } /** * SQL执行流程： * * UPDATE product * SET stock = stock - 10 * WHERE id = 1 * AND stock >= 10 -- 如果不满足，updated = 0 * * 这个SQL会： * 1. 自动对id=1的行加X锁 * 2. 检查库存是否>=10 * 3. 如果满足，扣减库存；否则不更新 * 4. 返回影响行数（0或1） * * 完美解决超卖问题！✅ */

性能测试

java

体验AI代码助手

代码解读

复制代码

/** * 悲观锁性能测试 */ public class PessimisticLockTest { public static void main(String[] args) throws Exception { // 初始库存：100件 // 并发数：100个线程 // 每个线程购买：1件 int concurrency = 100; CountDownLatch latch = new CountDownLatch(concurrency); for (int i = 0; i < concurrency; i++) { new Thread(() -> { try { stockService.buyProduct(1L, 1); } finally { latch.countDown(); } }).start(); } latch.await(); /** * 测试结果： * - 最终库存：0件 ✅ 正确 * - 成功订单：100个 ✅ 正确 * - 平均耗时：1500ms 💀 太慢了！ * - TPS：67 💀 低！ */ } }

🎯 方案2：乐观锁 - "版本号机制"

🌰 生活中的例子

编辑Word文档：

悲观锁：我编辑时，锁定文档，你不能看
乐观锁：我们都可以编辑，提交时检查版本，冲突了就重试

💻 技术实现

sql

体验AI代码助手

代码解读

复制代码

-- 数据库表结构 CREATE TABLE product ( id BIGINT PRIMARY KEY, name VARCHAR(100), stock INT, version INT DEFAULT 0, -- ⚡ 版本号字段（关键！） INDEX idx_version (version) );

java

体验AI代码助手

代码解读

复制代码

/** * 乐观锁方案：使用版本号 */ @Service public class OptimisticLockStockService { @Autowired private ProductMapper productMapper; /** * 购买商品（乐观锁 + 重试） */ public boolean buyProduct(Long productId, Integer quantity) { int maxRetry = 3; // 最多重试3次 for (int i = 0; i < maxRetry; i++) { try { // 1. 查询商品（不加锁！） Product product = productMapper.selectById(productId); // 2. 检查库存 if (product.getStock() < quantity) { log.warn("库存不足"); return false; } // 3. ⚡ 乐观锁扣减库存（基于版本号） int updated = productMapper.decreaseStockWithVersion( productId, quantity, product.getVersion() // 传入旧版本号 ); if (updated == 0) { // 版本号不匹配，说明被其他线程修改了，重试！ log.warn("版本冲突，重试第{}次", i + 1); Thread.sleep(10); // 短暂休眠后重试 continue; } // 4. 创建订单 orderService.createOrder(productId, quantity); log.info("购买成功：商品ID={}, 数量={}", productId, quantity); return true; } catch (Exception e) { log.error("购买失败", e); if (i == maxRetry - 1) { return false; } } } log.error("重试{}次后仍然失败", maxRetry); return false; } } @Mapper public interface ProductMapper { /** * 基于版本号扣减库存（乐观锁） */ @Update("UPDATE product " + "SET stock = stock - #{quantity}, " + " version = version + 1 " + // 版本号+1 "WHERE id = #{id} " + "AND stock >= #{quantity} " + "AND version = #{version}") // ⚡ 检查版本号（关键！） int decreaseStockWithVersion(@Param("id") Long id, @Param("quantity") Integer quantity, @Param("version") Integer version); } /** * 原理说明： * * 初始状态：stock=10, version=0 * * 线程A查询：stock=10, version=0 * 线程B查询：stock=10, version=0 * * 线程A更新：UPDATE ... WHERE version=0 ✅ 成功，version变为1 * 线程B更新：UPDATE ... WHERE version=0 ❌ 失败，version已经是1了！ * * 线程B重试： * - 重新查询：stock=9, version=1 * - 再次更新：UPDATE ... WHERE version=1 ✅ 成功 * * 优点： * ✅ 性能比悲观锁好 * ✅ 不会死锁 * ✅ 并发度高 * * 缺点： * ⚠️ 高并发下重试次数多 * ⚠️ 可能导致活锁 * ⚠️ 需要业务层重试逻辑 */

改进版：ABA问题处理

java

体验AI代码助手

代码解读

复制代码

/** * 使用时间戳替代版本号（防止ABA问题） */ @Service public class TimestampOptimisticLockService { @Transactional(rollbackFor = Exception.class) public boolean buyProduct(Long productId, Integer quantity) { // 1. 查询商品 Product product = productMapper.selectById(productId); // 2. 检查库存 if (product.getStock() < quantity) { return false; } // 3. ⚡ 使用update_time作为乐观锁 int updated = productMapper.decreaseStockWithTimestamp( productId, quantity, product.getUpdateTime() // 传入旧的更新时间 ); if (updated == 0) { throw new OptimisticLockException("库存更新冲突"); } // 4. 创建订单 orderService.createOrder(productId, quantity); return true; } } @Mapper public interface ProductMapper { /** * 基于时间戳的乐观锁 */ @Update("UPDATE product " + "SET stock = stock - #{quantity}, " + " update_time = NOW() " + // 更新时间戳 "WHERE id = #{id} " + "AND stock >= #{quantity} " + "AND update_time = #{oldUpdateTime}") // 检查旧时间戳 int decreaseStockWithTimestamp(@Param("id") Long id, @Param("quantity") Integer quantity, @Param("oldUpdateTime") Date oldUpdateTime); }

⚡ 方案3：Redis原子操作 - "高性能之王"（推荐！）

🌰 生活中的例子

抢红包：

数据库方案：每个人都要去银行排队（慢）
Redis方案：微信服务器直接扣减，瞬间完成（快！）

💻 技术实现

实现1：Redis DECR原子操作

java

体验AI代码助手

代码解读

复制代码

/** * Redis方案：使用DECR原子操作 */ @Service public class RedisStockService { @Autowired private StringRedisTemplate redisTemplate; @Autowired private OrderService orderService; /** * 初始化库存到Redis */ public void initStock(Long productId, Integer stock) { String key = "product:stock:" + productId; redisTemplate.opsForValue().set(key, String.valueOf(stock)); } /** * 购买商品（Redis原子操作） */ public boolean buyProduct(Long productId, Integer quantity) { String stockKey = "product:stock:" + productId; // ⚡ Redis原子操作扣减库存 Long remainStock = redisTemplate.opsForValue() .decrement(stockKey, quantity); if (remainStock == null || remainStock < 0) { // 库存不足，回滚 if (remainStock != null && remainStock < 0) { redisTemplate.opsForValue().increment(stockKey, quantity); } log.warn("库存不足：商品ID={}, 剩余库存={}", productId, remainStock); return false; } try { // 异步创建订单（不阻塞扣减库存） orderService.createOrderAsync(productId, quantity); log.info("购买成功：商品ID={}, 数量={}, 剩余库存={}", productId, quantity, remainStock); return true; } catch (Exception e) { // 订单创建失败，回滚库存 redisTemplate.opsForValue().increment(stockKey, quantity); log.error("创建订单失败，已回滚库存", e); return false; } } /** * 查询剩余库存 */ public Integer getStock(Long productId) { String stockKey = "product:stock:" + productId; String stock = redisTemplate.opsForValue().get(stockKey); return stock != null ? Integer.parseInt(stock) : 0; } } /** * 性能对比： * * 数据库悲观锁：TPS = 67 💀 * 数据库乐观锁：TPS = 500 ✅ * Redis原子操作：TPS = 10000 🚀 快150倍！ */

实现2：Lua脚本（更强大）

java

体验AI代码助手

代码解读

复制代码

/** * Redis Lua脚本方案（原子性更强） */ @Service public class RedisLuaStockService { @Autowired private StringRedisTemplate redisTemplate; /** * Lua脚本：扣减库存 */ private static final String LUA_SCRIPT = "local stock_key = KEYS[1] " + "local quantity = tonumber(ARGV[1]) " + "local stock = tonumber(redis.call('get', stock_key) or '0') " + "" + "if stock >= quantity then " + " redis.call('decrby', stock_key, quantity) " + " return stock - quantity " + // 返回剩余库存 "else " + " return -1 " + // 库存不足 "end"; /** * 购买商品（Lua脚本） */ public boolean buyProduct(Long productId, Integer quantity) { String stockKey = "product:stock:" + productId; // ⚡ 执行Lua脚本（原子操作） Long remainStock = redisTemplate.execute( RedisScript.of(LUA_SCRIPT, Long.class), Collections.singletonList(stockKey), quantity.toString() ); if (remainStock == null || remainStock < 0) { log.warn("库存不足：商品ID={}", productId); return false; } // 异步创建订单 orderService.createOrderAsync(productId, quantity); log.info("购买成功：剩余库存={}", remainStock); return true; } } /** * Lua脚本的优势： * * ✅ 原子性：整个脚本作为一个原子操作 * ✅ 减少网络往返：一次请求完成多个操作 * ✅ 性能极高：TPS可达10万+ * ✅ 逻辑灵活：可以实现复杂的业务逻辑 */

实现3：Redis + 数据库最终一致性

java

体验AI代码助手

代码解读

复制代码

/** * 完整的Redis+数据库方案 * Redis：扣减库存（快） * DB：订单数据（持久化） * 异步：同步库存到DB（最终一致性） */ @Service public class RedisDbStockService { @Autowired private StringRedisTemplate redisTemplate; @Autowired private ProductMapper productMapper; @Autowired private OrderService orderService; @Autowired private RabbitTemplate rabbitTemplate; /** * 购买商品（完整流程） */ public boolean buyProduct(Long productId, Integer quantity) { // 1. ⚡ Redis扣减库存（快速响应） String stockKey = "product:stock:" + productId; Long remainStock = redisTemplate.opsForValue() .decrement(stockKey, quantity); if (remainStock == null || remainStock < 0) { // 库存不足 if (remainStock != null && remainStock < 0) { redisTemplate.opsForValue().increment(stockKey, quantity); } return false; } // 2. 发送MQ消息（异步处理订单） OrderMessage message = OrderMessage.builder() .productId(productId) .quantity(quantity) .userId(getCurrentUserId()) .build(); rabbitTemplate.convertAndSend( "order.exchange", "order.create", message ); log.info("扣减Redis库存成功，已发送MQ消息"); return true; } /** * MQ消费者：创建订单并同步库存到DB */ @RabbitListener(queues = "order.create.queue") public void handleOrderCreate(OrderMessage message) { try { // 1. 创建订单 Order order = orderService.createOrder( message.getProductId(), message.getQuantity(), message.getUserId() ); // 2. 扣减数据库库存（最终一致性） int updated = productMapper.decreaseStock( message.getProductId(), message.getQuantity() ); if (updated == 0) { // DB库存不足（理论上不应该出现） log.error("DB库存不足，需要回滚"); // 回滚Redis库存 String stockKey = "product:stock:" + message.getProductId(); redisTemplate.opsForValue().increment( stockKey, message.getQuantity()); // 取消订单 orderService.cancelOrder(order.getId()); } log.info("订单创建完成，库存已同步到DB"); } catch (Exception e) { log.error("处理订单失败", e); // 重试或人工介入 } } /** * 定时任务：Redis库存同步到DB */ @Scheduled(cron = "0 */5 * * * ?") // 每5分钟 public void syncStockToDb() { // 扫描所有商品，同步库存 List<Product> products = productMapper.selectAll(); for (Product product : products) { String stockKey = "product:stock:" + product.getId(); String redisStock = redisTemplate.opsForValue().get(stockKey); if (redisStock != null) { int stock = Integer.parseInt(redisStock); // 对比DB库存，不一致则更新 if (stock != product.getStock()) { productMapper.updateStockDirectly(product.getId(), stock); log.info("同步库存：商品ID={}, Redis={}, DB={}->{}", product.getId(), stock, product.getStock(), stock); } } } } }

📦 方案4：消息队列削峰 - "流量控制"

🌰 生活中的例子

排队买奶茶：

没有队列：所有人挤在柜台前（混乱）
有队列：排队等候，一个个处理（有序）

💻 技术实现

java

体验AI代码助手

代码解读

复制代码

/** * MQ削峰方案 * * 流程： * 1. 用户下单 -> 发送MQ消息 * 2. MQ消费者 -> 串行处理订单 * 3. 扣减库存 -> 创建订单 */ @Service public class MqStockService { @Autowired private RabbitTemplate rabbitTemplate; /** * 提交购买请求（快速返回） */ public String submitBuyRequest(Long productId, Integer quantity) { // 生成订单ID String orderId = generateOrderId(); // 发送MQ消息 OrderMessage message = OrderMessage.builder() .orderId(orderId) .productId(productId) .quantity(quantity) .userId(getCurrentUserId()) .createTime(System.currentTimeMillis()) .build(); rabbitTemplate.convertAndSend( "order.exchange", "order.submit", message ); log.info("订单提交成功：orderId={}", orderId); // ⚡ 立即返回订单ID（用户可以轮询查询结果） return orderId; } /** * MQ消费者：处理订单（串行） */ @RabbitListener(queues = "order.submit.queue", concurrency = "1") // 单线程消费 public void handleOrderSubmit(OrderMessage message) { try { // 1. 检查库存 Product product = productMapper.selectById(message.getProductId()); if (product.getStock() < message.getQuantity()) { // 库存不足，更新订单状态 orderService.updateOrderStatus( message.getOrderId(), OrderStatus.STOCK_INSUFFICIENT ); // 发送通知 notifyService.notifyStockInsufficient(message); return; } // 2. 扣减库存（串行处理，不会超卖！） int updated = productMapper.decreaseStock( message.getProductId(), message.getQuantity() ); if (updated == 0) { // 扣减失败 orderService.updateOrderStatus( message.getOrderId(), OrderStatus.FAILED ); return; } // 3. 创建订单 orderService.createOrder(message); // 4. 更新订单状态 orderService.updateOrderStatus( message.getOrderId(), OrderStatus.SUCCESS ); // 5. 发送成功通知 notifyService.notifyOrderSuccess(message); log.info("订单处理成功：orderId={}", message.getOrderId()); } catch (Exception e) { log.error("订单处理失败", e); // 更新订单状态 orderService.updateOrderStatus( message.getOrderId(), OrderStatus.FAILED ); } } /** * 查询订单状态 */ public OrderStatus queryOrderStatus(String orderId) { return orderService.getOrderStatus(orderId); } } /** * 优点： * ✅ 削峰填谷：控制处理速度，保护系统 * ✅ 异步处理：用户体验好 * ✅ 解耦：订单和库存系统解耦 * * 缺点： * ⚠️ 延迟：用户需要等待处理结果 * ⚠️ 复杂度：需要状态查询接口 * ⚠️ 消息可能丢失：需要持久化 */

🎪 方案5：分布式锁 + 预扣库存

💻 技术实现

java

体验AI代码助手

代码解读

复制代码

/** * 分布式锁 + 预扣库存方案 * * 适用场景：秒杀活动 * * 流程： * 1. 活动开始前：预扣库存到Redis * 2. 用户抢购：扣减Redis库存（加分布式锁） * 3. 支付成功：扣减真实库存 * 4. 支付超时：回滚Redis库存 */ @Service public class DistributedLockStockService { @Autowired private RedissonClient redissonClient; @Autowired private StringRedisTemplate redisTemplate; @Autowired private ProductMapper productMapper; /** * 预扣库存（活动开始前） */ public void preOccupyStock(Long productId, Integer stock) { // 1. 将库存加载到Redis String stockKey = "seckill:stock:" + productId; redisTemplate.opsForValue().set(stockKey, String.valueOf(stock)); // 2. 将真实库存记录下来 String realStockKey = "seckill:real_stock:" + productId; redisTemplate.opsForValue().set(realStockKey, String.valueOf(stock)); log.info("预扣库存完成：商品ID={}, 库存={}", productId, stock); } /** * 抢购商品（分布式锁） */ public String rushToBuy(Long productId, Integer quantity) { String lockKey = "seckill:lock:" + productId; RLock lock = redissonClient.getLock(lockKey); try { // ⚡ 获取分布式锁（最多等待3秒，锁自动释放10秒） boolean isLocked = lock.tryLock(3, 10, TimeUnit.SECONDS); if (!isLocked) { log.warn("获取锁失败：商品ID={}", productId); return null; } // 1. 检查Redis库存 String stockKey = "seckill:stock:" + productId; Long remainStock = redisTemplate.opsForValue() .decrement(stockKey, quantity); if (remainStock == null || remainStock < 0) { // 库存不足，回滚 if (remainStock != null && remainStock < 0) { redisTemplate.opsForValue().increment(stockKey, quantity); } log.warn("库存不足"); return null; } // 2. 创建预订单（待支付） String orderId = orderService.createPreOrder(productId, quantity); // 3. 设置订单超时（15分钟未支付自动取消） String orderKey = "seckill:order:" + orderId; redisTemplate.opsForValue().set( orderKey, orderId, Duration.ofMinutes(15) ); log.info("抢购成功：orderId={}, 剩余库存={}", orderId, remainStock); return orderId; } catch (InterruptedException e) { log.error("获取锁被中断", e); return null; } finally { if (lock.isHeldByCurrentThread()) { lock.unlock(); } } } /** * 支付成功：扣减真实库存 */ @Transactional(rollbackFor = Exception.class) public boolean paySuccess(String orderId) { Order order = orderService.getById(orderId); // 1. 扣减数据库真实库存 int updated = productMapper.decreaseStock( order.getProductId(), order.getQuantity() ); if (updated == 0) { log.error("扣减真实库存失败"); return false; } // 2. 更新订单状态 orderService.updateOrderStatus(orderId, OrderStatus.PAID); // 3. 删除订单超时key String orderKey = "seckill:order:" + orderId; redisTemplate.delete(orderKey); log.info("支付成功，真实库存已扣减"); return true; } /** * 支付超时：回滚Redis库存 */ @Scheduled(fixedRate = 60000) // 每分钟执行一次 public void handleExpiredOrders() { // 扫描过期订单 Set<String> keys = redisTemplate.keys("seckill:order:*"); if (keys == null || keys.isEmpty()) { return; } for (String key : keys) { String orderId = redisTemplate.opsForValue().get(key); if (orderId != null) { continue; // 订单未过期 } // 订单已过期，回滚库存 Order order = orderService.getById(extractOrderId(key)); if (order != null && order.getStatus() == OrderStatus.PENDING) { // 回滚Redis库存 String stockKey = "seckill:stock:" + order.getProductId(); redisTemplate.opsForValue().increment( stockKey, order.getQuantity()); // 取消订单 orderService.cancelOrder(order.getId()); log.info("订单超时，已回滚库存：orderId={}", order.getId()); } } } } /** * 优点： * ✅ 高性能：Redis操作快 * ✅ 支持预扣：适合秒杀场景 * ✅ 自动回滚：超时自动释放库存 * * 缺点： * ⚠️ 复杂度高：需要处理超时、回滚 * ⚠️ 数据一致性：Redis和DB可能不一致 */

📊 方案对比总结

方案	性能(TPS)	并发度	一致性	复杂度	适用场景
悲观锁（DB）	67 💀	低	强一致	简单	低并发
乐观锁（DB）	500 ✅	中	强一致	中等	中并发
Redis原子操作	10000 🚀	高	最终一致	中等	高并发（推荐）
MQ削峰	5000 ✅	高	最终一致	高	超高并发
分布式锁+预扣	8000 🚀	高	最终一致	高	秒杀活动

✅ 最佳实践

markdown

体验AI代码助手

代码解读

复制代码

生产环境推荐方案：Redis + MQ + 数据库架构设计： □ L1：Redis扣减库存（快速响应，TPS=1万+） □ L2：MQ异步处理订单（削峰填谷） □ L3：数据库持久化（最终一致性）关键要点： □ 库存预热：提前加载到Redis □ 原子操作：使用Lua脚本 □ 异步处理：MQ削峰 □ 降级策略：限流、熔断 □ 监控告警：库存、延迟、失败率极端场景处理： □ Redis宕机 -> 降级到数据库 □ MQ堵塞 -> 限流保护 □ 数据不一致 -> 定时对账 □ 恶意刷单 -> 黑名单、验证码