Hello！我是小小，今天開始本周的最后一篇，在北京做Java如何做到月薪上萬，很簡單，只要會秒殺，即可輕松做到月薪上萬，

系統的特點

高性能： 秒殺設計大量的并發讀和并發寫，因此支持高并發訪問這點相當的重要，

一致性：秒殺商品減庫存的實作方式同樣很關鍵，有限數量的商品在同一時刻被很多倍的請求同時來減少庫存，在大并發更新的時候都要保證資料的準確性，

高可用：秒殺系統在一瞬間都會涌入大量的流量，為了避免系統宕機，需要高可用，需要做好流量限制，

優化思路

后端優化：請求攔截在系統的上游，

1. 限流：屏蔽掉無用的流量，允許少部分流量走后端，假設庫存現在為10，有1000個購買請求，最終只有10個成功，99%無效，

2. 削峰：秒殺請求在時間上高度集中，一瞬間很容易壓垮系統，因此需要對系統進行削峰處理，緩沖流量，盡量讓服務器對資源進行平緩處理，

3. 異步：將同步請求轉換為異步請求，來提高流量，本質上也是削峰處理，

4. 利用快取，創建訂單時，每次都需要先查詢判斷庫存，只有少部分成功的請求才能創建訂單，因此可以把商品資訊放入快取中，減少資料庫的壓力，

前端優化：

1. 限流：前端答題，或者驗證碼，來分散用戶的請求，

2. 禁止重復提交，限定每個用戶發起一次秒殺之后，需要等待才可以發起另外一次請求，從而減少用戶重復的請求，

3. 本地標記，用戶成功秒殺到商品后，將提交按鈕重置為灰色，禁止用戶再次提交請求，

4. 動靜分離，將前端靜態資料直接快取到用戶最賤的地方，例如用戶的瀏覽器中，

反作弊優化：

1. 隱藏秒殺介面，如果秒殺地址直接暴露，在秒殺開始的時候會被惡意用戶來耍介面，因此需要用戶在秒殺之后才能拿到url和驗證md5.

2. 同一個賬號多次發出請求，只有一個生肖，

3. 多個賬號一次發出多個請求，直接需要彈出驗證碼，

4. 多個賬號不同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 {

    @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 又更新了快取，這種情況下快取中就是臟資料，具體邏輯如下圖所示：

那么，如果避免這個問題呢？就是快取不做更新，僅做洗掉，先更新資料庫再洗掉快取，對于上面的問題，A 更新了資料庫，還沒來得及洗掉快取，B 又更新了資料庫，接著洗掉了快取，然后 A 洗掉了快取，這樣只有下次快取未命中時，才會從資料庫中重建快取，避免了臟資料，但是，也會有極端情況出現臟資料，A 做查詢操作，沒有命中快取，從資料庫中查詢，但是還沒來得及更新快取，B 就更新了資料庫，接著洗掉了快取，然后 A 又重建了快取，這時 A 中的就是臟資料，如下圖所示，但是這種極端情況需要資料庫的寫操作前進入資料庫，又晚于寫操作洗掉快取來更新快取，發生的概率極其小，不過為了避免這種情況，可以為快取設定過期時間，

安裝先更新資料庫再洗掉快取的策略來執行，代碼如下所示

@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("并發更新庫存失敗");
    }
}

由于使用了樂觀鎖更新資料庫，因此在使用先更新資料庫資料再更新快取的方式，實際情況是：

@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 異步

需要spring+高并發+分布式+JVM+mysql+springcloud+演算法與資料結構+netty+計算機底層知識加扣扣群：792133408

服務器的資源是恒定的，你用或者不用它的處理能力都是一樣的，所以出現峰值的話，很容易導致忙到處理不過來，閑的時候卻又沒有什么要處理，因此可以通過削峰來延緩用戶請求的發出，讓服務端處理變得更加平穩，

專案中采用的是用訊息佇列 Kafka 來緩沖瞬時流量，將同步的直接呼叫轉成異步的間接推送，中間通過一個佇列在一端承接瞬時的流量洪峰，在另一端平滑地將訊息推送出去，

// 向 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());
      需要spring+高并發+分布式+JVM+mysql+springcloud+演算法與資料結構+netty+計算機底層知識加扣扣裙：792133408
        // 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;
}

需要spring+高并發+分布式+JVM+mysql+springcloud+演算法與資料結構+netty+計算機底層知識加扣扣群：792133408

關于作者

我是小小，雙魚座的程式猿，我們下期再見~