Hello！我是小小，今天开始本周的最后一篇，在北京做Java如何做到月薪上万，很简单，只要会秒杀，即可轻松做到月薪上万。

系统的特点

高性能：秒杀设计大量的并发读和并发写，因此支持高并发访问这点相当的重要。一致性：秒杀商品减库存的实现方式同样很关键，有限数量的商品在同一时刻被很多倍的请求同时来减少库存，在大并发更新的时候都要保证数据的准确性。高可用：秒杀系统在一瞬间都会涌入大量的流量，为了避免系统宕机，需要高可用，需要做好流量限制。

优化思路

后端优化：请求拦截在系统的上游。

1. 限流：屏蔽掉无用的流量，允许少部分流量走后端，假设库存现在为10，有1000个购买请求，最终只有10个成功，99%无效。

2. 削峰：秒杀请求在时间上高度集中，一瞬间很容易压垮系统，因此需要对系统进行削峰处理，缓冲流量，尽量让服务器对资源进行平缓处理。

3. 异步：将同步请求转换为异步请求，来提高流量，本质上也是削峰处理。

4. 利用缓存，创建订单时，每次都需要先查询判断库存，只有少部分成功的请求才能创建订单，因此可以把商品信息放入缓存中，减少数据库的压力。前端优化：

5. 限流：前端答题，或者验证码，来分散用户的请求。

6. 禁止重复提交，限定每个用户发起一次秒杀之后，需要等待才可以发起另外一次请求。从而减少用户重复的请求。

7. 本地标记，用户成功秒杀到商品后，将提交按钮重置为灰色，禁止用户再次提交请求。

8. 动静分离，将前端静态数据直接缓存到用户最贱的地方，例如用户的浏览器中。反作弊优化：

9. 隐藏秒杀接口，如果秒杀地址直接暴露，在秒杀开始的时候会被恶意用户来耍接口，因此需要用户在秒杀之后才能拿到url和验证md5.

10. 同一个账号多次发出请求，只有一个生肖。

11. 多个账号一次发出多个请求，直接需要弹出验证码。

12. 多个账号不同ip发起不同请求，通过检测账号活跃度以及等级信息获取参与秒杀的资格。

代码优化

Jmetter压力测试并发量变化图

基本的秒杀逻辑

@Override
public int createWrongOrder(int sid) throws Exception {// 数据库校验库存Stock stock = checkStock(sid);// 扣库存(无锁)saleStock(stock);// 生成订单int res = createOrder(stock);return res;
}
private Stock checkStock(int sid) throws Exception {Stock stock = stockService.getStockById(sid);if (stock.getCount() < 1) {throw new RuntimeException("库存不足");}return stock;
}
private int saleStock(Stock stock) {stock.setSale(stock.getSale() + 1);stock.setCount(stock.getCount() - 1);return stockService.updateStockById(stock);
}
private int createOrder(Stock stock) throws Exception {StockOrder order = new StockOrder();order.setSid(stock.getId());order.setName(stock.getName());order.setCreateTime(new Date());int res = orderMapper.insertSelective(order);if (res == 0) {throw new RuntimeException("创建订单失败");}return res;
}
// 扣库存 Mapper 文件
@Update("UPDATE stock SET count = #{count, jdbcType = INTEGER}, name = #{name, jdbcType =                  VARCHAR}, " + "sale = #{sale,jdbcType = INTEGER},version = #{version,jdbcType = INTEGER} " + "WHERE id = #{id, jdbcType = INTEGER}")

乐观锁更新库存，解决超卖的问题

超卖问题出现场景悲观锁虽然可以解决超卖问题，但是由于加锁时间更长，会长时间的限制其他用户的访问，导致很多请求等待锁，卡死在这里，如果这种请求很多就会耗尽连接，系统出现异常，乐观锁默认不加锁，可以承受较高并发。

@Override
public int createOptimisticOrder(int sid) throws Exception {// 校验库存Stock stock = checkStock(sid);// 乐观锁更新saleStockOptimstic(stock);// 创建订单int id = createOrder(stock);return id;
}
// 乐观锁 Mapper 文件
@Update("UPDATE stock SET count = count - 1, sale = sale + 1, version = version + 1 WHERE " +"id = #{id, jdbcType = INTEGER} AND version = #{version, jdbcType = INTEGER}")

Redis 限流

当有10个商品，只有1000个并发请求，最终只有10个订单会创建成功，即，990个请求是无效的，所以这里就需要使用限流方法。

@Slf4j
public class RedisLimit {private static final int FAIL_CODE = 0;private static Integer limit = 5;/*** Redis 限流*/public static Boolean limit() {Jedis jedis = null;Object result = null;try {// 获取 jedis 实例jedis = RedisPool.getJedis();// 解析 Lua 文件String script = ScriptUtil.getScript("limit.lua");// 请求限流String key = String.valueOf(System.currentTimeMillis() / 1000);// 计数限流result = jedis.eval(script, Collections.singletonList(key), Collections.singletonList(String.valueOf(limit)));if (FAIL_CODE != (Long) result) {log.info("成功获取令牌");return true;}} catch (Exception e) {log.error("Limit 获取 Jedis 实例失败：", e);} finally {RedisPool.jedisPoolClose(jedis);}return false;}
}
// 在 Controller 中，每个请求到来先取令牌，获取到令牌再执行后续操作，获取不到直接返回 ERROR
public String createOptimisticLimitOrder(HttpServletRequest request, int sid) {int res = 0;try {if (RedisLimit.limit()) {res = orderService.createOptimisticOrder(sid);}} catch (Exception e) {log.error("Exception: " + e);}return res == 1 ? success : error;
}

Redis 缓存商品库存信息更新

即使能够过滤掉大部分请求，但是仍然会有大部分落到数据库中，这里直接使用缓存来减少数据库的使用。以及对数据库的压力。

缓存预热

在秒杀开始前，秒杀商品信息可以缓存到Redis中，那么秒杀开始后可以直接从Redis中获取。

@Component
public class RedisPreheatRunner implements ApplicationRunner {@Autowiredprivate StockService stockService;@Overridepublic void run(ApplicationArguments args) throws Exception {// 从数据库中查询热卖商品，商品 id 为 1Stock stock = stockService.getStockById(1);// 删除旧缓存RedisPoolUtil.del(RedisKeysConstant.STOCK_COUNT + stock.getCount());RedisPoolUtil.del(RedisKeysConstant.STOCK_SALE + stock.getSale());RedisPoolUtil.del(RedisKeysConstant.STOCK_VERSION + stock.getVersion());//缓存预热int sid = stock.getId();RedisPoolUtil.set(RedisKeysConstant.STOCK_COUNT + sid, String.valueOf(stock.getCount()));RedisPoolUtil.set(RedisKeysConstant.STOCK_SALE + sid, String.valueOf(stock.getSale()));RedisPoolUtil.set(RedisKeysConstant.STOCK_VERSION + sid,            String.valueOf(stock.getVersion()));}
}

缓存和数据一致性

首先看下先更新数据库，再更新缓存策略，假设 A、B 两个线程，A 成功更新数据，在要更新缓存时，A 的时间片用完了，B 更新了数据库接着更新了缓存，这是 CPU 再分配给 A，则 A 又更新了缓存，这种情况下缓存中就是脏数据，具体逻辑如下图所示：那么，如果避免这个问题呢？就是缓存不做更新，仅做删除，先更新数据库再删除缓存。对于上面的问题，A 更新了数据库，还没来得及删除缓存，B 又更新了数据库，接着删除了缓存，然后 A 删除了缓存，这样只有下次缓存未命中时，才会从数据库中重建缓存，避免了脏数据。但是，也会有极端情况出现脏数据，A 做查询操作，没有命中缓存，从数据库中查询，但是还没来得及更新缓存，B 就更新了数据库，接着删除了缓存，然后 A 又重建了缓存，这时 A 中的就是脏数据，如下图所示。但是这种极端情况需要数据库的写操作前进入数据库，又晚于写操作删除缓存来更新缓存，发生的概率极其小，不过为了避免这种情况，可以为缓存设置过期时间。安装先更新数据库再删除缓存的策略来执行，代码如下所示

@Override
public int createOrderWithLimitAndRedis(int sid) throws Exception {// 校验库存，从 Redis 中获取Stock stock = checkStockWithRedis(sid);// 乐观锁更新库存和RedissaleStockOptimsticWithRedis(stock);// 创建订单int res = createOrder(stock);return res;
}
// Redis 校验库存
private Stock checkStockWithRedisWithDel(int sid) throws Exception {Integer count = null;Integer sale = null;Integer version = null;List<String> data = RedisPoolUtil.listGet(RedisKeysConstant.STOCK + sid);if (data.size() == 0) {// Redis 不存在，先从数据库中获取，再放到 Redis 中Stock newStock = stockService.getStockById(sid);RedisPoolUtil.listPut(RedisKeysConstant.STOCK + newStock.getId(), String.valueOf(newStock.getCount()),String.valueOf(newStock.getSale()), String.valueOf(newStock.getVersion()));count = newStock.getCount();sale = newStock.getSale();version = newStock.getVersion();} else {count = Integer.parseInt(data.get(0));sale = Integer.parseInt(data.get(1));version = Integer.parseInt(data.get(2));}if (count < 1) {log.info("库存不足");throw new RuntimeException("库存不足 Redis currentCount: " + sale);}Stock stock = new Stock();stock.setId(sid);stock.setCount(count);stock.setSale(sale);stock.setVersion(version);// 此处应该是热更新，但是在数据库中只有一个商品，所以直接赋值stock.setName("手机");return stock;
}
private void saleStockOptimsticWithRedisWithDel(Stock stock) throws Exception {// 乐观锁更新数据库int res = stockService.updateStockByOptimistic(stock);// 删除缓存，应该使用 Redis 事务RedisPoolUtil.del(RedisKeysConstant.STOCK + stock.getId());log.info("删除缓存成功");if (res == 0) {throw new RuntimeException("并发更新库存失败");}
}

由于使用了乐观锁更新数据库，因此在使用先更新数据库数据再更新缓存的方式，实际情况是：

@Override
public int createOrderWithLimitAndRedis(int sid) throws Exception {// 校验库存，从 Redis 中获取Stock stock = checkStockWithRedis(sid);// 乐观锁更新库存和RedissaleStockOptimsticWithRedis(stock);// 创建订单int res = createOrder(stock);return res;
}
// Redis 中校验库存
private Stock checkStockWithRedis(int sid) throws Exception {Integer count = Integer.parseInt(RedisPoolUtil.get(RedisKeysConstant.STOCK_COUNT + sid));Integer sale = Integer.parseInt(RedisPoolUtil.get(RedisKeysConstant.STOCK_SALE + sid));Integer version = Integer.parseInt(RedisPoolUtil.get(RedisKeysConstant.STOCK_VERSION + sid));if (count < 1) {log.info("库存不足");throw new RuntimeException("库存不足 Redis currentCount: " + sale);}Stock stock = new Stock();stock.setId(sid);stock.setCount(count);stock.setSale(sale);stock.setVersion(version);// 此处应该是热更新，但是在数据库中只有一个商品，所以直接赋值stock.setName("手机");return stock;
}
// 更新 DB 和 Redis
private void saleStockOptimsticWithRedis(Stock stock) throws Exception {int res = stockService.updateStockByOptimistic(stock);if (res == 0){throw new RuntimeException("并发更新库存失败") ;}// 更新 RedisStockWithRedis.updateStockWithRedis(stock);
}
// Redis 多个写入操作的事务
public static void updateStockWithRedis(Stock stock) {Jedis jedis = null;try {jedis = RedisPool.getJedis();// 开始事务Transaction transaction = jedis.multi();// 事务操作RedisPoolUtil.decr(RedisKeysConstant.STOCK_COUNT + stock.getId());RedisPoolUtil.incr(RedisKeysConstant.STOCK_SALE + stock.getId());RedisPoolUtil.incr(RedisKeysConstant.STOCK_VERSION + stock.getId());// 结束事务List<Object> list = transaction.exec();} catch (Exception e) {log.error("updateStock 获取 Jedis 实例失败：", e);} finally {RedisPool.jedisPoolClose(jedis);}
}

kafak 异步

服务器的资源是恒定的，你用或者不用它的处理能力都是一样的，所以出现峰值的话，很容易导致忙到处理不过来，闲的时候却又没有什么要处理，因此可以通过削峰来延缓用户请求的发出，让服务端处理变得更加平稳。

项目中采用的是用消息队列 Kafka 来缓冲瞬时流量，将同步的直接调用转成异步的间接推送，中间通过一个队列在一端承接瞬时的流量洪峰，在另一端平滑地将消息推送出去。

// 向 Kafka 发送消息
public void createOrderWithLimitAndRedisAndKafka(int sid) throws Exception {// 校验库存Stock stock = checkStockWithRedis(sid);// 下单请求发送至 kafka，需要序列化 stockkafkaTemplate.send(kafkaTopic, gson.toJson(stock));log.info("消息发送至 Kafka 成功");
}
// 监听器从 Kafka 拉取消息
public class ConsumerListen {private Gson gson = new GsonBuilder().create();@Autowiredprivate OrderService orderService;@KafkaListener(topics = "SECONDS-KILL-TOPIC")public void listen(ConsumerRecord<String, String> record) throws Exception {Optional<?> kafkaMessage = Optional.ofNullable(record.value());// Object -> StringString message = (String) kafkaMessage.get();// 反序列化Stock stock = gson.fromJson((String) message, Stock.class);// 创建订单orderService.consumerTopicToCreateOrderWithKafka(stock);}
}
// Kafka 消费消息执行创建订单业务
public int consumerTopicToCreateOrderWithKafka(Stock stock) throws Exception {// 乐观锁更新库存和 RedissaleStockOptimsticWithRedis(stock);int res = createOrder(stock);if (res == 1) {log.info("Kafka 消费 Topic 创建订单成功");} else {log.info("Kafka 消费 Topic 创建订单失败");}return res;
}

关于作者

我是小小，双鱼座的程序猿，我们下期再见~

END

「 往期文章 」

数据 | 爬一次给你10W你敢爬么！！！

红包 | 让真正抢红包的人都能抢到红包！

领路人 | 想要成为2.0版马云？除了机会，你更需要一位领路人

扫描二维码

获取更多精彩

小明菜市场

来源：网络（侵删）

图片来源：网络（侵删）