以下文章来自Java必修指南。日常开发中有很多类似库存扣减操作的地方，比如电商系统中的商品库存，彩票系统中的奖品库存等。方案使用mysql数据库，使用一个字段存储库存，每次扣除库存时更新这个字段。数据库还是用的，但是库存是分层存储的，多条记录，在扣库存的时候进行路由。这样增加了并发量，但是还是避免不了大量访问数据库来更新库存。将库存放入redis，利用redis的incrby特性扣减库存。上面的第一种和第二种分析方法是根据数据来扣除库存。在第一种基于数据库单库存的方法中，所有的请求都会在这里等待锁，锁的获取要扣除库存。并发不高的时候可以用，但是一旦并发大了，大量的请求就会阻塞在这里，导致请求超时，然后整个系统就会雪崩；并且会频繁访问数据库，占用大量的数据库资源，所以在高并发的情况下不适用这种方法。第二种基于多个数据库库存的方法其实是第一种方法的优化版，一定程度上增加了并发量，但是仍然需要大量的数据库更新占用大量的数据库资源_java训练。基于数据库的扣存还存在一些问题：在用数据库扣存的方式中，扣存操作必须在一条语句中执行，不能先更新select，这样会造成过度并发下会发生扣减。例如：updatenumbersetx=x-1wherex>0mysql本身就会存在高并发处理性能的问题。一般来说，MySQL的处理性能会随着并发线程的增加而增加，但是在并发达到一定程度后会有一个明显的拐点，然后一路下降，最终甚至比单线程的性能还差。当库存减少和高并发相遇时，由于库存操作数在同一行，会出现争夺InnoDB行锁的问题，导致相互等待甚至死锁，大大降低MySQL的处理性能，并最终导致前端页面出现超时异常。基于redis存在以上问题，我们有第三种解决方案，就是将库存放在缓存中，利用redis的incrby特性来扣除库存，解决了超额订阅和性能问题。但是一旦缓存丢失，就需要考虑恢复计划了。例如，彩票系统在扣除奖品库存时，初始库存=总库存-已发放奖品数量。但是，如果奖励是异步发放的，则需要等到MQ消息被消费完后，再重启redis来初始化库存。否则会出现库存不一致的情况。基于redis扣库存的具体实现我们使用redis的lua脚本来实现扣库存。由于是分布式环境，所以需要一个分布式锁，只控制一个服务来初始化库存。需要提供回调函数。初始化股票时，调用该函数获取初始化股票。初始化股票回调函数(IStockCallback)/**获取股票回调@authoryuhao.wang*/publicinterfaceIStockCallback{/**获取股票@return*/intgetStock();}扣除库存服务(StockService)/**扣除库存*@authoryuhao.wang*/@ServicepublicclassStockService{Loggerlogger=LoggerFactory.getLogger(StockService.class);/**无限库存*/publicstaticfinallongUNINITIALIZED_STOCK=-3L;/**Redisclient*/@AutowiredprivateRedisTemplateredisTemplate;/**执行扣库存脚本*/publicstaticfinalStringSTOCK_LUA;static{/***@desc扣库存Lua脚本stock(stock)-1:表示unlimitedstock库存(stock)0：表示没有库存stock(stock)大于0：表示剩余库存*@paramsstockkey@return-3：库存未初始化-2：库存不足-1：无限库存大于或等于0：剩余库存（扣除后剩余库存）redis缓存库存（值）为-1表示无限库存，直接return1*/StringBuildersb=newStringBuilder();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("return-1;");sb.append("end;");sb.append("if(stock>=num)then");sb.append("returnredis.call('incrby',KEYS[1],0-num);");sb.append("end;");sb.append("return-2;");sb.append("end;");sb.append("return-3;");STOCK_LUA=sb.toString();}/**@paramkeystockkey@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){//获取初始库存finalintinitStock=stockCallback.getStock();//将库存设置为redisredisTemplate.opsForValue().set(key,initStock,expire,TimeUnit.SECONDS);//调整一次扣除库存的操作stock=stock(key,num);}}}catch(Exceptione){logger.error(e.getMessage(),e);}finally{redisLock.unlock();}}returnstock;}/**添加库存（恢复库存）*@paramkeystockkey@paramnumstockquantity@return*/publiclongaddStock(Stringkey,intnum){returnaddStock(key,null,num);}/**添加股票*@paramkeystockkey@paramexpire到期时间(秒)@paramnum股票数量@return*/publiclongaddStock(Stringkey,Longexpire,intnum){booleanhasKey=redisTemplate.hasKey(key);//判断key是否存在，存在则直接更新if(hasKey){returnredisTemplate.opsForValue().increment(key,num);}Assert.notNull(expire,"初始化库存失败，库存过期时间不能为空");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);}finally{redisLock.unlock();}returnnum;}/**Getinventory*@paramkeyinventorykey@return-1:无限库存；大于等于0：剩余存货*/publicintgetStock(Stringkey){Integerstock=(Integer)redisTemplate.opsForValue().get(key);返回库存==null？-1:stock;}/**扣除库存*@paramkeystockkey@paramnum扣除库存数量@退还剩余库存【-3:库存未初始化;-2：库存不足；-1：无限库存；大于等于0：扣除库存后的剩余库存]*/privateLongstock(Stringkey,intnum){//脚本中的KEYS参数Listkeys=newArrayList<>();keys.add(key);//ARGV参数Listargs=newArrayList<>();args.add(Integer.toString(num));longresult=redisTemplate.execute(newRedisCallback(){@OverridepublicLongdoInRedis(RedisConnectionconnection)throwsDataAccessException{ObjectnativeConnection=connection.getNativeConnection();//集群模式和单机模式虽然执行脚本的方法相同，但是没有通用的接口，所以只能单独执行//集群模式if(nativeConnectioninstanceofJedisCluster){return(Long)((JedisCluster)nativeConnection).eval(STOCK_LUA,keys,args);}//单机模式elseif(nativeConnectioninstanceofJedis){return(Long)((Jedis)nativeConnection).eval(STOCK_LUA,keys,args);}returnUNINITIALIZED_STOCK;}});returnresult;}}call/**@authoryuhao.wang*/@RestControllerpublicclassStockController{@AutowiredprivateStockServicestockService;@RequestMapping(value="stock",produces=MediaType.APPLICATION_JSON_UTF8_VALUE)publicObjectstock(){//commodityIDlongcommodityId=1;//stockIDStringredisKey="redis_key:stock:"+commodityId;longstock=stockService.stock(redisKey,60*60,2,()->initStock(commodityId));returnstock>=0;}/**获取初始库存*@return*/privateintinitStock(longcommodityId){//TODO在这里做一些库存初始化操作return1000;}@RequestMapping(value="getStock",produces=MediaType.APPLICATION_JSON_UTF8_VALUE)publicObjectgetStock(){//CommodityIDlongcommodityId=1;//股票IDStringredisKey="redis_key:stock:"+commodityId;returnstockService.getStock(redisKey);}@RequestMapping(value="addStock",produces=MediaType.APPLICATION_JSON_UTF8_VALUE)publicObjectaddStock(){//商品IDlongcommodityId=2;//股票IDStringredisKey="redis_key:stock:"+商品编号；返回stockService.addStock(redisKey,2);}}
