电商库存扣减方案详解：基于Redis Lua脚本防止超卖的设计与实践

在电商和各类秒杀活动中，库存扣减是核心且敏感的环节。设计不当，极易导致商品超卖或系统崩溃。那么，如何设计一个既能支撑高并发，又能严格防止超卖的库存系统呢？

常见的解决方案主要有以下几种思路：

• 使用 MySQL 数据库，用一个字段存储库存，每次扣减时更新该字段。
• 依然使用数据库，但将库存拆分到多条记录中（分桶），扣减时进行路由。这能在一定程度上提升并发能力，但仍需频繁访问数据库。
• 将库存预置到 Redis 中，利用其 incrby 命令的原子性进行扣减。

方案深度剖析

上述第一、二种方案都重度依赖数据库进行实时扣减。

基于数据库的单行库存

这种方式下，所有扣减请求都会竞争同一行数据的锁。在并发量不高时尚可应对，一旦遇到秒杀等高并发场景，大量请求将阻塞等待，导致接口超时，进而可能引发整个系统的雪崩。同时，频繁的数据库更新操作也会大量占用宝贵的数据库资源，因此在高并发下此方案不适用。

基于数据库的多行库存（分库分表/分桶）

这是对第一种方案的优化，通过将库存分散来提升并发处理的吞吐量。然而，它本质上仍然是对数据库进行大量更新操作，未能从根本上解决数据库 高并发 写入的压力问题。

此外，基于数据库实现扣减库存还存在一些固有难题：

• 原子性操作：扣减库存必须在一条 SQL 语句中完成（如 update stock set quantity = quantity - 1 where id = ? and quantity > 0），不能先查询 (select) 再更新 (update)，否则在并发下必然出现超扣（库存扣减为负数）。
• 性能瓶颈：MySQL 的性能在处理高并发线程时，到达一个临界点后不升反降，甚至可能低于单线程的性能。
• 锁竞争：所有操作集中在同一行数据时，会引发激烈的 InnoDB 行锁争用，导致大量线程等待甚至死锁，进一步降低数据库性能，最终使前端服务异常。

基于Redis的缓存方案

为了解决数据库方案在性能和一致性上的瓶颈，第三种方案应运而生：将库存加载到 Redis 缓存，利用其 INCRBY 命令的原子性执行扣减。这完美解决了超扣和性能两大核心问题。

当然，此方案也引入了新的考量点：缓存数据的可靠性。一旦 Redis 宕机或数据丢失，需要有可靠的恢复机制。例如，在抽奖系统中，初始化库存应为“总库存 - 已发放数”。如果发奖是异步的（通过 MQ），则必须等待所有 MQ 消息消费完毕才能安全地重新初始化 Redis 库存，否则也会出现数据不一致。

基于Redis的库存扣减核心实现

接下来，我们看一个具体的实现方案。该方案主要包含三个要点：

1. 使用 Redis Lua 脚本：确保查询库存和扣减库存的原子性。
2. 引入分布式锁：在分布式环境下，确保库存初始化操作仅由一个服务实例执行。
3. 回调函数设计：提供灵活的库存初始化数据源获取方式。

1. 库存初始化回调接口 (IStockCallback)

该接口定义了如何获取初始库存数据（例如从数据库加载）。

/**
 * 获取库存回调
 * @author yuhao.wang
 */
public interface IStockCallback {

 /**
  * 获取库存
  * @return
  */
 int getStock();
}

2. 核心库存服务 (StockService)

这是最核心的类，负责库存的扣减、增加和查询。它使用 Lua 脚本保障原子性，并用分布式锁保护初始化过程。

/**
 * 扣库存服务
 * 
 * @author yuhao.wang
 */
@Service
public class StockService {
    Logger logger = LoggerFactory.getLogger(StockService.class);

    /**
     * 不限库存标识
     */
    public static final long UNINITIALIZED_STOCK = -3L;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 扣减库存Lua脚本
     * 逻辑：
     * 1. 检查key是否存在。
     * 2. 若库存值为-1（不限库存），直接返回-1。
     * 3. 若库存充足（stock >= num），执行扣减并返回剩余库存。
     * 4. 若库存不足，返回-2。
     * 5. 若key不存在，返回-3（未初始化）。
     */
    public static final String STOCK_LUA;

    static {
        StringBuilder sb = new StringBuilder();
        sb.append("if (redis.call('exists', KEYS[1]) == 1) then");
        sb.append("    local stock = tonumber(redis.call('get', KEYS[1]));");
        sb.append("    local num = tonumber(ARGV[1]);");
        sb.append("    if (stock == -1) then");
        sb.append("        return -1;");
        sb.append("    end;");
        sb.append("    if (stock >= num) then");
        sb.append("        return redis.call('incrby', KEYS[1], 0 - num);");
        sb.append("    end;");
        sb.append("    return -2;");
        sb.append("end;");
        sb.append("return -3;");
        STOCK_LUA = sb.toString();
    }

    /**
     * 扣减库存
     * @param key          库存key
     * @param expire       库存有效时间，单位秒
     * @param num          扣减数量
     * @param stockCallback 初始化库存回调函数
     * @return -2:库存不足; -1:不限库存; >=0:扣减后剩余库存
     */
    public long stock(String key, long expire, int num, IStockCallback stockCallback) {
        long stock = stock(key, num);
        // 如果库存未初始化
        if (stock == UNINITIALIZED_STOCK) {
            RedisLock redisLock = new RedisLock(redisTemplate, key);
            try {
                // 获取分布式锁，防止并发重复初始化
                if (redisLock.tryLock()) {
                    // 双重检查，避免获取锁过程中库存已被其他线程初始化
                    stock = stock(key, num);
                    if (stock == UNINITIALIZED_STOCK) {
                        // 通过回调获取初始库存（例如从数据库读取）
                        final int initStock = stockCallback.getStock();
                        // 将库存设置到Redis，并设置过期时间
                        redisTemplate.opsForValue().set(key, initStock, expire, TimeUnit.SECONDS);
                        // 初始化后，再次尝试扣减
                        stock = stock(key, num);
                    }
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            } finally {
                redisLock.unlock();
            }
        }
        return stock;
    }

    /**
     * 增加库存（还原库存）
     */
    public long addStock(String key, int num) {
        return addStock(key, null, num);
    }

    public long addStock(String key, Long expire, int num) {
        boolean hasKey = redisTemplate.hasKey(key);
        if (hasKey) {
            return redisTemplate.opsForValue().increment(key, num);
        }
        // 如果key不存在，需要初始化，此时expire不能为空
        Assert.notNull(expire,"初始化库存失败，库存过期时间不能为null");
        RedisLock redisLock = new RedisLock(redisTemplate, key);
        try {
            if (redisLock.tryLock()) {
                hasKey = redisTemplate.hasKey(key);
                if (!hasKey) {
                    redisTemplate.opsForValue().set(key, num, expire, TimeUnit.SECONDS);
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            redisLock.unlock();
        }
        return num;
    }

    /**
     * 查询库存
     */
    public int getStock(String key) {
        Integer stock = (Integer) redisTemplate.opsForValue().get(key);
        return stock == null ? -1 : stock;
    }

    /**
     * 执行Lua脚本扣减库存（内部方法）
     */
    private Long stock(String key, int num) {
        List<String> keys = Collections.singletonList(key);
        List<String> args = Collections.singletonList(Integer.toString(num));

        long result = redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                Object nativeConnection = connection.getNativeConnection();
                // 适配集群和单机模式
                if (nativeConnection instanceof JedisCluster) {
                    return (Long) ((JedisCluster) nativeConnection).eval(STOCK_LUA, keys, args);
                }
                else if (nativeConnection instanceof Jedis) {
                    return (Long) ((Jedis) nativeConnection).eval(STOCK_LUA, keys, args);
                }
                return UNINITIALIZED_STOCK;
            }
        });
        return result;
    }
}

3. 控制器调用示例 (StockController)

最后，我们通过一个 Java Spring Boot 风格的控制器来看看如何调用上述服务。

/**
 * 库存控制器
 * @author yuhao.wang
 */
@RestController
public class StockController {

    @Autowired
    private StockService stockService;

    @RequestMapping(value = "stock", produces = MediaType.APPLICATION_JSON_VALUE)
    public Object stock() {
        long commodityId = 1;
        String redisKey = "stock:" + commodityId;
        // 尝试扣减2个库存，有效期1小时，初始化回调指向initStock方法
        long stock = stockService.stock(redisKey, 60 * 60, 2, () -> initStock(commodityId));
        return stock >= 0; // 返回true表示扣减成功，false表示失败
    }

    /**
     * 初始化库存回调函数的具体实现
     */
    private int initStock(long commodityId) {
        // 这里应实现从数据库或其他持久化存储中读取商品初始库存的逻辑
        // 示例返回1000
        return 1000;
    }

    @RequestMapping(value = "getStock", produces = MediaType.APPLICATION_JSON_VALUE)
    public Object getStock() {
        long commodityId = 1;
        String redisKey = "stock:" + commodityId;
        return stockService.getStock(redisKey);
    }

    @RequestMapping(value = "addStock", produces = MediaType.APPLICATION_JSON_VALUE)
    public Object addStock() {
        long commodityId = 2;
        String redisKey = "stock:" + commodityId;
        return stockService.addStock(redisKey, 2);
    }
}

总结

本文详细分析了电商库存扣减从数据库方案到 Redis 缓存方案的演进历程。基于 Redis Lua 脚本的原子操作，结合分布式锁保护初始化过程，是构建高并发、防超卖库存系统的有效实践。当然，在真正的生产环境中，还需要考虑缓存穿透、击穿、雪崩、数据持久化与恢复等更复杂的 分布式系统 问题。

希望这篇深入的技术解析能为你带来启发。欢迎在 云栈社区 交流讨论更多后端架构与高并发设计的相关话题。

原文作者：xiaolyuh
原文链接：my.oschina.net/xiaolyuh/blog/161563