一、分布式鎖介紹

之前我們都是使用本地鎖（synchronize、lock等）來避免共享資源并發操作導致數據問題，這種是鎖在當前進程內。

那么在集群部署下，對于多個節點，我們要使用分布式鎖來避免共享資源并發操作導致數據問題，雖然還是鎖，但是是多個進程共用的鎖標記，可以用Redis、Zookeeper、Mysql等都可以實現。

案例：優惠券領劵限制張數、商品庫存超賣。

我們設計分布式鎖應該要考慮的東西：

• 排他性：在分布式應用集群中，同一個方法在同一時間只能被一臺機器上的一個線程執行。

• 容錯性：分布式鎖一定能得到釋放，比如客戶端奔潰或者網絡中斷，可能會導致鎖一直不被釋放，從而導致死鎖，我們可以設置鎖的過期時間。

• 滿足可重入、高性能、高可用（集群部署）。

• 注意分布式鎖的開銷、鎖粒度。

二、分布式鎖的實現

實現分布式鎖可以用 Redis、Zookeeper、Mysql數據庫這幾種 , 性能最好的是Redis且是最容易理解。

分布式鎖離不開 key - value 設置，key 是鎖的唯一標識，一般按業務來決定命名，比如想要給一種優惠券活動加鎖，key 命名為 “coupon:id” 。value就可以使用固定值，比如設置成1。

基于redis實現分布式鎖：

（1）、加鎖 setnx key value：

setnx 的含義就是 set if not exists，有兩個參數 setnx(key, value)，該方法是原子性操作，如果 key 不存在，則設置當前 key 成功，返回 1；如果當前 key 已經存在，則設置當前 key 失敗，返回 0。

（2）、解鎖 del (key)：

得到鎖的線程執行完任務，需要釋放鎖，以便其他線程可以進入,調用 del(key)。

（3)、配置鎖超時 expire (key，30s）：

客戶端奔潰或者網絡中斷，資源將會永遠被鎖住,即死鎖，因此需要給key配置過期時間，以保證即使沒有被顯式釋放，這把鎖也要在一定時間后自動釋放。

綜合的偽代碼：

method(){String key = "coupon:id"
?if（setnx（key，1） == 1）{expire(key,30,TimeUnit.MILLISECONDS)try {//做對應的業務邏輯//查詢用戶是否已經領券//如果沒有則扣減庫存//新增領劵記錄} finally {del（key）}}else{
?//睡眠100毫秒，然后自旋調用本方法method()}
}

三、?基于Redis實現分布式鎖的幾種坑

上面我們寫的偽代碼中有幾個坑，我們分別來分析一下。

1、多個命令之間不是原子性操作，如setnx和expire之間，如果setnx成功，但是expire失敗，且宕機了，則這個資源就是死鎖。

解決方法：使用原子命令來設置和配置過期時間 setnx / setex，在java里面是

redisTemplate.opsForValue().setIfAbsent("key","value",30,TimeUnit.MILLISECONDS)

成功了返回true，失敗了返回false。?

2、業務超時，存在其他線程勿刪，設置key30秒過期，假如線程A執行很慢超過30秒，則key就被釋放了，其他線程B就得到了鎖，這個時候線程A執行完成，而B還沒執行完成，結果就是線程A刪除了線程B加的鎖，所以我們的value不能單單只是1。

解決方法：可以在 del 釋放鎖之前做一個判斷，驗證當前的鎖是不是自己加的鎖, 那 value 應該是當前線程的標識或者uuid。

String key = "coupon:id"
String value = Thread.currentThread().getId()
?
if（setnx（key，value） == 1）{expire(key,30,TimeUnit.MILLISECONDS)try {//做對應的業務邏輯} finally {//刪除鎖，判斷是否是當前線程加的if(get(key).equals(value)){//還存在時間間隔del（key）}}
}else{//睡眠100毫秒，然后自旋調用本方法
?
}

?3、進一步細化誤刪，當線程A獲取到正常值value時，返回帶代碼中判斷期間鎖過期了，線程B剛好重新設置了新值，線程A那邊有判斷value是自己的標識，然后調用del方法，結果就是刪除了新設置的線程B的值。

解決辦法：由于redis沒有相關的原子性api，所以采用 lua腳本+redis來實現多個命令的原子性。由于【判斷和刪除】是lua腳本執行，所以要么全成功，要么全失敗。

總結：核心是保證多個指令原子性，加鎖使用setnx setex 可以保證原子性，解鎖采用 lua腳本+redis來保證原子性。

【判斷和刪除】的lua腳本：

//獲取lock的值和傳遞的值一樣，調用刪除操作返回1，否則返回0
String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
?
//Arrays.asList(lockKey)是key列表，uuid是參數
Long result = redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockKey), uuid);

?四、原生分布式鎖的具體實現

@RestController
@RequestMapping("/api/v1/coupon")
public class CouponController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@GetMapping("add")public JsonData saveCoupon(@RequestParam (value = "coupon_id",required = true)int couponId){//防止其他線程誤刪String uuid = UUID.randomUUID().toString();String lockKey = "lock:coupon:" + couponId;lock(couponId,uuid,lockKey);return JsonData.buildSuccess();}private void lock(int couponId,String uuid,String lockKey){//lua腳本String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";Boolean nativeLock = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,uuid, Duration.ofSeconds(30));System.out.println(uuid+"加鎖狀態:"+nativeLock);if(nativeLock){//加鎖成功try{//TODO 做相關業務邏輯TimeUnit.SECONDS.sleep(10L);} catch (InterruptedException e) {} finally {//解鎖Long result = stringRedisTemplate.execute( new DefaultRedisScript<>(script,Long.class), Arrays.asList(lockKey),uuid);System.out.println("解鎖狀態:"+result);}}else {//自旋操作try {System.out.println("加鎖失敗，睡眠5秒 進行自旋");TimeUnit.MILLISECONDS.sleep(5000);} catch (InterruptedException e) { }//睡眠一會再嘗試獲取鎖lock(couponId,uuid,lockKey);}}}

運行結果：

d124ae03-5de6-4e25-82b8-fb0b30d7c7fc加鎖狀態:true
54041d23-ab3c-492e-977b-99c9b531534f加鎖狀態:false
加鎖失敗，睡眠5秒 進行自旋
51f16a96-45cd-476b-95ff-2ee6cc398e37加鎖狀態:false
加鎖失敗，睡眠5秒 進行自旋
54041d23-ab3c-492e-977b-99c9b531534f加鎖狀態:false
加鎖失敗，睡眠5秒 進行自旋
51f16a96-45cd-476b-95ff-2ee6cc398e37加鎖狀態:false
加鎖失敗，睡眠5秒 進行自旋
解鎖狀態:1
54041d23-ab3c-492e-977b-99c9b531534f加鎖狀態:true
51f16a96-45cd-476b-95ff-2ee6cc398e37加鎖狀態:false
加鎖失敗，睡眠5秒 進行自旋
51f16a96-45cd-476b-95ff-2ee6cc398e37加鎖狀態:false
加鎖失敗，睡眠5秒 進行自旋
解鎖狀態:1
51f16a96-45cd-476b-95ff-2ee6cc398e37加鎖狀態:true
解鎖狀態:1

?