前言

大流量情況下的庫存是老生常談的問題了，在這里我整理一下mysql和redis應(yīng)對(duì)扣除庫存的方案，采用jmeter進(jìn)行壓測。

JMETER設(shè)置

庫存初始值50，線程數(shù)量1000個(gè)，1秒以內(nèi)啟動(dòng)全部，一個(gè)線程循環(huán)2次，共2000個(gè)請(qǐng)求

MySQL方案

初始方案

    <update id="decreaseStock">
        UPDATE stock
        SET stock_num = stock_num - 1
        WHERE id = #{id}
    </update>

這種情況下，在并發(fā)條件肯定會(huì)出現(xiàn)超賣的

進(jìn)行修改：

    <update id="decreaseStock">
        UPDATE stock
        SET stock_num = stock_num - 1
        WHERE id = #{id} AND stock_num >= 1
    </update>

增加AND stock_num >= 1條件，即可避免超賣。

相關(guān)代碼：

    @PostMapping(value = "/decreaseStock/{id}")	
    public ResponseEntity<Object> decreaseStock(@PathVariable("id") Integer id) {
        int result = stockService.decreaseStock(id);
        return result == 1 ? new ResponseEntity<>("decreaseStock successfully", HttpStatus.OK) : new ResponseEntity<>("decreaseStock failed", HttpStatus.OK);
    }

壓測情況：

根據(jù)Throught可知一秒可以處理200個(gè)事務(wù)（TPS)

如果說系統(tǒng)的并發(fā)量不高，則可以以這種方案進(jìn)行防止庫存超賣，但要注意，在可重復(fù)讀隔離級(jí)別情況下，如果where的條件字段沒有索引的話，進(jìn)行update語句會(huì)使整個(gè)表被鎖住，如果這里使用的where條件不是主鍵id而是product_name，那么需要給這個(gè)字段加索引。

在RR可重復(fù)讀隔離級(jí)別下，如果where條件沒有命中索引，那么會(huì)基于next-key lock（記錄鎖和間隙鎖的組合）對(duì)整個(gè)表的所有記錄加上這個(gè)鎖，進(jìn)行全表掃描，這個(gè)時(shí)候其他記錄想要更新就會(huì)被阻塞。

但是不一定是有了索引就不會(huì)鎖住整個(gè)表，這是由優(yōu)化器決定的，可以使用Explain語句來查看當(dāng)前語句是走的索引還是全表掃描，如果優(yōu)化器走的還是全標(biāo)掃描，可以使用 force index([index_name]) 強(qiáng)制使用某個(gè)索引。

改進(jìn)

在MySQL情況下還能有其他方案來提升性能嗎，在不借助Redis的情況（曾經(jīng)面試招銀網(wǎng)絡(luò)被問了這道題）

我當(dāng)時(shí)給出的回答是，把單個(gè)商品的庫存比如50個(gè)庫存，拆分成好幾份，一份10個(gè)，5份庫存，由于秒殺情況下流量很大，可以把這五份庫存分別放到五個(gè)數(shù)據(jù)庫里面，這樣性能至少是原先方案的5倍，那么還會(huì)出現(xiàn)新的問題，就是有些問題，負(fù)載均衡上的問題，可能會(huì)出現(xiàn)某些庫里還存在庫存，但是請(qǐng)求卻沒有打進(jìn)這個(gè)數(shù)據(jù)庫，而是打到庫存已經(jīng)沒有的數(shù)據(jù)庫里面。我當(dāng)時(shí)的想法是再搞個(gè)庫存表，這個(gè)庫存表采集各個(gè)商品的總庫存以及商品在各個(gè)分庫里面的庫存數(shù)量，然后再寫個(gè)服務(wù)，包含負(fù)載均衡的算法，將用戶的請(qǐng)求平均打到各個(gè)分庫去，當(dāng)某個(gè)分庫的庫存達(dá)到0的時(shí)候，去通知該服務(wù)，服務(wù)將這個(gè)庫剔除，使新的請(qǐng)求不會(huì)轉(zhuǎn)發(fā)過去。實(shí)際這種情況也是存在問題的，高并發(fā)下庫存為0的庫來不及被剔除，也會(huì)導(dǎo)致請(qǐng)求被打到庫存0的庫。

Redis方案

將庫存暫時(shí)放到Redis，然后從Redis進(jìn)行庫存扣減，能大大提升性能

壓測結(jié)果：

可見性能幾乎是MySQL的10倍了，但是這樣子在Redis里面會(huì)導(dǎo)致超賣

要確保Redis不超買，需要先查詢當(dāng)前的數(shù)量，如果大于0則進(jìn)行扣減，并且查詢和扣減需要為原子性，這里就需要借助lua腳本，將這兩次操作寫到一起。

加了Lua腳本的代碼：

    private static final String LUA_DECRESE_STOCK_PATH = "lua/decreseStock.lua";

    @PostMapping(value = "/decreaseStockByRedis/{id}")
    public ResponseEntity<Object> decreaseStockByRedis(@PathVariable("id") Integer id) {

        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource(LUA_DECRESE_STOCK_PATH)));
        redisScript.setResultType(Long.class);
        
        // 執(zhí)行Lua腳本
        Long result = (Long) redisTemplate.execute(redisScript, Collections.singletonList(id));

        // 返回結(jié)果判斷
        return (result != null && result == 1) ? new ResponseEntity<>("decreaseStock successfully", HttpStatus.OK) : new ResponseEntity<>("decreaseStock failed", HttpStatus.OK);
    }

lua腳本放在resource/lua/decreseStock.lua

local key = KEYS[1]

-- 檢查鍵是否存在
local exists = redis.call('EXISTS', key)
if exists == 1 then
    -- 鍵存在，獲取值
    local value = redis.call('GET', key)
    if tonumber(value) > 0 then
        -- 如果值大于0，則遞減
        redis.call('DECR', key)
        return 1  -- 表示遞減成功
    else
        return 0  -- 表示遞減失敗，值不大于0
    end
else
    return -1  -- 表示遞減失敗，鍵不存在
end

Redis同步庫存到MySQL

但是在Redis扣減了庫存，總需要同步到MySQL里面

@PostMapping(value = "/decreaseStockByRedis/{id}")
    public ResponseEntity<Object> decreaseStockByRedis(@PathVariable("id") Integer id) {

        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource(LUA_DECRESE_STOCK_PATH)));
        redisScript.setResultType(Long.class);

        // 執(zhí)行Lua腳本
        Long redisResult = (Long) redisTemplate.execute(redisScript, Collections.singletonList(id));
        int dataBaselResult = 0;
        if (redisResult == 1) {
            dataBaselResult = stockService.decreaseStock(id);
        }
        // 返回結(jié)果判斷
        return (dataBaselResult == 1 && redisResult == 1) ? new ResponseEntity<>("decreaseStock successfully", HttpStatus.OK) : new ResponseEntity<>("decreaseStock failed", HttpStatus.OK);
    }

直接按照上述代碼來寫，刪Redis后同時(shí)將庫存同步到MySQL，相當(dāng)于使用了Redis性能又沒有提升。

其實(shí)選擇了Redis來進(jìn)行庫存扣減，那么MySQL的庫存并不需要去實(shí)時(shí)進(jìn)行更新，只需要庫存達(dá)到最終一致性即可，即先對(duì)Redis的庫存進(jìn)行更新，然后再異步同步到MySQL的庫存。

如果使用spring的異步線程來解決，會(huì)不會(huì)出現(xiàn)同步MySQL失敗導(dǎo)致數(shù)據(jù)最終不一致呢，在流量很多的情況下，系統(tǒng)本身就處于壓力大的情況，再使用異步線程會(huì)占用額外的資源，最好的方法是引入MQ，把庫存的同步信息交給MQ，MQ再交到消費(fèi)系統(tǒng)，進(jìn)行減庫存的操作，由MQ保證消息被消費(fèi)，實(shí)現(xiàn)最終一致性。

部分代碼如下，由MQ product發(fā)出，再由consumer進(jìn)行消費(fèi)：

    private final DecreaseStockProduce decreaseStockProduce;

    @PostMapping(value = "/decreaseStockByRedis/{id}")
    public ResponseEntity<Object> decreaseStockByRedis(@PathVariable("id") String id) {

        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource(LUA_DECRESE_STOCK_PATH)));
        redisScript.setResultType(Long.class);

        // 執(zhí)行Lua腳本
        Long redisResult = (Long) redisTemplate.execute(redisScript, Collections.singletonList(id));
        if (redisResult == 1) {
            // 發(fā)送消息
            try {
                DecreaseStockEvent decreaseStockEvent = DecreaseStockEvent.builder()
                        .id(id)
                        .build();
                SendResult sendResult = decreaseStockProduce.sendMessage(decreaseStockEvent);
                if (!Objects.equals(sendResult.getSendStatus(), SendStatus.SEND_OK)) {
                    log.error("消息發(fā)送錯(cuò)誤，請(qǐng)求參數(shù)：{}", id);
                }
            } catch (Exception e) {
                log.error("消息發(fā)送錯(cuò)誤，請(qǐng)求參數(shù)：{}", id, e);
            }
        }

        // 返回結(jié)果判斷
        return (redisResult == 1) ? new ResponseEntity<>("decreaseStock successfully", HttpStatus.OK) : new ResponseEntity<>("decreaseStock failed", HttpStatus.OK);
    }

MQ [TIMEOUT_CLEAN_QUEUE] broker busy問題

這里直接壓測會(huì)報(bào)下面的錯(cuò)誤，并且這個(gè)時(shí)候查看redis庫存已經(jīng)減到0，到是MySQL只減到了37

針對(duì)MQ [TIMEOUT_CLEAN_QUEUE] broker busy問題，需要去修改MQ的broker.conf文件

針對(duì)TIMEOUT_CLEAN_QUEUE broker busy問題，需要去修改MQ的broker.conf文件，上述的201ms超時(shí)了，我這里將等待時(shí)間改為400，并且將線程數(shù)設(shè)置為64，這個(gè)線程數(shù)可以根據(jù)實(shí)際壓測情況進(jìn)行調(diào)整。

# 發(fā)消息線程池?cái)?shù)量
sendMessageThreadPoolNums=64
# 拉消息線程池?cái)?shù)量
pullMessageThreadPoolNums=64
waitTimeMillsInSendQueue=400

現(xiàn)在再進(jìn)行壓測，發(fā)現(xiàn)tps能跑到1000，相比直接入庫mysql的200已經(jīng)是提升很大了。

雖然性能提高，也實(shí)現(xiàn)庫存的同步，但這個(gè)性能下還是會(huì)存在一些問題：

比如MQ消息發(fā)送失敗、或者M(jìn)ySQL庫存扣減失敗，并且實(shí)際情況還有訂單的生成和庫存之間的一致性也要考慮。

對(duì)于上述這些問題，可以查看我的另外一篇博客：

RocketMQ事務(wù)消息在訂單創(chuàng)建和庫存扣減的使用 - Scotyzh - 博客園 (cnblogs.com)

總結(jié)

以上是生活随笔為你收集整理的mysql和redis库存扣减和优化的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。