在电商项目中，或者在面试中，绝大部分面试者都会被问到redis相关的问题，那么Redis 如何实现库存扣减操作和防止被超卖？

Redis 如何实现库存扣减操作和防止被超卖

• • • 解决方案：
    • 分析
    • • 基于数据库单库存
      • 基于数据库多库存
    • 基于redis
    • • 基于redis实现扣减库存的具体实现
      • 初始化库存回调函数(IStockCallback )
      • 扣减库存服务（StockService）
      • 调用

【#|请问Redis 如何实现库存扣减操作和防止被超卖（）】

文章图片

其实在日常开发中有很多地方都有类似扣减库存的操作，比如电商系统中的商品库存，抽奖系统中的奖品库存等。
解决方案：

1. 使用mysql数据库，使用一个字段来存储库存，每次扣减库存去更新这个字段。
2. 还是使用数据库，但是将库存分层多份存到多条记录里面，扣减库存的时候路由一下，这样子增大了并发量，但是还是避免不了大量的去访问数据库来更新库存。
3. 将库存放到redis使用redis的incrby特性来扣减库存。

分析
在上面的第一种和第二种方式都是基于数据来扣减库存。
基于数据库单库存 第一种方式在所有请求都会在这里等待锁，获取锁有去扣减库存。
在并发量不高的情况下可以使用，但是一旦并发量大了就会有大量请求阻塞在这里，导致请求超时，进而整个系统雪崩；而且会频繁的去访问数据库，大量占用数据库资源，所以在并发高的情况下这种方式不适用。
基于数据库多库存 第二种方式其实是第一种方式的优化版本，在一定程度上提高了并发量，但是在还是会大量的对数据库做更新操作大量占用数据库资源。
基于数据库来实现扣减库存还存在的一些问题：

1. 用数据库扣减库存的方式，扣减库存的操作必须在一条语句中执行，不能先select在update，这样在并发下会出现超扣的情况。如：

update number set x=x-1 where x > 0

2. MySQL自身对于高并发的处理性能就会出现问题，一般来说，MySQL的处理性能会随着并发thread上升而上升，但是到了一定的并发度之后会出现明显的拐点，之后一路下降，最终甚至会比单thread的性能还要差。
3. 当减库存和高并发碰到一起的时候，由于操作的库存数目在同一行，就会出现争抢InnoDB行锁的问题，导致出现互相等待甚至死锁，从而大大降低MySQL的处理性能，最终导致前端页面出现超时异常。

基于redis
针对上述问题的问题我们就有了第三种方案：
将库存放到缓存，利用redis的incrby特性来扣减库存，解决了超扣和性能问题。
但是一旦缓存丢失需要考虑恢复方案。比如抽奖系统扣奖品库存的时候，初始库存=总的库存数-已经发放的奖励数，但是如果是异步发奖，需要等到MQ消息消费完了才能重启redis初始化库存，否则也存在库存不一致的问题。
基于redis实现扣减库存的具体实现

1. 我们使用redis的lua脚本来实现扣减库存
2. 由于是分布式环境下所以还需要一个分布式锁来控制只能有一个服务去初始化库存
3. 需要提供一个回调函数，在初始化库存的时候去调用这个函数获取初始化库存

初始化库存回调函数(IStockCallback )

/** * 获取库存回调 * @author yuhao.wang */ public interface IStockCallback { /** * 获取库存 * @return */ int getStock(); }

扣减库存服务（StockService）

/** * 扣库存 * * @author yuhao.wang */ @Service public class StockService { Logger logger = LoggerFactory.getLogger(StockService.class); /** * 不限库存 */ public static final long UNINITIALIZED_STOCK = -3L; /** * Redis 客户端 */ @Autowired private RedisTemplate, Object> redisTemplate; /** * 执行扣库存的脚本 */ public static final String STOCK_LUA; static { /** * * @desc 扣减库存Lua脚本 * 库存（stock）-1：表示不限库存 * 库存（stock）0：表示没有库存 * 库存（stock）大于0：表示剩余库存 * * @params 库存key * @return *-3:库存未初始化 *-2:库存不足 *-1:不限库存 *大于等于0:剩余库存（扣减之后剩余的库存） *redis缓存的库存(value)是-1表示不限库存，直接返回1 */ 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; }/** * 加库存(还原库存) * * @param key库存key * @param num库存数量 * @return */ public long addStock(String key, int num) {return addStock(key, null, num); }/** * 加库存 * * @param key库存key * @param expire 过期时间（秒） * @param num库存数量 * @return */ public long addStock(String key, Long expire, int num) { boolean hasKey = redisTemplate.hasKey(key); // 判断key是否存在，存在就直接更新 if (hasKey) { return redisTemplate.opsForValue().increment(key, num); }Assert.notNull(expire,"初始化库存失败，库存过期时间不能为null"); RedisLock redisLock = new RedisLock(redisTemplate, key); try { if (redisLock.tryLock()) { // 获取到锁后再次判断一下是否有key 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; }/** * 获取库存 * * @param key 库存key * @return -1:不限库存; 大于等于0:剩余库存 */ public int getStock(String key) { Integer stock = (Integer) redisTemplate.opsForValue().get(key); return stock == null ? -1 : stock; }/** * 扣库存 * * @param key 库存key * @param num 扣减库存数量 * @return 扣减之后剩余的库存【-3:库存未初始化; -2:库存不足; -1:不限库存; 大于等于0:扣减库存之后的剩余库存】 */ private Long stock(String key, int num) { // 脚本里的KEYS参数 List> keys = new ArrayList<>(); keys.add(key); // 脚本里的ARGV参数 List> args = new ArrayList<>(); args.add(Integer.toString(num)); long result = redisTemplate.execute(new RedisCallback() { @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; }}

调用

/** * @author yuhao.wang */ @RestController public class StockController {@Autowired private StockService stockService; @RequestMapping(value = "https://www.it610.com/article/stock", produces = MediaType.APPLICATION_JSON_UTF8_VALUE) public Object stock() { // 商品ID long commodityId = 1; // 库存ID String redisKey = "redis_key:stock:" + commodityId; long stock = stockService.stock(redisKey, 60 * 60, 2, () -> initStock(commodityId)); return stock >= 0; }/** * 获取初始的库存 * * @return */ private int initStock(long commodityId) { // TODO 这里做一些初始化库存的操作 return 1000; }@RequestMapping(value = "https://www.it610.com/article/getStock", produces = MediaType.APPLICATION_JSON_UTF8_VALUE) public Object getStock() { // 商品ID long commodityId = 1; // 库存ID String redisKey = "redis_key:stock:" + commodityId; return stockService.getStock(redisKey); }@RequestMapping(value = "https://www.it610.com/article/addStock", produces = MediaType.APPLICATION_JSON_UTF8_VALUE) public Object addStock() { // 商品ID long commodityId = 2; // 库存ID String redisKey = "redis_key:stock:" + commodityId; return stockService.addStock(redisKey, 2); } }

推荐阅读

• 

  电商模式 mv是电商的什么，电商平台
• 

  蓝光驱蚊灯可以和人在一起吗
• 

  兼职跨境电商怎么样 跨境电商可以做什么兼职，哪个跨境电商平台适合兼职做
• 

  香菇酿肉的方法
• 

  方牌玉件的佩戴方法 方形玉牌吊坠怎么佩戴
• 

  纳兰词全集及赏析 温室的花
• 

  cdrx8安装提示已安装另一个版本解决方法
• 

  win10自带硬盘安装系统怎么安装，win10自带硬盘安装系统怎么安装不了
• 

  微信没声音怎么回事 没有声音解决办法
• 

  智齿|智齿发炎，疼痛得厉害还不能拔牙，有什么应急措施？
• 

  人工智能论文，人工智能的发展前景及其应用的论文怎么写
• 

  北太行在什么地方 北太行在哪里
• 

  博世壁挂炉德国还土耳其原因？
• 

  2023款一汽大众辉腾 一汽大众辉腾字母
• 

  3点到5点醒是癌症征兆有科学依据吗
• 

  樟脑丸可以放床下吗
• 

  干燥综合征如何治疗?
• 

  建设自由贸易区的意义 12个自贸试验区共有哪些
• 

  儿童可以使用免洗洗手液吗
• 

  登鹳雀楼古诗看到了哪四种景物？

• #|Redis 如何实现库存扣减操作和防止被超卖（（荣耀典藏版））
• 后端|面渣逆袭（MySQL六十六问，两万字+五十图详解！有点六！）
• 后端|面渣逆袭（Redis连环五十二问，三万字+八十图详解。）
• 数据库|2020年MySQL数据库面试题总结（50道题含答案解析）
• mysql|mysql limit 使用_mysql limit 使用方法实例解析
• #|详解MySQL架构设计及SQL执行流程
• MySql|数据库InnoDB-MVCC-多版本并发控制
• MySql|Mysql事务详解-[数据库的隔离级别、脏读、不可重复读、幻读以及ACID性质与redo log与undo log]
• sql|OA项目之我的会议（查询）