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

系统的特点

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

优化思路

后端优化:请求拦截在系统的上游。
1. 限流:屏蔽掉无用的流量,允许少部分流量走后端,假设库存现在为10,有1000个购买请求,最终只有10个成功,99%无效。
2. 削峰:秒杀请求在时间上高度集中,一瞬间很容易压垮系统,因此需要对系统进行削峰处理,缓冲流量,尽量让服务器对资源进行平缓处理。
3. 异步:将同步请求转换为异步请求,来提高流量,本质上也是削峰处理。
4. 利用缓存,创建订单时,每次都需要先查询判断库存,只有少部分成功的请求才能创建订单,因此可以把商品信息放入缓存中,减少数据库的压力。
前端优化:
1. 限流:前端答题,或者验证码,来分散用户的请求。
2. 禁止重复提交,限定每个用户发起一次秒杀之后,需要等待才可以发起另外一次请求。从而减少用户重复的请求。
3. 本地标记,用户成功秒杀到商品后,将提交按钮重置为灰色,禁止用户再次提交请求。
4. 动静分离,将前端静态数据直接缓存到用户最贱的地方,例如用户的浏览器中。
反作弊优化:
1. 隐藏秒杀接口,如果秒杀地址直接暴露,在秒杀开始的时候会被恶意用户来耍接口,因此需要用户在秒杀之后才能拿到url和验证md5.
2. 同一个账号多次发出请求,只有一个生肖。
3. 多个账号一次发出多个请求,直接需要弹出验证码。
4. 多个账号不同ip发起不同请求,通过检测账号活跃度以及等级信息获取参与秒杀的资格。

代码优化

Jmetter压力测试并发量变化图

2万 | 北京做Java开发月薪 2 万,so easy!插图

基本的秒杀逻辑

@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}")

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

超卖问题出现场景
2万 | 北京做Java开发月薪 2 万,so easy!插图1
悲观锁虽然可以解决超卖问题,但是由于加锁时间更长,会长时间的限制其他用户的访问,导致很多请求等待锁,卡死在这里,如果这种请求很多就会耗尽连接,系统出现异常,乐观锁默认不加锁,可以承受较高并发。
2万 | 北京做Java开发月薪 2 万,so easy!插图2

@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个请求是无效的,所以这里就需要使用限流方法。
2万 | 北京做Java开发月薪 2 万,so easy!插图3

@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 缓存商品库存信息更新

即使能够过滤掉大部分请求,但是仍然会有大部分落到数据库中,这里直接使用缓存来减少数据库的使用。以及对数据库的压力。
2万 | 北京做Java开发月薪 2 万,so easy!插图4

缓存预热

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

@Component
public class RedisPreheatRunner implements ApplicationRunner {

    @Autowired
    private StockService stockService;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 从数据库中查询热卖商品,商品 id 为 1
        Stock 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 又更新了缓存,这种情况下缓存中就是脏数据,具体逻辑如下图所示:
2万 | 北京做Java开发月薪 2 万,so easy!插图5
那么,如果避免这个问题呢?就是缓存不做更新,仅做删除,先更新数据库再删除缓存。对于上面的问题,A 更新了数据库,还没来得及删除缓存,B 又更新了数据库,接着删除了缓存,然后 A 删除了缓存,这样只有下次缓存未命中时,才会从数据库中重建缓存,避免了脏数据。但是,也会有极端情况出现脏数据,A 做查询操作,没有命中缓存,从数据库中查询,但是还没来得及更新缓存,B 就更新了数据库,接着删除了缓存,然后 A 又重建了缓存,这时 A 中的就是脏数据,如下图所示。但是这种极端情况需要数据库的写操作前进入数据库,又晚于写操作删除缓存来更新缓存,发生的概率极其小,不过为了避免这种情况,可以为缓存设置过期时间。
2万 | 北京做Java开发月薪 2 万,so easy!插图6
安装先更新数据库再删除缓存的策略来执行,代码如下所示

@Override
public int createOrderWithLimitAndRedis(int sid) throws Exception {
    // 校验库存,从 Redis 中获取
    Stock stock = checkStockWithRedis(sid);
    // 乐观锁更新库存和Redis
    saleStockOptimsticWithRedis(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("并发更新库存失败");
    }
}

由于使用了乐观锁更新数据库,因此在使用先更新数据库数据再更新缓存的方式,实际情况是:
2万 | 北京做Java开发月薪 2 万,so easy!插图7

@Override
public int createOrderWithLimitAndRedis(int sid) throws Exception {
    // 校验库存,从 Redis 中获取
    Stock stock = checkStockWithRedis(sid);
    // 乐观锁更新库存和Redis
    saleStockOptimsticWithRedis(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("并发更新库存失败") ;
    }
    // 更新 Redis
    StockWithRedis.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 来缓冲瞬时流量,将同步的直接调用转成异步的间接推送,中间通过一个队列在一端承接瞬时的流量洪峰,在另一端平滑地将消息推送出去。
2万 | 北京做Java开发月薪 2 万,so easy!插图8

// 向 Kafka 发送消息
public void createOrderWithLimitAndRedisAndKafka(int sid) throws Exception {
    // 校验库存
    Stock stock = checkStockWithRedis(sid);
    // 下单请求发送至 kafka,需要序列化 stock
    kafkaTemplate.send(kafkaTopic, gson.toJson(stock));
    log.info("消息发送至 Kafka 成功");
}
// 监听器从 Kafka 拉取消息
public class ConsumerListen {

    private Gson gson = new GsonBuilder().create();

    @Autowired
    private OrderService orderService;

    @KafkaListener(topics = "SECONDS-KILL-TOPIC")
    public void listen(ConsumerRecord<String, String> record) throws Exception {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        // Object -> String
        String message = (String) kafkaMessage.get();
        // 反序列化
        Stock stock = gson.fromJson((String) message, Stock.class);
        // 创建订单
        orderService.consumerTopicToCreateOrderWithKafka(stock);
    }
}
// Kafka 消费消息执行创建订单业务
public int consumerTopicToCreateOrderWithKafka(Stock stock) throws Exception {
    // 乐观锁更新库存和 Redis
    saleStockOptimsticWithRedis(stock);
    int res = createOrder(stock);
    if (res == 1) {
        log.info("Kafka 消费 Topic 创建订单成功");
    } else {
        log.info("Kafka 消费 Topic 创建订单失败");
    }

    return res;
}

关于作者

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