你所不知道的庫存超限做法

在互聯網企業中，限購的做法，多種多樣，有的別出心裁，有的因循守舊，但是種種做法皆想達到的目的，無外乎幾種，商品賣的完，系統抗的住，庫存不超限。雖然短短數語，卻有着說不完，道不盡，輕者如釋重負，重者涕淚橫流的架構體驗。 但是，在實際開發過程中，庫存超限，作為其中最核心的一員，到底該怎么做，如何做才會是最合適的呢？

今天這篇文章，我將會展示給大家庫存限購的五種常見的做法，並對其利弊一一探討，由於這五種做法，有的在設計之初當做提案被否定掉的，有的在線上跑着，但是在沒有任何單元測試和壓測情況下，這幾種超限控制的做法也許是不符合你的業務的，所以不建議直接用於生產環境。我這里權當是做拋磚引玉，期待大家更好的做法。

工欲善其事必先利其器，在這里，我們將利用一台測試環境的redis服務器當做庫存超限控制的主戰場，先設置庫存量為10進去，然后根據此庫存量，一一展開，設置庫存代碼如下：

def set_storage():
    conn = redis_conn()
    key = "storage_seckill"
    current_storage = conn.get(key)
    if current_storage == None:
        conn.set(key, 10)

為了方便性，我這里使用了python語言來書寫邏輯，但是今天我們只是講解思想，語言這類的，大家可以自己嘗試轉一下。

上面就是我們的設置庫存到redis中的做法，很簡單，就是在redis中設置一個storage_seckill的庫存key，然后里面放上庫存量10.

 

超限限制做法一：先獲取當前庫存值進行比對，然后進行扣減操作

def storage_scenario_one():
    conn = redis_conn()
    key = "storage_seckill"
    current_storage = conn.get(key)
    current_storage_int = int(current_storage)
    if current_storage_int<=0 :
        return 0
    result = conn.decr(key)
    return result

首先，我們拿到當前的庫存值，然后看看是否已經扣減到了零，如果扣減到了零，則不繼續扣減，直接返回；如果庫存還有，則利用decr原子操作進行扣減，同時返回扣減后的庫存值。

此種做法在小並發量下訪問，問題不大；在對庫存量控制不嚴格的業務中，問題也不大。但是如果並發量比較大一些，同時業務要求嚴格控制庫存，那么此種做法是非常不合適的，原因在於，在高並發情況下，get命令，decr命令，都是分開發給redis的，這樣會導致比對的時候，很容易出現限制不住的情況，也就是會造成第六行的比對失效。

設想如下一個場景，AB兩個請求進來，A獲取的庫存值為1，B獲取的庫存值為1，然后兩個請求都被發到redis中進行扣減操作，然后這種場景下，A最后得到的庫存值為0；但是B最后得到的庫存值為-1，超限。

所以此種場景，由於在高並發下，get和decr操作不是一組原子性操作，會引發超限問題，被直接pass。

 

超限限制做法二：先扣減庫存，然后比對，最后根據情況回滾

def storage_scenario_two():
    conn = redis_conn()
    key = "storage_seckill"
    current = conn.decr(key)
    if current>=0:
        return current
    else:
        #回滾庫存
        conn.incr(key)
        return 0

首先，請求進來，直接對庫存值進行扣減，然后得到當前的庫存值；然后，對此庫存值進行校驗，如果庫存還有，則返回庫存值，如果庫存沒有了，則回滾庫存，以便於防止負庫存量的存在。

此做法，相比做法一，要稍微可靠一些，由於redis的decr操作直接返回真實的庫存值，所以每個請求進來，只要執行了decr操作，拿到的肯定是當前最准確的庫存值。然后進行比對，如果庫存值大於等於零，返回當前庫存值，如果小於零，則將庫存進行回滾。

此種做法，最大的一個問題就是，如果大批量的並發請求過來，redis承受的寫操作的量，相對於方法一來說，是加倍的，因為回滾庫存的存在導致的。所以這種情況下，高並發量進來，極有可能將redis的寫操作打出極限值，然后會出現很多redis寫失敗的錯誤警告。 另一個問題和做法一是一樣的，就是第五行的比對在高並發下，也是限不住的，具體的壓測結果請看我的這篇stackoverflow的提問：Will redis incr command can be limitation to specific number?

所以此種場景，雖然在高並發情況下避免了redis命令的分開操作，但是卻大大增加了redis的寫並發量，被pass。

 

超限限制做法三：先遞減庫存，然后通過整數溢出控制，最后根據情況回滾

def storage_scenario_three():
    conn = redis_conn()
    key = "storage_seckill"
    current = conn.decr(key)
    #通過整數控制溢出的做法
    if storage_overflow_checker(current):
        return current
    else:
        #回滾庫存
        conn.incr(key)
        return 0

def storage_overflow_checker(current_storage):
    #如果當前庫存未被遞減到0，則check_number為int類型，isinstance方法檢測結果為true
    #如果當前庫存已被遞減到負數，則check_number為long類型，isinstance方法檢測結果為false
    check_number = sys.maxint - current_storage
    check_result = isinstance(check_number,int)
    return check_result

說明一下，當前庫存，如果為負數，則利用python的isinstance(check_number,int)檢測的時候，check_result返回是false；如果為非負數，則檢測的時候，check_result返回的是true，上面的storage_overflow_checker的做法，和下面的C#語言的做法是一樣的，利用C#語言描述，大家可能對上面的代碼更清晰一些：

    /**
     * 通過讓Integer溢出的方式來控制數量超賣(遞減導致溢出)
     * @param current
     * @return
     */
    public boolean StorageOverFillChecker(long current) {
        try {
            //當前數值的結果計算
            Long value = Integer.MAX_VALUE - current;
            //嘗試轉變為Inter類型，如果超賣，則轉換會出錯；如果未超賣，則轉換不會出錯
            Integer.parseInt(value.toString());
        } catch (Exception ex) {
            //值溢出
            return true;
        }

        return false;
    }

可以看出，此種做法和方法二很相似，只是比對部分由，直接和零比對，變成了通過檢測integer是否溢出的方式來進行。這樣就徹底解決了高並發情況下，直接和零比對，限制不住的問題了。

雖然此種做法，相對於做法二說來，要靠譜很多，但是仍然解決不了在高並發情況下，redis寫並發量加倍的問題，極有可能某個促銷活動，在開始的那一刻，直接將redis的寫操作打出問題來。

 

超限限制做法四：共享鎖

def storage_scenario_four():
    conn = redis_conn()
    key = "storage_seckill"
    key_lock = key + "_lock"
    if conn.setnx(key_lock, "1"):
        #客戶端掛掉，設置過期時間，防止其不釋放鎖
        conn.pexpire(key_lock, 5)
        current_storage = conn.get(key)
        if int(current_storage)<=0 :
            return 0
        result = conn.decr(key)
        #客戶端正常，刪除共享鎖，提高性能
        conn.delete(key_lock)
        return result
    else :
        return "someone in it"

前面三種，由於在高並發下都有問題，所以本做法，主要是通過setnx設置共享鎖，然后請求到鎖的用戶請求，正常進行庫存扣減操作；請求不到鎖的用戶請求，則直接提示有其他人在操作庫存。

由於setnx的特殊性，當客戶端掛掉的時候，是不會釋放這個鎖的，所以當請求進來的時候，首先通過pexpire命令，為鎖設置過期時間，防止死鎖不釋放。然后執行正常的庫存扣減操作，當操作完畢，刪掉共享鎖，可以極大的提高程序性能，否則只能等待鎖慢慢過期了。

此種做法相對於上面的三種操作，通過采用共享鎖，犧牲了部分性能，來規避了高並發的問題，比較推薦，但是由於redis操作命令還是很多，並且每條都要發送到redis端執行，所以在網絡傳輸上，耗費的時間開銷是不小的。這是后面需要着力優化的方向。

 

看了上面四種做法，都不是很完美，其中最大的問題在於，高並發情況下，多條redis命令分開操作庫存，極容易發生庫存限不住的問題；同時，由於加了rollback庫存操作，極容易由於redis寫命令的操作數加倍導致壓垮redis的風險。加了鎖，雖然犧牲了部分性能，規避了高並發問題，但是redis命令操作量過多。

其實我上面一直在強調高並發，高並發。上面的四個場景，只有在高並發的情況下，才會出現問題，如果你的用戶請求量沒有那么多，那么采用上面四種方式之一，也不是不可以。但是如何才能知道采用起來沒問題呢？其實最簡單的一個方式，就是在你們自己的集群機器上，模擬活動的真實用戶量，進行壓測，看看會不會超限就行了，不超限的話，上面四種做法完全滿足需求。

 

那么，就沒有比較好一些的解決方案了嗎？

 

也不是，雖然解決這個問題，沒有絕對好用的銀彈，但是有相對好用的大蒜和聖水。下面的講解，會涉及到Redisson的Redlock的源碼實現，當然也會涉及一點lua方面的知識，還請提前預備一下。

偶然在研究分布式鎖的時候，嘗試翻閱過Redisson的Redlock的實現，並對其底層的實現方式有所記錄，我們先來看看其加鎖過程的源碼實現：

從上面的方法中，我們可以看到，分布式鎖的上鎖過程，是首先判斷一個key存不存在，如果不存在，則設置這個key，然后pexpire設置一個過期時間，來防止客戶端訪問的時候，掛掉了后，不釋放鎖的問題。為什么這段lua代碼就能實現分布式鎖的核心呢？ 原因就是，這段代碼放到一個lua腳本中，那么這段lua腳本就是一個原子性的操作。redis在執行這段lua腳本的過程中，不會摻雜任何其他的命令。所以從根本上避免了並發操作命令的問題。

我們都知道，一個key如果設置了過期時間，key過期后，redis是不會刪掉這個key的，只有用戶訪問才會刪除掉這個key，所以，當使用分布式鎖的時候，如果設置的pexpire過期時間為5ms，那么一秒鍾只能處理200個並發，性能非常低。如何解決這種性能問題呢？來看來解鎖的操作：

從上面解鎖的方法中，我們可以看到，如果這個鎖用完了之后，Redisson的做法是是直接刪除掉的。這樣可以提高不少的性能。（源碼參閱，屬於我自己的理解，如有謬誤，還請指教）

那么按照上面這種設計思路，新的超限做法就出來了。

 

超限做法五：基於lua的共享鎖

def storage_scenario_five():
    conn = redis_conn()
    key = "storage_seckill"
    key_lock = key + "_lock"
    key_val = "get_the_key"
    lua = """
            local key    = KEYS[1]
            local expire = KEYS[2]
            local value  = KEYS[3]

            local result = redis.call('setnx',key,value)
            if result == 1 then
               redis.call('pexpire', key, expire)
            end
            return result
          """
    locked = conn.eval(lua, 3, key_lock, 5, key_val)
    print (locked == 1)
    if locked == 1:
        val = storage_scenario_one()
        print("val:"+str(val))
        #刪掉共享key，用以提高性能, 否則只能默默的等其過期
        conn.delete(key_lock)
        return val
    else:
        return "someone in it"

這種做法，其實是做法四的衍生優化版本，優化的地方在於，將多條redis操作命令多次發送，改成了將多條redis操作命令放在了一個原子性操作事務中一次性執行完畢，省去了很多的網絡請求。如果可以，其實你也可以將業務邏輯糅合到上面的lua代碼中，這樣一來，性能當然會更好。

上面這種做法，如果 storage_scenario_one()這種操作是直接操作的mysql庫存，則非常推薦這種做法，但是如果storage_scenario_one()這種操作直接操作的redis中的虛擬庫存，則不是很推薦這種做法，不如直接用限流操作。

 

超限做法六： All In Lua

def storage_scenario_six():
    conn = redis_conn()
    lua = """
            local storage = redis.call('get','storage_seckill')
            if  storage ~= false then
                if tonumber(storage) > 0 then
                    return redis.call('decr','storage_seckill')
                else
                    return 'storage is zero now, can't perform decr action'
                end
            else
                return redis.call('set','storage_seckill',10)
            end
          """
    result = conn.eval(lua,0)
    print(result)

此種做法是當前最好的做法，所有的庫存扣減操作都放在lua腳本中進行，形成一個原子性操作，redis在執行上面的lua腳本的時候，是不會摻雜任何其他的執行命令的。所以這樣從根本上避免了高並發下，多條命令執行帶來的問題。而且上面的redis命令執行，都直接在redis服務器上，省去了網絡傳輸時間，也沒有共享鎖的限制，從性能上而言，是最好的。但是，業務邏輯的lua化，相對而言是比較麻煩的，所以對於追求極限庫存控制的業務，可以考慮這種做法。

 

好了，這就是我今天為大家帶來的六種庫存超限的做法，每種做法都有自己的優缺點，好使的限不住，限的住的性能不行，性能好的又需要引入lua，真心不知道如何選擇了。

 

聲明：上面六種庫存超限做法，有些屬於本人的推理，線上並未實際用過，如果你貿然使用而未經過壓測，由此造成的損失，找老板去討論吧。

參考資源：

redis lua can't work properly due to wrong type comparison

《深入理解redis》

 

Edit：2018年2月9日11:19:35

修改了第五種方法中的說明部分。