集群环境下Redis 商品库存系统设计

环境

我们现在要做商品秒杀系统。功能很简单，就是库存删减。用户先下单减库存，之后再进行扣款。

实现

基本结构代码

那么我们先看下如何搭建好基本的代码。

业务代码主体

基本步骤就以下几点

删减库存
填写订单基本信息


public class SecKillBusinessService {// 库存 serviceprivate StockDataService stockService;// 订单 serviceprivate OrderService orderService;public Response order( String userId , String productId ){// 获取当前时间点LocalDateTime  time = LocalDateUtils.now();// 1. 删减库存this.stockService.reduceStock( userId , productId ,  1 );//2. 下单OrderEntity order = new OrderEntity();order.setId(xxxx);order.setUserId(userId);order.setTime( time );this.orderService.add( order );return Response.success();}
}

库存管理模块


public class StockService {// 库存底层数据private StockDataService dataService;public void reduceStock(String userId , String productId , Integer number ){// 获取剩余库存数量int surplusStock = this.dataService.getStock( productId );if( surplusStock == 0 || surplusStock-number < 0 ){throw new ResponseFailedExpection( "库存不足");}// 自减数量 , 当库存不足时扣减失败，当前失败码暂定为-1int surplusNumber = this.dataService.decrementStock( productId ,  number );if( surplusNumber < 0 ){throw new ResponseFailedExpection("库存不足");}}}

StockDataService 我们先通过查询Mysql来实现。


public class StockDataServiceRedisImpl implement StockmentDataService {public int getStock( String productId ){// SELECT * FROM t_a_product WHERE product_id = #{productId}}@Transactionpublic int decrmentStock( String productId , Integer number ){// 简单的乐观锁// UPDATE t_a_product SET stock-=#{number} WHERE product_id = #{productId} AND stock>=#{number} 	}
}

后续问题

秒杀的主要问题复杂代码集中在如何在高并发环境下扣减库存，库存不会出现库存数据计数错误，且更高效。

高并发

当数据量上来的时候，我们很快就会发现问题。当流量大的时候，数据库IO很快就会打满。然后查询慢，插入慢。最后Mysql挂掉，服务不可用。

主要的问题，就是数据库难以应付高并发。那么我们如何处理？
很简单，我们使用Redis来替代Mysql ，我们新建一个新的StockDataService来进行替换。
为了保证计数问题，我们无非要么用乐观锁要么用悲观锁要么二者都用。高并发情况下，我们不可能用悲观锁来让程序在同一时间只允许一个请求在运行。（因为会引发大规模排队）因此我们采用乐观锁

public class StocklDataServiceRedisImpl implement StockmentDataService {private RedisService redisService;private static final String GET_STOCK_KEY = "GET_STOCK";private String getStockRedisKey( String productId ){return GET_STOCK_KEY + productId;}/**redis之中的库存数在其他模块便填充，我们可以放在后台配置的时候，也可以通过定时任务在商品生效一个小时之前。*/public int getStock( String productId ){return redisService.get(this.getStockRedisKey(productId) , Integer.class);}public int decrmentStock( String productId , Integer number ){String redisKey  = this.getStockRedisKey(productId);int surplusNumber = this.redisSerivce.decrement(redisKey  ,number);// 如果减少的数量超过库存上限，那么归还库存if( surplusNumber <0 ){this.redisService.incrment(redisKey ,number);return -1;}return surplusNumber;}

我们简单的用redis做了一个减库存的相关功能，并且还简单做了一个乐观锁逻辑。来处理临界值时库存扣减超量问题。

临界值与乐观锁问题

在讨论当前情况之前，我们得先对临界值有一个简单的认识。就是一个商品的临界值时多少？
由于本人水平有限，我先简单的做个定义。 0.8 * 当前剩余库存数 = 当前所需的数量
简单的说，假设当前库存10000份，当前库存数已经只剩下了500，当前服务器内计算到的所需要的总数达到400甚至更多时，我们就需要，那么我们就到达了临界值状态。

那么现在我们回到问题，
虽然我们乐观锁能简单解决大部分问题，但是当库存来到临界值的时候，我们就会悲伤的发现。大量的请求会失效。这些请求即无用又会给redis造成极大的压力。

问题的本质是什么呢？是因为查询+查库存的这两步骤无法原子化，库存数量在删减库存的时候并不可靠。

我们就直接说Redis的解决方案。

Redis Lua 脚本

不认识的可以简单的这样认为，他会把不同的脚本原子化处理。也可以说Redis会自己将一连串的Lua用分布式锁锁住然后执行。只是用它来实现分布式事务锁不太容易出现性能问题。

-- 方式 2：Lua 脚本实现原子扣减
local stockKey = KEYS[1]
local number = KEYS[2]
local stock = tonumber(redis.call('GET', stockKey))if stock >= number  thenredis.call('DECR', stockKey)return stock - number 
elsereturn -1
end

我们可以直接更改 StocklDataServiceRedisImpl

public class StocklDataServiceRedisImpl implement StockmentDataService {private RedisService redisService;private static final String GET_STOCK_KEY = "GET_STOCK";private String getStockRedisKey( String productId ){return GET_STOCK_KEY + productId;}public int decrmentStock( String productId , Integer number ){// Lua 脚本String script = "xxx";// 通过 Lua 一次性扣减库存DefaultRedisScript<Integer> redisScript = new DefaultRedisScript<>(script, Integer.class);List<String> keys = Arrays.asList(this.getStockRedisKey(productId , new StringBuilder.append(number).toString()));return this.redisService.executeLua(redisScript, keys);}

并且由于Redis Lua 能保证原子性，甚至能更改 StockService 逻辑不需要对当前库存进行校验。仅处理一个Redis命令即可。
自然可能由于其他因素，是否如此凭个人好恶


public class StockService {// 库存底层数据private StockDataService dataService;public void reduceStock(String userId , String productId , Integer number ){// 自减数量 , 当库存不足时扣减失败，当前失败码暂定为-1int surplusNumber = this.dataService.decrementStock( productId ,  number );if( surplusNumber < 0 ){throw new ResponseFailedExpection("库存不足");}}}

完整代码总结

完善之后，当前代码为

SecKillBusinessService .java


public class SecKillBusinessService {// 库存 serviceprivate StockDataService stockService;// 订单 serviceprivate OrderService orderService;public Response order( String userId , String productId ){// 获取当前时间点LocalDateTime  time = LocalDateUtils.now();// 1. 删减库存this.stockService.reduceStock( userId , productId ,  1 );//2. 下单OrderEntity order = new OrderEntity();order.setId(xxxx);order.setUserId(userId);order.setTime( time );this.orderService.add( order );return Response.success();}
}

StockService .java


public class StockService {// 库存底层数据private StockDataService dataService;public void reduceStock(String userId , String productId , Integer number ){// 获取剩余库存数量int surplusStock = this.dataService.getStock( productId );if( surplusStock == 0 || surplusStock-number < 0 ){throw new ResponseFailedExpection( "库存不足");}// 自减数量 , 当库存不足时扣减失败，当前失败码暂定为-1int surplusNumber = this.dataService.decrementStock( productId ,  number );if( surplusNumber < 0 ){throw new ResponseFailedExpection("库存不足");}}}

StocklDataServiceRedisImpl .java

public class StocklDataServiceRedisImpl implement StockmentDataService {private RedisService redisService;private static final String GET_STOCK_KEY = "GET_STOCK";private String getStockRedisKey( String productId ){return GET_STOCK_KEY + productId;}/**redis之中的库存数在其他模块便填充，我们可以放在后台配置的时候，也可以通过定时任务在商品生效一个小时之前。*/public int getStock( String productId ){return redisService.get(this.getStockRedisKey(productId) , Integer.class);}public int decrmentStock( String productId , Integer number ){// Lua 脚本String script = "xxx";// 通过 Lua 一次性扣减库存DefaultRedisScript<Integer> redisScript = new DefaultRedisScript<>(script, Integer.class);List<String> keys = Arrays.asList(this.getStockRedisKey(productId , new StringBuilder.append(number).toString()));return this.redisService.executeLua(redisScript, keys);}

后话

我们可以想象一下，如果没有Redis Lua 功能，我们需要做什么？
为了减少乐观锁出现的大面积下单失败，我们只能依赖于悲观锁。
但是悲观锁严重影响性能不可取，因此我们只能折中。设置一个危险值，当库存大于危险值时使用乐观锁，低于危险值时采用悲观锁。
危险值应该大于接口请求数上限，且为了不让大量蜂拥而入的无用请求排队。我们需要登记每个请求，且当请求量大于库存数就直接拒绝服务。

这应该就是我们常说的，少即是多，以及磨刀不误砍柴工吧。