1. 高并发库存管理的核心挑战

在电商秒杀、限时抢购等高并发场景中，库存管理一直是系统设计的难点。当每秒数万甚至数十万的请求同时涌向同一个商品时，如何确保库存扣减的准确性成为技术团队必须面对的挑战。

超卖和少卖是库存管理中最常见的两类问题：

• 超卖：实际销售数量超过库存数量，导致无法履约
• 少卖：实际销售数量少于库存数量，造成库存浪费

这两种问题都会给企业带来直接的经济损失和信誉风险。以某电商平台"双11"活动为例，一款限量1000件的商品，如果出现超卖200件，意味着需要承担200笔订单的赔偿；如果少卖200件，则造成20%的库存积压。

2. 传统库存管理方案的问题分析

2.1 数据库行锁方案的局限性

传统方案通常采用数据库行锁实现库存扣减：

BEGIN; SELECT stock FROM products WHERE id=123 FOR UPDATE; -- 检查库存 UPDATE products SET stock=stock-1 WHERE id=123; COMMIT;

这种方案存在三个致命缺陷：

1. 性能瓶颈：数据库行锁在高并发下成为系统瓶颈
2. 死锁风险：多个事务互相等待导致死锁
3. 扩展困难：难以应对突发流量

实测数据显示，单机MySQL在100并发时QPS约500，而Redis可达10万+，性能差距达200倍。

2.2 分布式环境下的新挑战

在分布式系统中，库存管理还面临：

1. 网络延迟导致的数据不一致
2. 节点故障引发的库存异常
3. 分布式事务的性能损耗

3. Redis+Lua的高并发库存方案

3.1 整体架构设计

我们采用分层架构解决高并发库存问题：

用户请求 → 接入层 → Redis库存预扣 → 消息队列 → 数据库最终扣减

核心组件：

• Redis：承担库存预扣的流量洪峰
• Lua脚本：保证原子性操作
• 消息队列：实现异步削峰
• 数据库：持久化存储

3.2 Redis库存预扣实现

使用Lua脚本实现原子化库存扣减：

-- KEYS[1]: 库存key -- ARGV[1]: 扣减数量 local stock = tonumber(redis.call('GET', KEYS[1])) if stock >= tonumber(ARGV[1]) then return redis.call('DECRBY', KEYS[1], ARGV[1]) else return -1 end

Java调用示例：

String script = "local stock = tonumber(redis.call('GET', KEYS[1]))\n" + "if stock >= tonumber(ARGV[1]) then\n" + " return redis.call('DECRBY', KEYS[1], ARGV[1])\n" + "else\n" + " return -1\n" + "end"; Long result = jedis.eval(script, Collections.singletonList("stock:123"), Collections.singletonList("1"));

3.3 分布式锁防重复提交

为防止用户重复提交订单，需要加分布式锁：

public boolean tryLock(String key, String value, int expireTime) { String result = jedis.set(key, value, "NX", "PX", expireTime); return "OK".equals(result); } public boolean unlock(String key, String value) { String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " + "return redis.call('del', KEYS[1]) " + "else return 0 end"; Object result = jedis.eval(script, Collections.singletonList(key), Collections.singletonList(value)); return Long.valueOf(1L).equals(result); }

4. 异常处理与数据一致性

4.1 超时订单处理方案

对于未支付订单，采用两种处理方式：

1. RocketMQ延迟消息方案：

// 发送延迟消息 Message message = new Message("ORDER_TIMEOUT_TOPIC", orderId.getBytes()); message.setDelayTimeLevel(16); // 30分钟延迟 producer.send(message); // 消费端处理 consumer.subscribe("ORDER_TIMEOUT_TOPIC", "*"); consumer.registerMessageListener((msgs, context) -> { for (MessageExt msg : msgs) { String orderId = new String(msg.getBody()); if(orderService.isUnpaid(orderId)) { orderService.cancelOrder(orderId); stockService.rollbackStock(orderId); } } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; });

2. 定时任务扫描方案：

@Scheduled(fixedRate = 60000) // 每分钟执行 public void checkTimeoutOrders() { List<Order> timeoutOrders = orderDao.findTimeoutOrders(30); for (Order order : timeoutOrders) { orderService.cancelOrder(order.getId()); stockService.rollbackStock(order.getProductId(), order.getQuantity()); } }

4.2 数据最终一致性保障

采用本地消息表实现分布式事务：

-- 订单表 CREATE TABLE orders ( id BIGINT PRIMARY KEY, status VARCHAR(20), product_id BIGINT, quantity INT ); -- 本地消息表 CREATE TABLE message ( id BIGINT PRIMARY KEY AUTO_INCREMENT, biz_id BIGINT, content TEXT, status VARCHAR(20), created_time DATETIME );

事务处理流程：

1. 开启事务
2. 插入订单记录
3. 插入消息记录
4. 提交事务
5. 异步发送消息到MQ
6. 库存服务消费消息处理库存

5. 性能优化实践

5.1 Redis集群优化

1. 合理设置内存淘汰策略：

maxmemory-policy allkeys-lru

2. 使用Pipeline批量操作：

Pipeline pipeline = jedis.pipelined(); for (int i = 0; i < 100; i++) { pipeline.incr("counter"); } pipeline.sync();

3. 热点数据分片：

// 对商品ID取模分片 int shard = productId.hashCode() % 16; String redisKey = "stock_" + shard + ":" + productId;

5.2 数据库优化

1. 库存表分库分表策略：

# ShardingSphere配置 spring: shardingsphere: datasource: names: ds0,ds1 sharding: tables: product_stock: actual-data-nodes: ds$->{0..1}.product_stock_$->{0..15} table-strategy: inline: sharding-column: product_id algorithm-expression: product_stock_$->{product_id % 16}

2. 索引优化：

ALTER TABLE product_stock ADD INDEX idx_product (product_id); ALTER TABLE orders ADD INDEX idx_product_status (product_id, status);

6. 监控与报警体系

6.1 关键指标监控

1. Redis监控指标：

• 内存使用率
• 连接数
• 命中率
• 慢查询

2. 数据库监控指标：

• QPS
• 活跃连接数
• 锁等待时间
• SQL执行时间

6.2 报警规则配置

示例Prometheus报警规则：

groups: - name: inventory-alert rules: - alert: RedisHighMemoryUsage expr: redis_memory_used_bytes / redis_memory_max_bytes > 0.8 for: 5m labels: severity: critical annotations: summary: "Redis内存使用率过高" description: "Redis实例 {{ $labels.instance }} 内存使用率已达 {{ $value }}" - alert: InventoryInconsistency expr: (redis_stock - mysql_stock) != 0 for: 10m labels: severity: warning annotations: summary: "库存不一致告警" description: "商品 {{ $labels.product_id }} Redis与MySQL库存不一致"

7. 压测与性能数据

7.1 压测环境配置

测试环境：

• 服务器：8核16G × 3台
• Redis：6.2 集群模式 3主3从
• MySQL：8.0 主从架构

7.2 压测结果对比

方案对比数据：

7.3 性能优化效果

优化前后关键指标对比：

1. 99线耗时从86ms降至15ms
2. 系统吞吐量提升50倍
3. 服务器资源消耗降低40%
4. 超卖率从0.5%降至0.001%

8. 典型问题排查指南

8.1 库存不一致问题

排查步骤：

1. 检查Redis与MySQL网络连接
2. 验证消息队列消费延迟
3. 核对分布式锁有效性
4. 检查定时任务执行日志

解决方案：

// 定期执行库存校对任务 public void syncStock() { Map<Long, Integer> redisStocks = getAllRedisStocks(); Map<Long, Integer> mysqlStocks = getAllMysqlStocks(); redisStocks.forEach((productId, redisStock) -> { int mysqlStock = mysqlStocks.getOrDefault(productId, 0); if(redisStock != mysqlStock) { log.warn("库存不一致 product:{}, redis:{}, mysql:{}", productId, redisStock, mysqlStock); // 自动修复逻辑 if(redisStock > 0) { updateMysqlStock(productId, redisStock); } } }); }

8.2 热点Key问题

解决方案：

1. 本地缓存+Redis多级缓存

// 使用Caffeine本地缓存 LoadingCache<Long, Integer> localCache = Caffeine.newBuilder() .maximumSize(10_000) .expireAfterWrite(1, TimeUnit.SECONDS) .build(productId -> getRedisStock(productId));

2. Redis集群分片

// 对热点商品进行特殊分片 String getStockKey(Long productId) { if(hotProducts.contains(productId)) { return "hot_stock:" + ThreadLocalRandom.current().nextInt(10); } return "normal_stock:" + productId; }

9. 方案演进与扩展

9.1 库存预热策略

提前将热点商品库存加载到Redis：

public void preheatInventory(List<Long> productIds) { List<ProductStock> stocks = stockDao.batchGet(productIds); Pipeline pipeline = jedis.pipelined(); stocks.forEach(stock -> { pipeline.set("stock:" + stock.getProductId(), String.valueOf(stock.getQuantity())); }); pipeline.sync(); }

9.2 动态库存调整

支持运营实时调整库存：

@PostMapping("/adjust") public Result adjustStock(@RequestBody AdjustRequest request) { // 获取分布式锁 String lockKey = "lock:adjust:" + request.getProductId(); try { if(redisLock.tryLock(lockKey, 30_000)) { // 执行Lua脚本保证原子性 String script = "local curr = redis.call('GET', KEYS[1])\n" + "if not curr then return -1 end\n" + "local new = tonumber(curr) + tonumber(ARGV[1])\n" + "if new < 0 then return -2 end\n" + "redis.call('SET', KEYS[1], new)\n" + "return new"; Long result = jedis.eval(script, Collections.singletonList("stock:"+request.getProductId()), Collections.singletonList(String.valueOf(request.getDelta()))); if(result >= 0) { // 异步更新数据库 mqProducer.send(new StockAdjustMessage(request.getProductId(), result)); return Result.success(result); } return Result.fail(result == -1 ? "商品不存在" : "库存不足"); } return Result.fail("系统繁忙"); } finally { redisLock.unlock(lockKey); } }

10. 不同业务场景的适配

10.1 秒杀场景优化

特殊处理策略：

1. 库存分段：将1000件库存拆分为10个100件的库存段
2. 请求过滤：前置风控系统拦截异常请求
3. 页面静态化：减少服务器压力

10.2 预售模式处理

预售库存管理要点：

1. 独立库存池
2. 定金抵扣逻辑
3. 尾款支付超时处理

代码示例：

public boolean reserveInventory(Long productId, int quantity) { // 预售库存单独管理 String key = "pre_sale_stock:" + productId; Long remain = jedis.decrBy(key, quantity); if(remain >= 0) { // 生成预售订单 return true; } else { // 回滚库存 jedis.incrBy(key, quantity); return false; } }

在实际项目落地过程中，我们发现这套方案能够稳定支撑10万级QPS的秒杀场景，将超卖率控制在0.001%以下。关键在于Redis的合理使用、分布式锁的正确实现以及完善的异常处理机制。对于特别极端的场景，还可以考虑引入库存预扣+支付后实际扣减的两阶段方案，但这会增加系统复杂度，需要根据业务需求权衡。