2、分布式鎖實現原理與最佳實踐（二）

常見分布式鎖的原理

4.1 Redisson

Redis 2.6之后才可以執行lua腳本，比起管道而言，這是原子性的，模擬一個商品減庫存的原子操作：

//lua腳本命令執行方式：redis-cli --eval /tmp/test.lua , 10
jedis.set("product_stock_10016", "15");  
//初始化商品10016的庫存
String script = " local count = redis.call('get', KEYS[1]) " +" local a = tonumber(count) " +" local b = tonumber(ARGV[1]) " +" if a >= b then " +"   redis.call('set', KEYS[1], a-b) " +"   return 1 " +" end " +" return 0 ";
Object obj = jedis.eval(script, Arrays.asList("product_stock_10016"), Arrays.asList("10"));
System.out.println(obj);

在這里插入圖片描述
4.1.1 嘗試加鎖的邏輯

上面的org.redisson.RedissonLock#lock()通過調用自己方法內部的lock方法的org.redisson.RedissonLock#tryAcquire方法。之后調用 org.redisson.RedissonLock#tryAcquireAsync：

首先調用內部的org.redisson.RedissonLock#tryLockInnerAsync：設置對應的分布式鎖
在這里插入圖片描述
到這里獲取鎖的邏輯就結束了，如果這里沒有獲取到，在Future的回調里面就會直接return，會在外層有一個while true的循環，訂閱釋放鎖的消息準備被喚醒。如果說加鎖成功，就開始執行鎖續命邏輯。

4.1.2 鎖續命邏輯
lua腳本最后是以毫秒為單位返回key的剩余過期時間。成功加鎖之后org.redisson.RedissonLock#scheduleExpirationRenewal中將會調用org.redisson.RedissonLock#renewExpiration，這個方法內部就有鎖續命的邏輯，是一個定時任務，等10s執行。
執行的時候嘗試執行的續命邏輯使用的是Lua腳本，當前的鎖有值，就續命，沒有就直接返回0：
在這里插入圖片描述
返回0之后外層會判斷，延時成功就會再次調用自己，否則延時調用結束，不再為當前的鎖續命。所以這里的續命不是一個真正的定時，而是循環調用自己的延時任務。

4.1.3 循環間隔搶鎖機制
如果一開始就加鎖成功就直接返回。
如果一開始加鎖失敗，沒搶到鎖的線程就會在while循環中嘗試加鎖，加鎖成功就結束循環，否則等待當前鎖的超時時間之后再次嘗試加鎖。所以實現邏輯默認是非公平鎖：
在這里插入圖片描述
里面有一個subscribe的邏輯，會監聽對應加鎖的key，當鎖釋放之后publish對應的消息，此時如果沒有到達對應的鎖的超時時間，也會嘗試獲取鎖，避免時間浪費。
4.1.4 釋放鎖和喚醒其他線程的邏輯
前面沒有搶到鎖的線程會監聽對應的queue，后面搶到鎖的線程釋放鎖的時候會發送一個消息。
在這里插入圖片描述
訂閱的時候指定收到消息時候的邏輯：會喚醒阻塞之后執行while循環

4.1.5 重入鎖的邏輯
存在對應的鎖，就對對應的hash結構的value直接+1，和Java重入鎖的邏輯是一致的。
在這里插入圖片描述

4.2 RedLock解決非單體項目的Redis主從架構的鎖失效

https://redis.io/docs/manual/patterns/distributed-locks/
查看Redis官方文檔，對于單節點的Redis ，使用setnx和lua del刪除分布式鎖是足夠的，但是主從架構的場景下：鎖先加在一個master節點上，默認是異步同步到從節點，此時master掛了會選擇slave為master，此時又可以加鎖，就會導致超賣。但是如果使用zookeeper來實現的話，由于zk是CP的，所以CP不存在這樣的問題。
Redis文檔中給出了RedLock的解決辦法，使用redLock真的可以解決嗎？
4.2.1 RedLock 原理
基于客戶端的實現，是基于多個獨立的Redis Master節點的一種實現（一般為5）。client依次向各個節點申請鎖，若能從多數個節點中申請鎖成功并滿足一些條件限制，那么client就能獲取鎖成功。它通過獨立的N個Master節點，避免了使用主備異步復制協議的缺陷，只要多數Redis節點正常就能正常工作，顯著提升了分布式鎖的安全性、可用性。
在這里插入圖片描述
注意圖中所有的節點都是master節點。加鎖超過半數成功，就認為是成功。具體流程：
獲取鎖

獲取當前時間T1，作為后續的計時依據；

按順序地，依次向5個獨立的節點來嘗試獲取鎖 SET resource_name my_random_value NX PX 30000；

計算獲取鎖總共花了多少時間，判斷獲取鎖成功與否；

時間：T2-T1；

多數節點的鎖（N/2+1）；

當獲取鎖成功后的有效時間，要從初始的時間減去第三步算出來的消耗時間；

如果沒能獲取鎖成功，盡快釋放掉鎖。

釋放鎖

向所有節點發起釋放鎖的操作，不管這些節點有沒有成功設置過。

public String redlock() {String lockKey = "product_001";//這里需要自己實例化不同redis實例的redisson客戶端連接，這里只是偽代碼用一個redisson客戶端簡化了RLock lock1 = redisson.getLock(lockKey);RLock lock2 = redisson.getLock(lockKey);RLock lock3 = redisson.getLock(lockKey);/*** 根據多個 RLock 對象構建 RedissonRedLock （最核心的差別就在這里）*/RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);try {/*** waitTimeout 嘗試獲取鎖的最大等待時間，超過這個值，則認為獲取鎖失敗* leaseTime   鎖的持有時間,超過這個時間鎖會自動失效（值應設置為大于業務處理的時間，確保在鎖有效期內業務能處理完）*/boolean res = redLock.tryLock(10, 30, TimeUnit.SECONDS);if (res) {//成功獲得鎖，在這里處理業務}} catch (Exception e) {throw new RuntimeException("lock fail");} finally {//無論如何, 最后都要解鎖redLock.unlock();}return "end";
}

但是，它的實現建立在一個不安全的系統模型上的，它依賴系統時間，當時鐘發生跳躍時，也可能會出現安全性問題。分布式存儲專家Martin對RedLock的分析文章，Redis作者的也專門寫了一篇文章進行了反駁。

Martin Kleppmann：How to do distributed locking

https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html
Antirez：Is Redlock safe?

http://antirez.com/news/101
4.2.2 RedLock 問題一：持久化機制導致重復加鎖
如果是上面的架構圖，一般生產都不會配置AOF的每一條命令都落磁盤，一般會設置一些間隔時間，比如1s，如果ABC節點加鎖成功，有一個節點C恰好是在1s內加鎖，還沒有落盤，此時掛了，就會導致其他客戶端通過CDE又會加鎖成功。
4.2.3 RedLock 問題二：主從下重復加鎖
在這里插入圖片描述
除非多部署一些節點，但是這樣會導致加鎖時間變長，這樣比較下來效果就不如zk了。
4.2.4 RedLock 問題三：時鐘跳躍導致重復加鎖
C節點發生了時鐘跳躍，導致加上的鎖沒有到達實際的超時時間，就被誤以為超時而釋放，此時其他客戶端就可以重復加鎖了。
4.3 Curator

InterProcessMutex 可重入鎖的分析
在這里插入圖片描述
五、業務中使用分布式鎖的注意點

獲取的鎖要設置有效期，假設我們未設置key自動過期時間，在Set key value NX 后，如果程序crash或者發生網絡分區后無法與Redis節點通信，毫無疑問其他 client 將永遠無法獲得鎖，這將導致死鎖，服務出現中斷。
SETNX和EXPIRE命令去設置key和過期時間，這也是不正確的，因為你無法保證SETNX和EXPIRE命令的原子性。
自己使用 setnx 實現Redis鎖的時候，注意并發情況下不要釋放掉別人的鎖（業務邏輯執行時間超過鎖的過期時間），導致惡性循環。一般：
1）加鎖的時候需要指定value的內容是當前進程中的當前線程的唯一標記，不要使用線程ID作為當前線程的鎖的標記，因為不同實例上的線程ID可能是一樣的。
2）釋放鎖的邏輯會寫在finally ，釋放鎖時候要判斷鎖對應的value，而且要使用lua腳本實現原子 del 操作。因為if邏輯判斷完之后也可能失效導致刪除別人的鎖。
3）針對扣減庫存這個邏輯，lua腳本里面實現Redis比較庫存、扣減庫存操作的原子性。通過判斷Redis Decr命令的返回值即可。此命令會返回扣減后的最新庫存，若小于0則表示超賣。

5.1 自己實現分布式鎖的坑

setnx不關心鎖的順序導致刪除別人的鎖
鎖失效之后，別人加鎖成功，自己把別人的鎖刪了。
我們無法預估程序執行需要的鎖的時間。

public String deductStock() {String lockKey = "lock:product_101";Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "deltaqin");stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);try {int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")if (stock > 0) {int realStock = stock - 1;stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)System.out.println("扣減成功，剩余庫存:" + realStock);} else {System.out.println("扣減失敗，庫存不足");}} finally {stringRedisTemplate.delete(lockKey);}return "end";
}

setnx關心鎖的順序還是刪除了別人的鎖
并發會卡在各種地方，卡住的時候過期了，就會刪掉別人加的鎖：
錯誤的原因還是因為解鎖的邏輯不是原子性的，這里可以參考Redisson的解鎖邏輯使用lua腳本實現。

public String deductStock() {String lockKey = "lock:product_101";String clientId = UUID.randomUUID().toString();Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS); //jedis.setnx(k,v)if (!result) {return "error_code";}try {int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")if (stock > 0) {int realStock = stock - 1;stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)System.out.println("扣減成功，剩余庫存:" + realStock);} else {System.out.println("扣減失敗，庫存不足");}} finally {if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {// 卡在這里，鎖過期了，其他線程又可以加鎖，此時又把其他線程新加的鎖刪掉了stringRedisTemplate.delete(lockKey);}}return "end";
}

解決辦法
這種問題解決的辦法就是使用鎖續命，比如使用一個定時任務間隔小于鎖的超時時間，每隔一段時間就給鎖續命，除非線程自己主動刪除。這也是Redisson的實現思路。
5.2 鎖優化：分段加鎖邏輯

針對一個商品，要開啟秒殺的時候，會將商品的庫存預先加載到Redis緩存中，比如有100個庫存，此時可以分為5個key，每一個key有20個庫存。可以把分布式鎖的性能提升5倍。
例如：

product_10111_stock = 100product_10111_stock1 = 20product_10111_stock2 = 20product_10111_stock3 = 20product_10111_stock4 = 20product_10111_stock5 = 20