在互聯(lián)網(wǎng)企業(yè)中，限購的做法扬绪，多種多樣竖独，有的別出心裁，有的因循守舊挤牛，但是種種做法皆想達(dá)到的目的莹痢，無外乎幾種，商品賣的完，系統(tǒng)抗的住竞膳，庫存不超限航瞭。雖然短短數(shù)語，卻有著說不完坦辟，道不盡沧奴，輕者如釋重負(fù)，重者涕淚橫流的架構(gòu)體驗(yàn)长窄。 但是滔吠，在實(shí)際開發(fā)過程中，庫存超限挠日，作為其中最核心的一員疮绷，到底該怎么做，如何做才會是最合適的呢嚣潜？
今天這篇文章冬骚，我將會展示給大家?guī)齑嫦拶彽奈宸N常見的做法，并對其利弊一一探討懂算，由于這五種做法虹曙，有的在設(shè)計之初當(dāng)做提案被否定掉的箫爷，有的在線上跑著，但是在沒有任何單元測試和壓測情況下，這幾種超限控制的做法也許是不符合你的業(yè)務(wù)的贝搁，所以不建議直接用于生產(chǎn)環(huán)境。我這里權(quán)當(dāng)是做拋磚引玉倔韭，期待大家更好的做法贝奇。
工欲善其事必先利其器，在這里睡雇，我們將利用一臺測試環(huán)境的redis服務(wù)器當(dāng)做庫存超限控制的主戰(zhàn)場萌衬，先設(shè)置庫存量為10進(jìn)去，然后根據(jù)此庫存量它抱，一一展開秕豫，設(shè)置庫存代碼如下：
? 1:? def set_storage():
? 2:? ? ? conn = redis_conn()
? 3:? ? ? key = "storage_seckill"
? 4:? ? ? current_storage = conn.get(key)
? 5:? ? ? if current_storage == None:
? 6:? ? ? ? ? conn.set(key, 10)
為了方便性，我這里使用了python語言來書寫邏輯观蓄，但是今天我們只是講解思想混移，語言這類的，大家可以自己嘗試轉(zhuǎn)一下蜘腌。
上面就是我們的設(shè)置庫存到redis中的做法沫屡，很簡單，就是在redis中設(shè)置一個storage_seckill的庫存key撮珠，然后里面放上庫存量10.
超限限制做法一：先獲取當(dāng)前庫存值進(jìn)行比對沮脖，然后進(jìn)行扣減操作
? 1:? def storage_scenario_one():
? 2:? ? ? conn = redis_conn()
? 3:? ? ? key = "storage_seckill"
? 4:? ? ? current_storage = conn.get(key)
? 5:? ? ? current_storage_int = int(current_storage)
? 6:? ? ? if current_storage_int<=0 :
? 7:? ? ? ? ? return 0
? 8:? ? ? result = conn.decr(key)
? 9:? ? ? return result
首先金矛，我們拿到當(dāng)前的庫存值，然后看看是否已經(jīng)扣減到了零勺届，如果扣減到了零驶俊，則不繼續(xù)扣減，直接返回免姿；如果庫存還有饼酿，則利用decr原子操作進(jìn)行扣減，同時返回扣減后的庫存值胚膊。
此種做法在小并發(fā)量下訪問故俐，問題不大；在對庫存量控制不嚴(yán)格的業(yè)務(wù)中紊婉，問題也不大药版。但是如果并發(fā)量比較大一些，同時業(yè)務(wù)要求嚴(yán)格控制庫存喻犁，那么此種做法是非常不合適的槽片，原因在于，在高并發(fā)情況下肢础，get命令还栓，decr命令，都是分開發(fā)給redis的传轰，這樣會導(dǎo)致比對的時候剩盒，很容易出現(xiàn)限制不住的情況，也就是會造成第六行的比對失效路召。
設(shè)想如下一個場景勃刨，AB兩個請求進(jìn)來，A獲取的庫存值為1股淡，B獲取的庫存值為1，然后兩個請求都被發(fā)到redis中進(jìn)行扣減操作廷区，然后這種場景下唯灵，A最后得到的庫存值為0；但是B最后得到的庫存值為-1隙轻，超限埠帕。
所以此種場景，由于在高并發(fā)下玖绿，get和decr操作不是一組原子性操作敛瓷，會引發(fā)超限問題，被直接pass斑匪。
超限限制做法二：先扣減庫存呐籽，然后比對，最后根據(jù)情況回滾
? 1:? def storage_scenario_two():
? 2:? ? ? conn = redis_conn()
? 3:? ? ? key = "storage_seckill"
? 4:? ? ? current = conn.decr(key)
? 5:? ? ? if current>=0:
? 6:? ? ? ? ? return current
? 7:? ? ? else:
? 8:? ? ? ? ? #回滾庫存
? 9:? ? ? ? ? conn.incr(key)
? 10:? ? ? ? ? return 0
首先，請求進(jìn)來狡蝶，直接對庫存值進(jìn)行扣減庶橱，然后得到當(dāng)前的庫存值；然后贪惹，對此庫存值進(jìn)行校驗(yàn)苏章，如果庫存還有，則返回庫存值奏瞬，如果庫存沒有了枫绅，則回滾庫存，以便于防止負(fù)庫存量的存在硼端。
此做法并淋，相比做法一，要稍微可靠一些显蝌，由于redis的decr操作直接返回真實(shí)的庫存值预伺，所以每個請求進(jìn)來，只要執(zhí)行了decr操作曼尊，拿到的肯定是當(dāng)前最準(zhǔn)確的庫存值酬诀。然后進(jìn)行比對，如果庫存值大于等于零骆撇，返回當(dāng)前庫存值瞒御，如果小于零，則將庫存進(jìn)行回滾神郊。
此種做法肴裙，最大的一個問題就是，如果大批量的并發(fā)請求過來涌乳，redis承受的寫操作的量蜻懦，相對于方法一來說，是加倍的夕晓，因?yàn)榛貪L庫存的存在導(dǎo)致的宛乃。所以這種情況下，高并發(fā)量進(jìn)來蒸辆，極有可能將redis的寫操作打出極限值征炼，然后會出現(xiàn)很多redis寫失敗的錯誤警告。 另一個問題和做法一是一樣的躬贡，就是第五行的比對在高并發(fā)下谆奥，也是限不住的，具體的壓測結(jié)果請看我的這篇stackoverflow的提問：Will redis incr command can be limitation to specific number?
所以此種場景拂玻，雖然在高并發(fā)情況下避免了redis命令的分開操作酸些，但是卻大大增加了redis的寫并發(fā)量宰译，被pass。
超限限制做法三：先遞減庫存擂仍，然后通過整數(shù)溢出控制囤屹，最后根據(jù)情況回滾
? 1:? def storage_scenario_three():
? 2:? ? ? conn = redis_conn()
? 3:? ? ? key = "storage_seckill"
? 4:? ? ? current = conn.decr(key)
? 5:? ? ? #通過整數(shù)控制溢出的做法
? 6:? ? ? if storage_overflow_checker(current):
? 7:? ? ? ? ? return current
? 8:? ? ? else:
? 9:? ? ? ? ? #回滾庫存
? 10:? ? ? ? ? conn.incr(key)
? 11:? ? ? ? ? return 0
? 12:? ?
? 13:? def storage_overflow_checker(current_storage):
? 14:? ? ? #如果當(dāng)前庫存未被遞減到0，則check_number為int類型逢渔，isinstance方法檢測結(jié)果為true
? 15:? ? ? #如果當(dāng)前庫存已被遞減到負(fù)數(shù)肋坚，則check_number為long類型，isinstance方法檢測結(jié)果為false
? 16:? ? ? check_number = sys.maxint - current_storage
? 17:? ? ? check_result = isinstance(check_number,int)
? 18:? ? ? return check_result
說明一下肃廓，當(dāng)前庫存智厌，如果為負(fù)數(shù)，則利用python的isinstance(check_number,int)檢測的時候盲赊，check_result返回是false铣鹏；如果為非負(fù)數(shù)，則檢測的時候哀蘑，check_result返回的是true诚卸，上面的storage_overflow_checker的做法，和下面的C#語言的做法是一樣的绘迁，利用C#語言描述合溺，大家可能對上面的代碼更清晰一些：
? 1:? ? ? /**
? 2:? ? ? * 通過讓Integer溢出的方式來控制數(shù)量超賣(遞減導(dǎo)致溢出)
? 3:? ? ? * @param current
? 4:? ? ? * @return
? 5:? ? ? */
? 6:? ? ? public boolean StorageOverFillChecker(long current) {
? 7:? ? ? ? ? try {
? 8:? ? ? ? ? ? ? //當(dāng)前數(shù)值的結(jié)果計算
? 9:? ? ? ? ? ? ? Long value = Integer.MAX_VALUE - current;
? 10:? ? ? ? ? ? ? //嘗試轉(zhuǎn)變?yōu)镮nter類型，如果超賣缀台，則轉(zhuǎn)換會出錯棠赛；如果未超賣，則轉(zhuǎn)換不會出錯
? 11:? ? ? ? ? ? ? Integer.parseInt(value.toString());
? 12:? ? ? ? ? } catch (Exception ex) {
? 13:? ? ? ? ? ? ? //值溢出
? 14:? ? ? ? ? ? ? return true;
? 15:? ? ? ? ? }
? 16:? ?
? 17:? ? ? ? ? return false;
? 18:? ? ? }
可以看出膛腐，此種做法和方法二很相似睛约，只是比對部分由，直接和零比對哲身，變成了通過檢測integer是否溢出的方式來進(jìn)行辩涝。這樣就徹底解決了高并發(fā)情況下，直接和零比對勘天，限制不住的問題了膀值。
雖然此種做法，相對于做法二說來误辑，要靠譜很多，但是仍然解決不了在高并發(fā)情況下歌逢，redis寫并發(fā)量加倍的問題巾钉，極有可能某個促銷活動，在開始的那一刻秘案，直接將redis的寫操作打出問題來砰苍。
超限限制做法四：共享鎖
? 1:? def storage_scenario_four():
? 2:? ? ? conn = redis_conn()
? 3:? ? ? key = "storage_seckill"
? 4:? ? ? key_lock = key + "_lock"
? 5:? ? ? if conn.setnx(key_lock, "1"):
? 6:? ? ? ? ? #客戶端掛掉潦匈，設(shè)置過期時間，防止其不釋放鎖
? 7:? ? ? ? ? conn.pexpire(key_lock, 5)
? 8:? ? ? ? ? current_storage = conn.get(key)
? 9:? ? ? ? ? if int(current_storage)<=0 :
? 10:? ? ? ? ? ? ? return 0
? 11:? ? ? ? ? result = conn.decr(key)
? 12:? ? ? ? ? #客戶端正常赚导，刪除共享鎖茬缩，提高性能
? 13:? ? ? ? ? conn.delete(key_lock)
? 14:? ? ? ? ? return result
? 15:? ? ? else :
? 16:? ? ? ? ? return "someone in it"
前面三種，由于在高并發(fā)下都有問題吼旧，所以本做法凰锡，主要是通過setnx設(shè)置共享鎖，然后請求到鎖的用戶請求圈暗，正常進(jìn)行庫存扣減操作掂为；請求不到鎖的用戶請求，則直接提示有其他人在操作庫存员串。
由于setnx的特殊性勇哗，當(dāng)客戶端掛掉的時候，是不會釋放這個鎖的寸齐，所以當(dāng)請求進(jìn)來的時候欲诺，首先通過pexpire命令，為鎖設(shè)置過期時間渺鹦，防止死鎖不釋放扰法。然后執(zhí)行正常的庫存扣減操作，當(dāng)操作完畢海铆，刪掉共享鎖迹恐，可以極大的提高程序性能，否則只能等待鎖慢慢過期了卧斟。
此種做法相對于上面的三種操作殴边，通過采用共享鎖，犧牲了部分性能珍语，來規(guī)避了高并發(fā)的問題锤岸，比較推薦，但是由于redis操作命令還是很多板乙，并且每條都要發(fā)送到redis端執(zhí)行是偷，所以在網(wǎng)絡(luò)傳輸上，耗費(fèi)的時間開銷是不小的募逞。這是后面需要著力優(yōu)化的方向蛋铆。
看了上面四種做法，都不是很完美放接，其中最大的問題在于刺啦，高并發(fā)情況下，多條redis命令分開操作庫存纠脾，極容易發(fā)生庫存限不住的問題玛瘸；同時蜕青，由于加了rollback庫存操作，極容易由于redis寫命令的操作數(shù)加倍導(dǎo)致壓垮redis的風(fēng)險糊渊。加了鎖右核，雖然犧牲了部分性能，規(guī)避了高并發(fā)問題渺绒，但是redis命令操作量過多贺喝。
其實(shí)我上面一直在強(qiáng)調(diào)高并發(fā)，高并發(fā)芒篷。上面的四個場景搜变，只有在高并發(fā)的情況下，才會出現(xiàn)問題针炉，如果你的用戶請求量沒有那么多挠他，那么采用上面四種方式之一，也不是不可以篡帕。但是如何才能知道采用起來沒問題呢殖侵？其實(shí)最簡單的一個方式，就是在你們自己的集群機(jī)器上镰烧，模擬活動的真實(shí)用戶量拢军，進(jìn)行壓測，看看會不會超限就行了怔鳖，不超限的話茉唉，上面四種做法完全滿足需求。
那么结执，就沒有比較好一些的解決方案了嗎度陆？
也不是，雖然解決這個問題献幔，沒有絕對好用的銀彈懂傀，但是有相對好用的大蒜和圣水。下面的講解蜡感，會涉及到Redisson的Redlock的源碼實(shí)現(xiàn)蹬蚁，當(dāng)然也會涉及一點(diǎn)lua方面的知識，還請?zhí)崆邦A(yù)備一下郑兴。
偶然在研究分布式鎖的時候犀斋，嘗試翻閱過Redisson的Redlock的實(shí)現(xiàn)，并對其底層的實(shí)現(xiàn)方式有所記錄情连，我們先來看看其加鎖過程的源碼實(shí)現(xiàn)：
從上面的方法中闪水，我們可以看到，分布式鎖的上鎖過程，是首先判斷一個key存不存在球榆，如果不存在，則設(shè)置這個key禁筏，然后pexpire設(shè)置一個過期時間持钉，來防止客戶端訪問的時候，掛掉了后篱昔，不釋放鎖的問題每强。為什么這段lua代碼就能實(shí)現(xiàn)分布式鎖的核心呢？ 原因就是州刽，這段代碼放到一個lua腳本中空执，那么這段lua腳本就是一個原子性的操作。redis在執(zhí)行這段lua腳本的過程中穗椅，不會摻雜任何其他的命令辨绊。所以從根本上避免了并發(fā)操作命令的問題。
我們都知道匹表，一個key如果設(shè)置了過期時間门坷，key過期后，redis是不會刪掉這個key的袍镀，只有用戶訪問才會刪除掉這個key默蚌，所以，當(dāng)使用分布式鎖的時候苇羡，如果設(shè)置的pexpire過期時間為5ms绸吸，那么一秒鐘只能處理200個并發(fā)，性能非常低设江。如何解決這種性能問題呢锦茁？來看來解鎖的操作：
從上面解鎖的方法中，我們可以看到绣硝，如果這個鎖用完了之后蜻势，Redisson的做法是是直接刪除掉的。這樣可以提高不少的性能鹉胖。（源碼參閱握玛，屬于我自己的理解，如有謬誤甫菠，還請指教）
那么按照上面這種設(shè)計思路挠铲，新的超限做法就出來了。
超限做法五：基于lua的共享鎖
? 1:? def storage_scenario_five():
? 2:? ? ? conn = redis_conn()
? 3:? ? ? key = "storage_seckill"
? 4:? ? ? key_lock = key + "_lock"
? 5:? ? ? key_val = "get_the_key"
? 6:? ? ? lua = """
? 7:? ? ? ? ? ? ? local key? ? = KEYS[1]
? 8:? ? ? ? ? ? ? local expire = KEYS[2]
? 9:? ? ? ? ? ? ? local value? = KEYS[3]
? 10:? ?
? 11:? ? ? ? ? ? ? local result = redis.call('setnx',key,value)
? 12:? ? ? ? ? ? ? if result == 1 then
? 13:? ? ? ? ? ? ? ? redis.call('pexpire', key, expire)
? 14:? ? ? ? ? ? ? end
? 15:? ? ? ? ? ? ? return result
? 16:? ? ? ? ? ? """
? 17:? ? ? locked = conn.eval(lua, 3, key_lock, 5, key_val)
? 18:? ? ? print (locked == 1)
? 19:? ? ? if locked == 1:
? 20:? ? ? ? ? val = storage_scenario_one()
? 21:? ? ? ? ? print("val:"+str(val))
? 22:? ? ? ? ? #刪掉共享key寂诱，用以提高性能, 否則只能默默的等其過期
? 23:? ? ? ? ? conn.delete(key_lock)
? 24:? ? ? ? ? return val
? 25:? ? ? else:
? 26:? ? ? ? ? return "someone in it"
這種做法拂苹，其實(shí)是做法四的衍生優(yōu)化版本，優(yōu)化的地方在于痰洒，將多條redis操作命令多次發(fā)送瓢棒，改成了將多條redis操作命令放在了一個原子性操作事務(wù)中一次性執(zhí)行完畢浴韭，省去了很多的網(wǎng)絡(luò)請求。如果可以脯宿，其實(shí)你也可以將業(yè)務(wù)邏輯糅合到上面的lua代碼中念颈，這樣一來，性能當(dāng)然會更好连霉。
上面這種做法榴芳，如果?storage_scenario_one()這種操作是直接操作的mysql庫存，則非常推薦這種做法跺撼，但是如果storage_scenario_one()這種操作直接操作的redis中的虛擬庫存窟感，則不是很推薦這種做法，不如直接用限流操作歉井。
超限做法六： All In Lua
? 1:? def storage_scenario_six():
? 2:? ? ? conn = redis_conn()
? 3:? ? ? lua = """
? 4:? ? ? ? ? ? ? local storage = redis.call('get','storage_seckill')
? 5:? ? ? ? ? ? ? if? storage ~= false then
? 6:? ? ? ? ? ? ? ? ? if tonumber(storage) > 0 then
? 7:? ? ? ? ? ? ? ? ? ? ? return redis.call('decr','storage_seckill')
? 8:? ? ? ? ? ? ? ? ? else
? 9:? ? ? ? ? ? ? ? ? ? ? return 'storage is zero now, can't perform decr action'
? 10:? ? ? ? ? ? ? ? ? end
? 11:? ? ? ? ? ? ? else
? 12:? ? ? ? ? ? ? ? ? return redis.call('set','storage_seckill',10)
? 13:? ? ? ? ? ? ? end
? 14:? ? ? ? ? ? """
? 15:? ? ? result = conn.eval(lua,0)
? 16:? ? ? print(result)
此種做法是當(dāng)前最好的做法柿祈，所有的庫存扣減操作都放在lua腳本中進(jìn)行，形成一個原子性操作酣难，redis在執(zhí)行上面的lua腳本的時候谍夭，是不會摻雜任何其他的執(zhí)行命令的。所以這樣從根本上避免了高并發(fā)下憨募，多條命令執(zhí)行帶來的問題紧索。而且上面的redis命令執(zhí)行，都直接在redis服務(wù)器上菜谣，省去了網(wǎng)絡(luò)傳輸時間珠漂，也沒有共享鎖的限制，從性能上而言尾膊，是最好的媳危。但是，業(yè)務(wù)邏輯的lua化冈敛，相對而言是比較麻煩的待笑，所以對于追求極限庫存控制的業(yè)務(wù)，可以考慮這種做法抓谴。
好了暮蹂，這就是我今天為大家?guī)淼牧N庫存超限的做法，每種做法都有自己的優(yōu)缺點(diǎn)癌压，好使的限不住仰泻，限的住的性能不行，性能好的又需要引入lua滩届，真心不知道如何選擇了集侯。
聲明：上面六種庫存超限做法，有些屬于本人的推理，線上并未實(shí)際用過棠枉，如果你貿(mào)然使用而未經(jīng)過壓測浓体，由此造成的損失，找老板去討論吧术健。
歡迎工作一到五年的Java工程師朋友們加入Java程序員開發(fā)： 854393687
群內(nèi)提供免費(fèi)的Java架構(gòu)學(xué)習(xí)資料（里面有高可用汹碱、高并發(fā)、高性能及分布式荞估、Jvm性能調(diào)優(yōu)、Spring源碼稚新，MyBatis勘伺，Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多個知識點(diǎn)的架構(gòu)資料）合理利用自己每一分每一秒的時間來學(xué)習(xí)提升自己，不要再用"沒有時間“來掩飾自己思想上的懶惰褂删！趁年輕飞醉，使勁拼，給未來的自己一個交代屯阀！
