distributed-redis-tool ⬆️v1.0.3
为此还对 github.com/crossoverJi… 进行了小小的升级。
因为加上该组件之后所有的请求都会经过 Redis，所以对 Redis 资源的使用也是要非常小心。
API 更新
修改之后的 API 如下：
@Configuration public class RedisLimitConfig { private Logger logger = LoggerFactory.getLogger(RedisLimitConfig.class); @Value("${redis.limit}") private int limit; @Autowired private JedisConnectionFactory jedisConnectionFactory; @Bean public RedisLimit build() { RedisLimit redisLimit = new RedisLimit.Builder(jedisConnectionFactory, RedisToolsConstant.SINGLE) .limit(limit) .build(); return redisLimit; } }
这里构建器改用了 JedisConnectionFactory，所以得配合 Spring 来一起使用。
并在初始化时显示传入 Redis 是以集群方式部署还是单机（强烈建议集群，限流之后对 Redis 还是有一定的压力）。
限流实现
既然 API 更新了，实现自然也要修改：
/** * limit traffic * @return if true */ public boolean limit() { //get connection Object connection = getConnection(); Object result = limitRequest(connection); if (FAIL_CODE != (Long) result) { return true; } else { return false; } } private Object limitRequest(Object connection) { Object result = null; String key = String.valueOf(System.currentTimeMillis() / 1000); if (connection instanceof Jedis){ result = ((Jedis)connection).eval(script, Collections.singletonList(key), Collections.singletonList(String.valueOf(limit))); ((Jedis) connection).close(); }else { result = ((JedisCluster) connection).eval(script, Collections.singletonList(key), Collections.singletonList(String.valueOf(limit))); try { ((JedisCluster) connection).close(); } catch (IOException e) { logger.error("IOException",e); } } return result; } private Object getConnection() { Object connection ; if (type == RedisToolsConstant.SINGLE){ RedisConnection redisConnection = jedisConnectionFactory.getConnection(); connection = redisConnection.getNativeConnection(); }else { RedisClusterConnection clusterConnection = jedisConnectionFactory.getClusterConnection(); connection = clusterConnection.getNativeConnection() ; } return connection; }
如果是原生的 Spring 应用得采用 @SpringControllerLimit(errorCode = 200) 注解。
实际使用如下：
web 端：
/** * 乐观锁更新库存 限流 * @param sid * @return */ @SpringControllerLimit(errorCode = 200) @RequestMapping("/createOptimisticLimitOrder/{sid}") @ResponseBody public String createOptimisticLimitOrder(@PathVariable int sid) { logger.info("sid=[{}]", sid); int id = 0; try { id = orderService.createOptimisticOrder(sid); } catch (Exception e) { logger.error("Exception",e); } return String.valueOf(id); } 复制代码
Service 端就没什么更新了，依然是采用的乐观锁更新数据库。
再压测看下效果 /createOptimisticLimitOrderByRedis/1：
首先是看结果没有问题，再看数据库连接以及并发请求数都有明显的下降。
乐观锁更新 + 分布式限流 + Redis 缓存
其实仔细观察 Druid 监控数据发现这个 SQL 被多次查询：
其实这是实时查询库存的 SQL，主要是为了在每次下单之前判断是否还有库存。
这也是个优化点。
这种数据我们完全可以放在内存中，效率比在数据库要高很多。
由于我们的应用是分布式的，所以堆内缓存显然不合适，Redis 就非常适合。
这次主要改造的是 Service 层：
- 每次查询库存时走 Redis。
- 扣库存时更新 Redis。
- 需要提前将库存信息写入 Redis（手动或者程序自动都可以）。
主要代码如下：
@Override public int createOptimisticOrderUseRedis(int sid) throws Exception { //检验库存，从 Redis 获取 Stock stock = checkStockByRedis(sid); //乐观锁更新库存 以及更新 Redis saleStockOptimisticByRedis(stock); //创建订单 int id = createOrder(stock); return id ; } private Stock checkStockByRedis(int sid) throws Exception { Integer count = Integer.parseInt(redisTemplate.opsForValue().get(RedisKeysConstant.STOCK_COUNT + sid)); Integer sale = Integer.parseInt(redisTemplate.opsForValue().get(RedisKeysConstant.STOCK_SALE + sid)); if (count.equals(sale)){ throw new RuntimeException("库存不足 Redis currentCount=" + sale); } Integer version = Integer.parseInt(redisTemplate.opsForValue().get(RedisKeysConstant.STOCK_VERSION + sid)); Stock stock = new Stock() ; stock.setId(sid); stock.setCount(count); stock.setSale(sale); stock.setVersion(version); return stock; } /** * 乐观锁更新数据库 还要更新 Redis * @param stock */ private void saleStockOptimisticByRedis(Stock stock) { int count = stockService.updateStockByOptimistic(stock); if (count == 0){ throw new RuntimeException("并发更新库存失败") ; } //自增 redisTemplate.opsForValue().increment(RedisKeysConstant.STOCK_SALE + stock.getId(),1) ; redisTemplate.opsForValue().increment(RedisKeysConstant.STOCK_VERSION + stock.getId(),1) ; }
压测看看实际效果 /createOptimisticLimitOrderByRedis/1：
最后发现数据没问题，数据库的请求与并发也都下来了。
乐观锁更新 + 分布式限流 + Redis 缓存 + Kafka 异步
最后的优化还是想如何来再次提高吞吐量以及性能的。
我们上文所有例子其实都是同步请求，完全可以利用同步转异步来提高性能啊。
这里我们将写订单以及更新库存的操作进行异步化，利用 Kafka 来进行解耦和队列的作用。
每当一个请求通过了限流到达了 Service 层通过了库存校验之后就将订单信息发给 Kafka ，这样一个请求就可以直接返回了。
消费程序再对数据进行入库落地。
因为异步了，所以最终需要采取回调或者是其他提醒的方式提醒用户购买完成。
这里代码较多就不贴了，消费程序其实就是把之前的 Service 层的逻辑重写了一遍，不过采用的是 SpringBoot。
感兴趣的朋友可以看下。
总结
其实经过上面的一顿优化总结起来无非就是以下几点：
- 尽量将请求拦截在上游。
- 还可以根据 UID 进行限流。
- 最大程度的减少请求落到 DB。
- 多利用缓存。
- 同步操作异步化。
- fail fast，尽早失败，保护应用。
