秒殺系統設計

準備工做

新建一個JMeter測試計劃，添加以下內容

添加步驟：

1. 右擊測試計劃 --> 添加 --> 線程 （用戶）--> 線程組，線程數設置5000，Ramp-Up設爲1，也就是1啓動5000個線程
2. 右擊線程組 --> 添加 -->取樣器 --> HTTP請求，配置要訪問的路徑
3. 右擊HTTP請求 --> 添加 --> 配置元件--> HTTP信息頭管理器，這個下面說明
4. 右擊線程組 --> 添加 -->監聽器 -->察看結果樹/聚合報告

說明一下：

因爲全部的資源都被OAuth2保護起來了，因此想要訪問必需要通過登錄受權的步驟，爲了方便起見，先在瀏覽器正常訪問一個資源進行受權，而後獲取裏面的cookie值，放入HTTP信息頭管理器中

在Cookie中有個 JSESSIONID ，把它們整個放到HTTP信息頭管理器

測試

1、最簡單的加鎖

程序加synchronized鎖，先讀取數據庫信息，而後自減，再更新，全部的邏輯操做都在程序中完成

@GetMapping("/order")
    public String reduceStack(@Param("id") String id) {
        Integer number;
        synchronized (this) {
            Seckill seckill = seckillService.getById(id);
            number = seckill.getNumber();

            if (number > 0) {
                seckill.setNumber(--number);
                seckillService.updateById(seckill);
                number = seckillService.getById(id).getNumber();
            }
        }

        ResultData<Integer> resultData = new ResultData<Integer>(20000, "number", number);

        Gson gson = new GsonBuilder().setDateFormat(DateFormat.FULL, DateFormat.FULL).create();
// System.out.println(System.currentTimeMillis()-start);
        return gson.toJson(resultData);
    }
複製代碼

顯然，能夠看到吞吐量低

由於這段程序進行了加鎖，並且全部的邏輯都在程序裏執行，和數據庫的交互也存在時間延遲

2、在數據庫中完成

自檢操做在數據庫中完成，SeckillMapper中添加函數

@Update("update seckill set number=number-1 where id=1 and number > 0 ")
    Integer minusOne();
複製代碼

減庫存邏輯函數修改成

@GetMapping("/order")
    public String reduceStack(@Param("id") Integer id) {
// long start=System.currentTimeMillis();
        Integer number=-1;
        if (seckillService.minusStack(id)!=0){
            number = seckillService.getById(id).getNumber();
        }
        ResultData<Integer> resultData = new ResultData<Integer>(20000, "number", number);

        Gson gson = new GsonBuilder().setDateFormat(DateFormat.FULL, DateFormat.FULL).create();
// System.out.println(System.currentTimeMillis()-start);
        return gson.toJson(resultData);
    }
複製代碼

測試結果

因爲去掉了鎖的限制，整個邏輯只有對數據庫一行代碼的操做，在提高速度的同時也保證了原子性

3、使用redis

分佈式鎖

這個屬於悲觀鎖，原理大概是一個線程在操做數據的時候加一把鎖，不容許其餘線程進行操做，主要利用setnx命令（ SET if Not exists ），就是當key不存在時，將key設爲value，並返回1，不然返回0。操做結束後將該key刪除。同時爲了防止發生死鎖，要設置key的過時時間

這裏設置key爲lock，value=1，過時時間是1s，具體的redisUtil方法後面貼出

goods的值在初始化中進行設置，預先加載到內存中

if (!redisUtil.setnx("lock",1,1))
            return "so busy";
        Integer number = (Integer) redisUtil.get("goods");
        if (number > 0) {
             number=redisUtil.decrBy("goods",1);
        }
        redisUtil.unlock("lock");
複製代碼

測試結果以下，同時也沒有出現超賣狀況

樂觀鎖

樂觀鎖是基於數據版本實現的，數據庫中是在表中添加version字段，在讀取數據時將version一同讀出，以後進行寫操做時對version加1。提交數據時若是該值比當前表中記錄的值大，則更新，不然就是過時數據。在redis中，可使用watch加事務實現，經過watch監視指定的key，當exec時若是key發生改變，則整個事務失敗

注意

redis是單線程，單個命令的執行是原子性的，可是redis在事務上沒有任何原子性的限制，因此事務不是原子性的。事務能夠理解爲一個打包的批量執行腳本，中間某條指令的失敗不會致使前面指令的回滾，也不會形成後續指令中止。

可是

1. discard能夠取消事務
2. watch一個key，若是事務exec以前這個key被改動，那麼事務將被打斷

public boolean watch(String key) {
        redisTemplate.watch(key);
        Integer number = (Integer) get(key);
        if (number <= 0) {
            return false;
        }
        redisTemplate.multi();
        redisTemplate.opsForValue().decrement(key, 1);
        List<Object> list = redisTemplate.exec();
        return list.size() != 0;
    }
複製代碼

從結果上看，兩種方法的吞吐量好像差很少，樂觀鎖的還低一點，不太明白這樣是否正常，按理說應該高一點纔對。

lua腳本

上面樂觀鎖的處理略顯臃腫，須要watch一個key，還要開啓事務等一系列操做。那麼若是優雅的進行原子性的操做呢？這時候lua就出來了

使用lua腳本後，redis程序會有明顯的性能提高

• 減小網絡io操做：上節的操做會向redis服務器發起屢次請求，如今用一個請求便可完成
• 原子操做：redis會將整個腳本做爲一個總體運行
• 複用：腳本會永久存儲在redis中

redis-cli中先試試

eval "return redis.call('DECRBY',KEYS[1],1)" key-num [key1 key2 ....] [value1 value2....]
複製代碼

• eval命令表示執行lua腳本
• 雙引號裏是具體的內容
• KEYS[1]對應的是後面傳入的key參數，還有ARGV[1]對應的是value參數
• key-num表示key的個數
• [key1 key2]是key做爲參數傳給lua，要和key-num對應
• [value1 value2]也是參數

RedisUtil中添加

private DefaultRedisScript<Long> redisScript;

	@PostConstruct
    public void init() {
        redisScript = new DefaultRedisScript<>();
        redisScript.setResultType(Long.class);
        redisScript.setScriptText("local number = tonumber(redis.call('get',KEYS[1]))\n" +
                "if number <= 0 then\n" +
                " return 0;\n" +
                "end\n" +
                "return redis.call('DECRBY',KEYS[1],1);");
// redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("seckill.lua")));
// redisScript.setLocation(new ClassPathResource("seckill.lua"));
    }
複製代碼

注意：這裏貌似只能使用Long，Integer會報錯

直接把腳本寫成string，不用每次都要從文件加載，速度會快一些

public Long lua(String key) {
        List<String> keyList = new ArrayList<>();
        keyList.add(key);
        return redisTemplate.execute(redisScript, keyList);
    }
複製代碼

keyList用於存儲須要用到的key

4、RabbitMQ出場

RabbitMQ並非爲了取代redis，只是存儲秒殺信息用於訂單處理，因此在秒殺這部分功能仍是使用redis

配置一個隊列

@Configuration
public class RabbitDirectConfig {
    @Bean
    public Queue seckillQueue(){
        return new Queue("seckill");
    }
}
複製代碼

仍是用redis進行交互，成功後將手機號放入隊列

@GetMapping("/orderMq")
    public String reduceStackMq(@Param("id") Integer id, @Param("phone") String phone) {
        if (localOverMap.get(id))
            return commonUtil.toJson(ResponseState.OK, "number", -1);

        Long number = redisUtil.lua(KEY, SUCCESS, phone);
        if (number >= 0) {
// 成功
            amqpTemplate.convertAndSend("seckill",phone);
        }else
            localOverMap.put(id,true);
        return commonUtil.toJson(ResponseState.OK, "number", number);
    }
複製代碼

坑

注意

1. 若是線程數選的過大，好比10w，可能會報 Address already in use : connect 緣由：windows提供給TCP/IP連接的端口爲 1024-5000，而且要四分鐘來循環回收它們，就致使咱們在短期內跑大量的請求時將端口占滿了，致使如上報錯。

   解決辦法（在jmeter所在服務器操做）：

   1.cmd中輸入regedit命令打開註冊表；

   2.在 HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Services\Tcpip\Parameters右鍵Parameters；

   3.添加一個新的DWORD，名字爲MaxUserPort，若是有的話就不用新建；

   4.而後雙擊MaxUserPort，輸入數值數據爲65534，基數選擇十進制；

   5.完成以上操做，務必重啓機器，問題解決，親測有效；

2. org.apache.http.conn.HttpHostConnectException: Connect to localhost:80 JMeter的HTTP請求裏的服務器名稱要和工程裏application.yml配置同樣，好比都是localhost或者192.168.0.xxx 更多文章見我的博客 zheyday.github.io/