一. 單體架構和垂直應用架構

很久以前，還在我還年輕的時候大部分包括剛開始的淘寶，還不是亞馬遜中國的卓越等網站，那時候還沒有多少知道上網，大部分年輕人還停留在cs局域網對戰的時候，大多國人由于不太清楚上網這個概念時候那時候國內興起了一系列的互聯網公司（活下來的大部分成了行業巨頭）那時候流量很小，只需要一個基本應用，將所有的功能都部署在一起（前端和后端都放在一起），用來減少部署節點和成本。那時候，主要關注點就是對業務的增刪改查工作
特點：

• 所有的業務代碼都放在一起。
• 通過部署應用集群和數據庫集群來提高系統的性能。

優點：

• 項目架構簡單，開發成本低，周期短一人一馬搞定。

缺點：

• 全部代碼放在一個項目中耦合嚴重，對于往后發展不適合。
• 修改代碼需要人力，還要去學習前任的業務邏輯實現，成本高。
• 容錯率低。

垂直架構
當訪問量逐漸增大，功能逐漸復雜起來，單一應用架構就顯得有些捉襟見肘，由于所有的功能都寫在同一個工程中，整個工程會越來越龐大越來越臃腫，所以將應用拆成互不相干的幾個應用，以提升效率。

特點：

• 將項目以單一應用架構的方式，將一個大應用拆分成為幾個互不干擾的應用。
• 應用于應用之間互不相干，功能和數據都會存在冗余。

優點：

• 通過垂直拆分，防止單體項目無限擴大。
• 系統間相互獨立。

缺點：

• 項目拆分之后，項目與項目之間存在數據冗余，耦合性較大。
• 提高系統性能只能通過擴展集群，成本高，并且存在瓶頸。

上述其實就是多應用相結合，但是彼此之間應用無關。

二. 微服務的出現

當垂直應用越來越多，應用與應用之間的交互不可避免，這時需要將核心業務抽取出來，作為獨立的服務，逐漸形成穩定的服務，使前端應用能更快速的響應多變的市場需求。
特點：

• 系統服務層完全獨立，并且抽取成一個一個的獨立服務。
• 微服務遵循單一原則。
• 使用服務中心進行服務注冊與發現。
• 每個服務有自己的獨立數據源。
• 微服務之間采用 RESTful 等輕量協議傳輸。

優點：

• 通過分解巨大單體式應用為多個服務方法解決了復雜性問題。
• 可以更加精準的制定每個服務的優化方案，提高系統可維護性。
• 微服務架構模式是每個微服務獨立的部署，可以使每個服務獨立擴展。
• 微服務架構采用去中心化思想，服務之間采用 RESTful 等輕量協議通信。

缺點：

• 微服務過多，服務治理成本高，對系統運維團隊挑戰大。
• 分布式系統開發的技術成本高（容錯、分布式事務等），對團隊挑戰大。

三. 滿足人們日益增長的物質文化需要

說完微服務，并不是一蹴而就為用而用，如果只是一個非常小的系統，或者只是一個簡單的應用程序，后續不會再進行過多的提升或者是版本是直接替換性質的（比如曾經的某駕游，1.0-2.0直接就是替換了所有內容），這個一般都是適用于剛開始創業的小微公司，不會開始就直接上大型分布式架構系統，一方面考慮快速上架，后續迭代開發；另外一方面也是最重要的就是沒那么多錢！—體會

那成熟的軟件公司尤其是大中型企業他們本身已經經歷過洗禮，比如淘寶從開始的PHP—現在的自研等。所以已經有了穩固的和強大的（資本）流量和目標。為此這類型公司早就在行業內抓住了前瞻技術，尤其是業務復雜度和內容多樣性的出現，使得他們對待系統穩定/安全/數據一致性等方面都有著很強悍的經驗。同樣隨著系統的不斷擴展和內容，流量的不斷飆升（甚至可以達到億級），比如雙11，雙12，年中大促，各類所謂的節等等，如果只是單體或者簡單的微服務架構是肯定不行的，還需要各類所謂的機制保證和相關服務的支撐–人力，硬件，運維等等。

總結一下，微服務雖然可以通過ddd（領域設計）來實現一套完整的系統，但是還需要其他配合，比如人力/相關服務等配合完成。

本次內容不談那么大而深的概念和實現，而是我們討論一下在這個大型分布式環境下的一個肯定會用到的內容—分布式鎖的應用和實現。

四. 分布式鎖

什么是分布式鎖？

如果在一個分布式系統中，我們從數據庫中讀取一個數據，然后修改保存，這種情況很容易遇到并發問題。因為讀取和更新保存不是一個原子操作，在并發時就會導致數據的不正確。說白一點就是一次操作有讀又有寫，且讀和寫必須保證是當前這一次操作的，不能被其他情況給挾持。

這種場景其實并不少見，比如電商秒殺活動，庫存數量的更新就會遇到。如果是單機應用，線程鎖就可以避免。如果是分布式應用，不同服務器不同容器對應的應用都是不同的jvm，所以無法通過線程來實現，這時就需要引入分布式鎖來解決。—提個話題，大部分情況下想要保證分布式系統的cap，基本都依賴于中間件。

由此可見分布式鎖的目的其實很簡單，就是為了保證多臺服務器在執行某一段代碼時保證只有一臺服務器執行。

為了保證分布式鎖的可用性，至少要確保鎖的實現要同時滿足以下幾點：

• 互斥性。在任何時刻，保證只有一個客戶端持有鎖。
• 不能出現死鎖。如果在一個客戶端持有鎖的期間，這個客戶端崩潰了，也要保證后續的其他客戶端可以上鎖。
• 保證上鎖和解鎖都是同一個客戶端。

一般來說，實現分布式鎖的方式有以下幾種：

• 使用MySQL，基于唯一索引。
• 使用Redis，基于setnx命令。
• 使用ZooKeeper，基于臨時有序節點。

五. 實現方案

第一個數據庫操作

可以針對一列做唯一索引，或者做一張表，去做對應更新記錄；或者做樂觀鎖加version.

但是這種方案是否合適？！！！！！

問題很多，首先最明顯的問題就是io問題，其次就是連接耗時問題。

可是為了解決響應時間和分布式環境下的數據一致性問題。

redis實現

Redis實現分布式鎖主要利用Redis的setnx命令和Redis的單線程特性（由來以久的到底多還是單的問題，其實這個問題只是這對于處理能力進行細化，比如在6.x之前所有的操作都是由一個單工作線程來實現，對于一個數據的操作包括讀/計算/寫入等都由該單工作線程實現；而6.x之后分出一個io的子線程，把運算放入了工作線程操作，二讀和寫入兩個操作放入了獨立的io子線程操作；那如果多個數據操作，實際上就會出現一個單工作線程，多個io子線程，構成io復用—性能更高）。setnx是SET if not exists(如果不存在，則 SET)的簡寫。

上述是加了過期時間的，如果不加過期時間直接用setnx key:value。

為什么要使用nx這種模式，實際為了解決一個問題，就是我們如果定義了一個值，在其沒有被手動刪除或者時間戳自動失效前都不能被其他線程給操作和修改。明白了嗎？是不是這樣就可以保證在針對于這一個操作的時候可以保證數據的原子性？！

為了保證實現的最基本要求，本案例啟動了2個服務，分別是8081，8082，通過用Nginx做了網關，負載使用了輪詢。模擬訪問地址是Nginx網關地址。庫存模擬放入redis中 stock:1000

并發測試工具采用postman自帶測試（效果也就玩玩）

以及Jmeters5.3.

數據模擬采用了redis或者數據庫。

方案1:

public String updateStock01(String pId) throws Exception {String clientId = UUID.randomUUID().toString();try{Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(pId,clientId,30, TimeUnit.SECONDS);if(result) {int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if (stock > 0) {stock = stock - 1;stringRedisTemplate.opsForValue().set("stock", stock + "");log.info("扣除成功，目前庫存還有" + stock + "件");} else {log.info("庫存不足");}}}catch (Exception e){log.info(e.getMessage());}finally {if(clientId.equals(stringRedisTemplate.opsForValue().get(pId))) {stringRedisTemplate.delete(pId);}}return "success"; }

上述代碼可以看到我們通過生成一個30秒ttl的Key -> pId value->clientId的列子。在這個30秒的時間里面，任何一個同樣的Key都不可以進來修改。然后通過操縱對于庫存結果的扣件后，在最終finally里面保證刪除該key留給其他線程。看起來沒問題，但是真沒有問題嗎？

如果流量不大，其實出現問題的情況基本為0.但是如果說在壓測時候使用j meters通過大量線程同時進入，并保持不停壓測。會發現—涼涼！

我這邊操作2個相同服務去實現，如下圖：

發現結果有問題了！這個問題究其原因是怎么造成的呢？

首先我們這個是2個服務，都是依賴于獨立的jvm，所以有不同的Jmm,就算你做了線程加了重量級鎖結果還是一樣，畢竟是非原子性的操作。

其次雖然采用了set key value [EX seconds] [PX milliseconds] [NX|XX]的方式，但是如果一個線程a執行的時間較長沒有來得及釋放，鎖就過期了，此時l另外一個線程B是可以獲取到鎖的。當線程A執行完成之后，釋放鎖，實際上就把線程B的鎖釋放掉了。這個時候，再來一個線程C又是可以獲取到鎖的，而此時如果線程B執行完釋放鎖實際上就是釋放的線程C設置的鎖。

這邊只是本地機器，如果是云端/線上環境等，服務和服務之間的調度延遲，網絡訪問延遲等等都會造成問題。

方案1不可行！！！

那上述問題，我們如何規避呢？

可以這樣想，既然代碼層面直接實現不可以，但是可以通過交給redis這個所謂的單線程去處理不就行了！？

方案2:

set、del是一一映射的，不會出現把其他現成的鎖del的情況。從實際情況的角度來看，即使能做到set、del一一映射，也無法保障業務的絕對安全。因為鎖的過期時間始終是有界的，除非不設置過期時間或者把過期時間設置的很長，但這樣做也會帶來其他問題。故沒有意義。要想實現相對安全的分布式鎖，必須依賴key的value值。在釋放鎖的時候，通過value值的唯一性來保證不會勿刪。

通過lua腳本來實現；

通過redission實現：

public String updateStock02(String pId) throws Exception {//增加分布式鎖RLock rLock =redisson.getLock(pId);try{rLock.lock(5000,TimeUnit.MILLISECONDS);int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if(stock>0){stock = stock - 1;stringRedisTemplate.opsForValue().set("stock",stock+"");log.info("扣除成功，目前庫存還有"+stock+"件");}else{log.info("庫存不足");}}catch (Exception e){log.info(e.getMessage());}finally {rLock.unlock();}return "success"; }

按照原來還有982個，現在做1000個線程，最快執行完畢，連續循環2次，結果到0的時候，顯示庫存不足。

看代碼：

RLock rLock =redisson.getLock(pId);rLock.lock(5000,TimeUnit.MILLISECONDS);rLock.unlock();

眼熟嗎？是否和Lock很像？！

其實這邊和Lock相似的地方采用類似自旋鎖的方式，做了do while循環判斷是否獲得鎖，然后根據hash算法分配到不同的redis 哨兵主從環境（關于16384的槽以及丟失某個節點后數據問題后續可以討論），當然單機也沒有問題就是。并執行Lua腳本。

"if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('hset', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "return redis.call('pttl', KEYS[1]);"

其中hash保證通過lua腳本保證了原子性。

具體內容用了一段別人介紹的說法(不去深究，可以結合線程鎖的源碼來閱讀)：

原引：csdn 段子猿 給大家解釋一下，第一段if判斷語句，就是用“exists myLock”命令判斷一下，如果要加鎖的那個鎖key不存在的話，就進行加鎖。 如何加鎖呢？很簡單，用下面的命令：hset myLock 3d7b5418-a86d-48c5-ae15-7fe13ef0034c:1接著會執行“pexpire myLock 30000”命令，設置myLock這個鎖key的生存時間是30秒。這樣加鎖完成了。幾個參數所代表的意思：KEYS[1]代表的是你加鎖的那個key，比如說 RLock lock = redisson.getLock("myLock")，這里你自己設置了加鎖的那個鎖key就是“myLock”。ARGV[1]代表的就是鎖key的默認生存時間，默認30秒。ARGV[2]代表的是加鎖的客戶端的ID，類似于這樣： 3d7b5418-a86d-48c5-ae15-7fe13ef0034c:1 2.鎖互斥機制在這個時候，如果客戶端2來嘗試加鎖，執行了同樣的一段lua腳本，會咋樣呢？按照腳本流程，第一個if判斷會執行“exists myLock”，發現myLock這個鎖key已經存在了。接著第二個if判斷，判斷一下，myLock鎖key的hash數據結構中，是否包含客戶端2的ID，但是明顯客戶端id不一致。所以，客戶端2會獲取到pttl myLock返回的一個數字，這個數字代表了myLock這個鎖key的剩余生存時間。比如還剩15000毫秒的生存時間， 此時客戶端2會進入一個while循環，不停的嘗試加鎖。3.watch dog自動延期機制客戶端加鎖的key默認生存時間是30秒，如果超過了30秒，客戶端還想一直持有這把鎖，怎么辦呢？只要客戶端一旦加鎖成功，就會啟動一個watch dog看門狗，他是一個后臺線程，每到過期時間1/3（默認生存時間是30s的話，就是10s）就去重新刷一次，如果客戶端還持有鎖key，那么就會不斷的延長鎖key的生存時間，如果key不存在則停止刷新。4.可重入加鎖機制那如果客戶端已經持有了這把鎖，結果可重入的加鎖會怎么樣呢？比如下面這種代碼：RLock lock = redisson.getLock("myLock"); lock.lock(); //do something lock.lock(); //do something lock.unlock(); lock.unlock(); 這時我們來分析一下上面那段lua腳本。第一個if判斷肯定不成立，“exists myLock”會顯示鎖key已經存在了。第二個if判斷會成立，因為myLock的hash數據結構中包含的那個ID，就是客戶端那個ID，也就是“3d7b5418-a86d-48c5-ae15-7fe13ef0034c:1” ，此時就會執行可重入加鎖的邏輯，他會用： incrby myLock 3d7b5418-a86d-48c5-ae15-7fe13ef0034c:1 1通過這個命令，對客戶端1的加鎖次數，累加1。此時myLock數據結構變為下面這樣： 大家看到了吧，那個myLock的hash數據結構中的那個客戶端ID，就對應著加鎖的次數5.釋放鎖機制如果執行lock.unlock()，就可以釋放分布式鎖，此時的業務邏輯其實很簡單。就是每次都對myLock數據結構中的那個加鎖次數減1。如果發現加鎖次數是0了，說明這個客戶端已經不再持有鎖了，此時就會用： “del myLock”命令，從redis里刪除這個key。然后呢，另外的客戶端就可以嘗試完成加鎖了。這就是所謂的分布式鎖的開源Redisson框架的實現機制。

方案3: Spring Integration

Spring Integration不需要你去關注它到底是基于什么存儲技術實現的，它是面向接口編程，低耦合讓你不需要關注底層實現。你要做的僅僅是做簡單的選擇，然后用相同的一套api即可完成分布式鎖的操作。

@Override public String updateStock03(String pId) throws Exception {//增加分布式鎖Lock lock = redisLockRegistry.obtain(pId);boolean flag = false;try{flag = lock.tryLock(10,TimeUnit.SECONDS);int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if(stock>0){stock = stock - 1;stringRedisTemplate.opsForValue().set("stock",stock+"");log.info("購買成功，庫存還有"+stock+"件");}else{log.info("庫存不足");}}catch (Exception e){log.info(e.getMessage());}finally {lock.unlock();}return "success"; }

通過測試
共計100個。

測試成功。

zookeeper實現

通過zk來實現，其中zk通過定義EPHEMERAL_SEQUENTIAL臨時順序節點.比如創建一個/lock/臨時有序；

創建節點成功后，獲取/lock目錄下的所有臨時節點，再判斷當前線程創建的節點是否是所有的節點的序號最小的節點

如果當前線程創建的節點是所有節點序號最小的節點，則認為獲取鎖成功。

如果當前線程創建的節點不是所有節點序號最小的節點，則對節點序號的前一個節點添加一個事件監聽。
比如當前線程獲取到的節點序號為/lock/003,然后所有的節點列表為[/lock/001,/lock/002,/lock/003],則對/lock/002這個節點添加一個事件監聽器。

如果鎖被釋放，會喚醒下一個序號的節點，然后重新執行第3步，判斷是否自己的節點序號是最小。比如/lock/001釋放了，/lock/002監聽到時間，此時節點集合為[/lock/002,/lock/003],則/lock/002為最小序號節點，獲取到鎖。

偷懶的寫法—上面寫法實話太麻煩了：

Curator是一個zookeeper的開源客戶端，也提供了分布式鎖的實現：

public String updateStock04(String pId) throws Exception {InterProcessMutex interProcessMutex = new InterProcessMutex(curatorFramework, "/product_" + pId);try{//加鎖interProcessMutex.acquire(10,TimeUnit.SECONDS);int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));if(stock>0){stock = stock - 1;stringRedisTemplate.opsForValue().set("stock",stock+"");log.info("購買成功，庫存還有"+stock+"件");}else{log.info("------沒有庫存------");}}catch (Exception e){e.printStackTrace();}finally {interProcessMutex.release();}return "success";}

六. 總結

上述內容就是目前實現分布式鎖的幾種流行方式，那還有沒有其他的？有！比如consul也可以實現，這邊也不延展。這邊要說的是如果業務如果并沒有想象的那么夸張的時候，沒有必要不要考慮使用第三方實現。

另外Redission實際上就是將并行的請求，轉化為串行請求。這樣就降低了并發的響應速度.

可以通過鎖分段來實現，比如ps5昨天發布國行價格，商品只有2000臺貨品，暫時不考慮地區庫存計算；一下黃牛們紛涌而至，這個時候2000臺貨如果按照串行走的話，比如每個50ms,暫時不考慮分布式事務鎖或者是最終一致性的死信隊列情況，2000個大概需要2000x50ms=100000ms,大概也要100秒左右時間，那如果并行呢？比如我把2000個ps5分成20份，那就是每份里面存100個，按照庫存取余數（或者Hash或者就100，100的放等等）分別放置。這樣每次可以定義多個鎖進行加鎖即可。這邊注意一個問題就是因為涉及到20個分段，所以注意如果某個分段已經沒有庫存的時候，需要解鎖進入下一個分段繼續。

也可以直接在redis預設好（個人比較推崇），我可以給ps5的2000臺分成不同的stock:product_ps5_xxx:01: 100;stock:product_ps5_xxx:02: 100…

最后無論你身處一個什么樣的公司，最開始的工作可能都需要從最簡單的做起。所以能根據自己公司業務場景，選擇適合自己項目的方案。

總結

以上是生活随笔為你收集整理的高并发分布式场景下的应用---分布式锁的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。