来源：my.oschina.net/xiaolyuh/blog/1615639在日常开发中，有很多地方有类似扣库存的操作，比如电商系统中的商品库存，彩票中的奖品库存系统等。方案使用mysql数据库，使用一个字段存储库存，每次扣除库存时更新这个字段。数据库还是用的，但是库存是分层存储的，多条记录，在扣库存的时候进行路由。这样增加了并发量，但是还是避免不了大量访问数据库来更新库存。将库存放入redis，利用redis的incrby特性扣减库存。上面的第一种和第二种分析方法是根据数据来扣除库存。在第一种基于数据库单库存的方法中，所有的请求都会在这里等待锁，锁的获取要扣除库存。并发不高的时候可以用，但是一旦并发大了，大量的请求就会阻塞在这里，导致请求超时，然后整个系统就会雪崩；并且会频繁访问数据库，占用大量的数据库资源，所以在高并发的情况下不适用这种方法。第二种基于多个数据库库存的方法其实是第一种方法的优化版，一定程度上提高了并发量，但是仍然需要大量的数据库更新，占用大量的数据库资源。基于数据库的扣存还存在一些问题：在用数据库扣存的方式中，扣存操作必须在一条语句中执行，不能先更新select，这样会造成过度并发下会发生扣减。例如：updatenumbersetx=x-1wherex>0mysql本身就会存在高并发处理性能的问题。一般来说，MySQL的处理性能会随着并发线程的增加而增加，但是在并发达到一定程度后会有一个明显的拐点，然后一路下降，最终甚至比单线程的性能还差。当库存减少和高并发相遇时，由于库存操作数在同一行，会出现争夺InnoDB行锁的问题，导致相互等待甚至死锁，大大降低MySQL的处理性能，并最终导致前端页面出现超时异常。基于redis存在以上问题，我们有第三种解决方案，就是将库存放在缓存中，利用redis的incrby特性来扣除库存，解决了超额订阅和性能问题。但是一旦缓存丢失，就需要考虑恢复计划了。例如，彩票系统在扣除奖品库存时，初始库存=总库存-已发放奖品数量。但是，如果奖励是异步发放的，则需要等到MQ消息被消费完后，再重启redis来初始化库存。否则会出现库存不一致的情况。基于redis扣库存的具体实现我们使用redis的lua脚本来实现扣库存。由于是分布式环境，所以需要一个分布式锁，只控制一个服务来初始化库存。需要提供回调函数。初始化股票时，调用该函数获取初始化股票。初始化股票回调函数(IStockCallback)/***获取股票回调*@authoryuhao.wang*/publicinterfaceIStockCallback{/***获取股票*@return*/intgetStock();}推荐一个SpringBoot基础教程和实例：https://github.com/javastacks...扣除库存服务（StockService）/***扣除库存**@authoryuhao.wang*/@ServicepublicclassStockService{Loggerlogger=LoggerFactory.getLogger(StockService.class);/***无限库存*/publicstaticfinallongUNINITIALIZED_STOCK=-3L;/***Redis客户端*/@AutowiredprivateRedisTemplateredisTemplate;/***执行库存扣除脚本*/publicstaticfinalStringSTOCK_LUA;static{/****@desc扣除库存的Lua脚本*Stock（库存）-1：表示无限库存*stock（库存）0：表示没有库存*Stock（库存）大于0：表示剩余库存**@paramsstockkey*@return*-3:stock未初始化*-2:Insufficientinventory*-1:Unlimitedinventory*大于等于0:Remaininginventory(扣除后的剩余库存)*Redis缓存库存(value)为-1表示无限库存，直接返回1*/StringBuildersb=新的StringBuilder();sb.append("if(redis.call('exists',KEYS[1])==1)then");sb.append("localstock=tonumber(redis.call('get',KEYS[1]));");sb.append("localnum=tonumber(ARGV[1]);");sb.append("if(stock==-1)then");sb.append("返回-1;");sb.append("结束;");sb.append("if(stock>=num)then");sb.append("returnredis.call('incrby',KEYS[1],0-num);");sb.append("结束;");sb.append("返回-2;");sb.append("结束;");sb.append("返回-3;");STOCK_LUA=sb.toString();}/***@paramkeyinventorykey*@paramexpire库存有效时间，单位秒*@paramnum扣除数量*@paramstockCallback初始化库存回调函数*@return-2：库存不足；-1：无限库存；大于等于0：扣除库存后剩余库存*/publiclongstock(Stringkey,longexpire,intnum,IStockCallbackstockCallback){longstock=stock(key,num);//初始化库存if(stock==UNINITIALIZED_STOCK){RedisLockredisLock=newRedisLock(redisTemplate,key);try{//获取锁if(redisLock.tryLock()){//双重验证，避免并发时重复回源数据库stock=stock(key,num);if(stock==UNINITIALIZED_STOCK){//获取初始股票finalinitStock=stockCallback.getStock();//设置存货到redisredisTemplate.opsForValue().set(key,initStock,expire,TimeUnit.SECONDS);//调整一次扣除库存的操作stock=stock(key,num);}}}catch(Exceptione){logger.error(e.getMessage(),e);}最后{redisLock.unlock();}}退货；}/***添加库存（恢复库存）**@paramkey库存键*@paramnum库存数量*@return*/publiclongaddStock(Stringkey,intnum){returnaddStock(key,null,num);}/***添加库存**@paramkey库存键*@paramexpire过期时间（秒）*@paramnum库存数量*@return*/publiclongaddStock(Stringkey,Longexpire,intnum){booleanhasKey=redisTemplate.hasKey(键);//判断key是否存在如果存在则直接更新if(hasKey){returnredisTemplate.opsForValue().increment(key,num);}Assert.notNull(expire,"初始化失败，库存过期时间不能为null");RedisLockredisLock=newRedisLock(redisTemplate,key);try{if(redisLock.tryLock()){//获取到达锁后，再次判断是否有keyhasKey=redisTemplate.hasKey(key);if(!hasKey){//初始化库存redisTemplate.opsForValue().set(key,num,expire,TimeUnit.SECONDS);}}}catch(Exceptione){logger.error(e.getMessage(),e);}最后{redisLock.unlock();}返回数；}/***获取库存**@paramkeyinventorykey*@return-1:无限库存；大于等于0：剩余库存*/publicintgetStock(Stringkey){Integerstock=(Integer)redisTemplate.opsForValue().get(key);退货==null？-1:股票;}/***扣除库存**@paramkeyinventorykey*@paramnum扣除库存数量*@return扣除后剩余库存[-3：库存未初始化；-2：库存不足；-1：无限库存；大于等于0：扣除库存后的剩余库存]*/privateLongstock(Stringkey,intnum){//脚本中的KEYS参数Listkeys=newArrayList<>();keys.add（键）；//脚本中的ARGV参数列表args=newArrayList<>();args.add(Integer.toString(num));longresult=redisTemplate.execute(newRedisCallback(){@OverridepublicLongdoInRedis(RedisConnectionconnection)throwsDataAccessException{ObjectnativeConnection=connection.getNativeConnection();//虽然集群模式和单机模式执行脚本在同样，它们没有共同的接口，所以它们只能单独执行//集群模式单机模式elseif(nativeConnectioninstanceofJedis){return(Long)((Jedis)nativeConnection).eval(STOCK_LUA,keys,args);}returnUNINITIALIZED_STOCK;}});返回结果；}}call/***@authoryuhao.wang*/@RestControllerpublicclassStockController{@AutowiredprivateStockServicestockService;@RequestMapping(value="stock",produces=MediaType.APPLICATION_JSON_UTF8_VALUE)publicObjectstock(){//产品IDlongcommodityId=1;//股票IDStringredisKey="redis_key:stock:"+commodityId;longstock=stockService.stock(redisKey,60*60,2,()->initStock(commodityId));退货>=0;}/***获取初始库存**@return*/privateintinitStock(longcommodityId){//TODO在这里做一些库存初始化操作return1000;}@RequestMapping(value="getStock",produces=MediaType.APPLICATION_JSON_UTF8_VALUE)publicObjectgetStock(){//商品IDlongcommodityId=1;//股票IDStringredisKey="redis_key:stock:"+commodityId;返回stockService.getStock(redisKey);}@RequestMapping(value="addStock",产生es=MediaType.APPLICATION_JSON_UTF8_VALUE)publicObjectaddStock(){//产品IDlongcommodityId=2;//股票IDStringredisKey="redis_key:stock:"+commodityId;返回stockService.addStock(redisKey,2);}}近期热点文章推荐：1.1000+Java面试题及答案（2022最新版）2.厉害了！Java协程来了。.3.SpringBoot2.x教程，太全面了！4.不要用爆破爆满画面，试试装饰者模式，这才是优雅的方式！！5.《Java开发手册（嵩山版）》最新发布，赶快下载吧！感觉不错，别忘了点赞+转发！
