秒殺系統(tǒng)之四：消息隊(duì)列異步處理訂單(RabbitMQ消息隊(duì)列)

5. 消息隊(duì)列異步處理訂單

我們之前通過數(shù)據(jù)庫中的樂觀鎖來控制超賣的問題大磺，并且也通過Jmeter壓力測試，那么如果并發(fā)量足夠大探膊，而且不對其進(jìn)行限制那么對于接口，對于數(shù)據(jù)庫和服務(wù)器都是一個很大的壓力待榔，此時逞壁，我們需要接口限流流济，我們通常使用令牌桶算法+樂觀鎖進(jìn)行對高并發(fā)的限制，但是如果遇到爬蟲進(jìn)行不斷的發(fā)送數(shù)據(jù)腌闯，這樣也會比正常用戶大概率秒殺到商品绳瘟，此時我們需要隱藏接口、帶MD5進(jìn)行雙向驗(yàn)證姿骏，和單用戶限制發(fā)送請求的頻率糖声。

除了這些方法，實(shí)際上我們還可以對于下單的異步處理分瘦，我們之前提過蘸泻，用戶進(jìn)行對商品秒殺的時候會在同一時間進(jìn)行高并發(fā)的請求流量到服務(wù)器中，如果每個請求都立即訪問數(shù)據(jù)庫進(jìn)行扣減庫存+寫入訂單的操作嘲玫，對數(shù)據(jù)庫的壓力是巨大的悦施。

那這樣我們可以通過RabbitMQ （消息隊(duì)列）對我們數(shù)據(jù)庫減輕壓力：當(dāng)"幸運(yùn)兒"成功的將其的秒殺請求放到消息隊(duì)列中，給其返回?fù)屬彸晒θネ牛瑢?shí)際上用戶并不關(guān)心自己的訂單號馬上返回抡诞，用戶只關(guān)心自己是否能夠成功搶購，所以對于生成訂單號土陪，減少庫存等操作我們可以通過異步處理訂單將數(shù)據(jù)寫入數(shù)據(jù)庫昼汗，比起多線程同步修改數(shù)據(jù)庫的操作，大大緩解了數(shù)據(jù)庫的連接壓力鬼雀，最主要的好處就表現(xiàn)在數(shù)據(jù)庫連接的減少

同步方式：大量請求快速占滿數(shù)據(jù)庫框架開啟的數(shù)據(jù)庫連接池顷窒，同時修改數(shù)據(jù)庫，導(dǎo)致數(shù)據(jù)庫讀寫性能驟減取刃。
異步方式：一條條消息以順序的方式寫入數(shù)據(jù)庫蹋肮，連接數(shù)幾乎不變（當(dāng)然，也取決于消息隊(duì)列消費(fèi)者的數(shù)量）璧疗。

image-20200930102122309.png

5.1 配置RabbitMQ

導(dǎo)入RabbitMQ依賴和fastJson

<!--spring boot stater data rabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.73</version>
</dependency>

使用Docker在服務(wù)器上安裝rabbitMQ

# 獲取最新的指定版本坯辩，該版本包含了web控制頁面
docker pull rabbitmq:management

# 默認(rèn)guest 用戶，密碼也是guest
docker run -d --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq:management

5672與15672的區(qū)別

5672：基于此協(xié)議的客戶端與消息中間件之間可以傳遞消息
15672：web控制頁面

5.2 properties配置文件

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
# ms為虛擬主機(jī)
spring.rabbitmq.virtual-host=/ms

5.3 配置config類

@Component
@Slf4j
public class OrderMqReceiver {
    @Autowired
    private StockDao stockDao;

    @Autowired
    private OrderDao orderDao;

    @Autowired
    private UserDao userDao;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @RabbitListener(queuesToDeclare = @Queue("orderQueue"))
    public void process(String msg){
        JSONObject jsonObject = JSONObject.parseObject(msg);
        log.info("OrderMqReceiver收到消息開始用戶下單流程：" + msg);
        //校驗(yàn)庫存
        Stock stock = checkStock(jsonObject.getInteger("id"));
        //更新庫存
        updateSale(stock);
        //創(chuàng)建訂單
        Integer order = createOrder(stock);
        log.info("訂單號為："+order);
        // 將訂單號和用戶id放入redis緩存
        stringRedisTemplate.opsForValue().set("orderId_"+ jsonObject.getInteger("id"),""+jsonObject.getInteger("userid"));
    }

    //校驗(yàn)庫存
    private Stock checkStock(Integer id){
        Stock stock = stockDao.checkStock(id);
        if(stock.getSale().equals(stock.getCount())){
            throw  new RuntimeException("庫存不足!!!");
        }
        return stock;
    }

    //扣除庫存
    private void updateSale(Stock stock){
        //在sql層面完成銷量的+1  和 版本號的+1 并且根據(jù)商品id和版本號同時查詢更新的商品
        int value = stockDao.updateSale(stock);
        // 更新失敗
        if (value == 0){
            throw new RuntimeException("購買失敗崩侠，請稍后重試");
        }
    }

    //創(chuàng)建訂單
    private Integer createOrder(Stock stock){
        // 因?yàn)镴ava是傳遞值漆魔，所以order對象地址傳給了Mybatis
        // mybatis根據(jù)表的創(chuàng)建id規(guī)則賦值飛order對象id
        // 所以當(dāng)創(chuàng)建訂單號的時候order就會得到一個id，我們可以直接獲取
        Order order = new Order();
        order.setSid(stock.getId()).setName(stock.getName()).setCreateDate(new Date());
        orderDao.createOrder(order);
        return order.getId();
    }
}

5.4 配置Service層

@Override
public void killMQ(Integer id, Integer userid, String md5) {

    //校驗(yàn)redis中秒殺商品是否超時
    //        if(!stringRedisTemplate.hasKey("kill"+id))
    //            throw new RuntimeException("當(dāng)前商品的搶購活動已經(jīng)結(jié)束啦~~");

    //先驗(yàn)證簽名
    String hashKey = "KEY_"+userid+"_"+id;
    String s = stringRedisTemplate.opsForValue().get(hashKey);
    if (s==null) throw  new RuntimeException("沒有攜帶驗(yàn)證簽名,請求不合法!");
    if (!s.equals(md5)) throw  new RuntimeException("當(dāng)前請求數(shù)據(jù)不合法,請稍后再試!");
    // 先從redis獲取該用戶訂單號的數(shù)量

    JSONObject object = new JSONObject();
    // 將商品id和用戶id放入JSON中却音，以至于多個參數(shù)進(jìn)行傳遞下
    object.put("id", id);
    object.put("userid", userid);
    // 定義一個消費(fèi)者,異步調(diào)用訂單操作
    // convertSendAndReceive可以接收返回值
    rabbitTemplate.convertAndSend("orderQueue",object.toJSONString());
}

測試調(diào)用

image-20200930112039095.png

5.4 限制購買數(shù)量

如果我們想要限制購買數(shù)量的話改抡，我們應(yīng)該在redis中存儲用戶購買信息，每次下單前獲取當(dāng)前已購買的數(shù)量系瓢，如果達(dá)到一定的數(shù)量則拋出異常阿纤，但是拋出異常我們必須捕獲，或者設(shè)置死信隊(duì)列

@RabbitListener(queuesToDeclare = @Queue("orderQueue"))
public void process(String msg){
    JSONObject jsonObject = JSONObject.parseObject(msg);
    Long numbers = stringRedisTemplate.opsForHash().size("order_userId_" + jsonObject.getInteger("userid"));
    // 驗(yàn)證購買次數(shù)有沒有超過5次
    // 如果不捕獲RuntimeException異常
    // 如果拋出異常夷陋，則消息消耗不掉欠拾，rabbitmq會一直不停的投送消息
    try {
        if (numbers < 5) {
            log.info("OrderMqReceiver收到消息開始用戶下單流程：" + msg);
            //校驗(yàn)庫存
            Stock stock = checkStock(jsonObject.getInteger("id"));
            //更新庫存
            updateSale(stock);
            //創(chuàng)建訂單
            Integer order = createOrder(stock);
            log.info("訂單號為："+order);
            // 如果想要一人限購一次將訂單號和用戶id放入redis緩存
            // 用戶id-訂單號-商品id
            stringRedisTemplate.opsForHash().put("order_userId_" + jsonObject.getInteger("userid"),order.toString(),jsonObject.getInteger("id").toString());
        };
        throw new RuntimeException("超過購買的數(shù)量!!!");
    }catch (RuntimeException e){

    }
}

測試調(diào)用

image-20200930112921807.png

數(shù)據(jù)庫查看確實(shí)只限制了5個購買

image-20200930112958778.png

image-20200930113019702.png

結(jié)束語：實(shí)際上秒殺系統(tǒng)并沒有那么簡單胰锌，還有很多復(fù)雜的東西，這里只是提供思路藐窄，每一步做什么资昧，下一步該做什么提供了思路，不至于到時候亂加中間件