本文曾经收录到Github仓库，该仓库蕴含计算机根底、Java外围知识点、多线程、JVM、常见框架、分布式、微服务、设计模式、架构等外围知识点，欢送star~
Github地址：https://github.com/Tyson0314/...
Gitee地址：https://gitee.com/tysondai/Ja...
电商当我的项目教训曾经十分广泛了，不论你是包装的还是实在的，起码要能讲清楚电商中常见的问题，比方库存的操作怎么避免商品被超卖
解决方案：
- 基于数据库单库存
- 基于数据库多库存
- 基于redis
基于redis实现扣减库存的具体实现
- 初始化库存回调函数（IStockCallback）
- 扣减库存服务（StockService）
- 调用
在日常开发中有很多中央都有相似扣减库存的操作，比方电商零碎中的商品库存，抽奖零碎中的奖品库存等。
解决方案
- 应用mysql数据库，应用一个字段来存储库存，每次扣减库存去更新这个字段。
- 还是应用数据库，然而将库存分层多份存到多条记录外面，扣减库存的时候路由一下，这样子增大了并发量，然而还是防止不了大量的去拜访数据库来更新库存。
- 将库存放到redis应用redis的incrby个性来扣减库存。
剖析
在下面的第一种和第二种形式都是基于数据来扣减库存。
基于数据库单库存
第一种形式在所有申请都会在这里期待锁，获取锁有去扣减库存。在并发量不高的状况下能够应用，然而一旦并发量大了就会有大量申请阻塞在这里，导致申请超时，进而整个零碎雪崩；而且会频繁的去拜访数据库，大量占用数据库资源，所以在并发高的状况下这种形式不实用。
基于数据库多库存
第二种形式其实是第一种形式的优化版本，在肯定水平上进步了并发量，然而在还是会大量的对数据库做更新操作大量占用数据库资源。
基于数据库来实现扣减库存还存在的一些问题：
- 用数据库扣减库存的形式，扣减库存的操作必须在一条语句中执行，不能先selec在update，这样在并发下会呈现超扣的状况。如：
update number set x=x-1 where x > 0- MySQL本身对于高并发的解决性能就会呈现问题，一般来说，MySQL的解决性能会随着并发thread回升而回升，然而到了肯定的并发度之后会呈现显著的拐点，之后一路降落，最终甚至会比单thread的性能还要差。
- 当减库存和高并发碰到一起的时候，因为操作的库存数目在同一行，就会呈现争抢InnoDB行锁的问题，导致呈现相互期待甚至死锁，从而大大降低MySQL的解决性能，最终导致前端页面呈现超时异样。
基于redis
针对上述问题的问题咱们就有了第三种计划，将库存放到缓存，利用redis的incrby个性来扣减库存，解决了超扣和性能问题。然而一旦缓存失落须要思考复原计划。比方抽奖零碎扣奖品库存的时候，初始库存=总的库存数-曾经发放的处分数，然而如果是异步发奖，须要等到MQ音讯生产完了能力重启redis初始化库存，否则也存在库存不统一的问题。
基于redis实现扣减库存的具体实现
- 咱们应用redis的lua脚本来实现扣减库存
- 因为是分布式环境下所以还须要一个分布式锁来管制只能有一个服务去初始化库存
- 须要提供一个回调函数，在初始化库存的时候去调用这个函数获取初始化库存
初始化库存回调函数(IStockCallback )
/** * 获取库存回调 */public interface IStockCallback { /** * 获取库存 * @return */ int getStock();}扣减库存服务（StockService）
/** * 扣库存 * */@Servicepublic class StockService { Logger logger = LoggerFactory.getLogger(StockService.class); /** * 不限库存 */ public static final long UNINITIALIZED_STOCK = -3L; /** * Redis 客户端 */ @Autowired private RedisTemplate<String, 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<String> keys = new ArrayList<>(); keys.add(key); // 脚本里的ARGV参数 List<String> args = new ArrayList<>(); args.add(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; }}调用
@RestControllerpublic class StockController { @Autowired private StockService stockService; @RequestMapping(value = "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 = "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 = "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); }}