參考：
https://www.jianshu.com/p/a1ebab8ce78a
https://www.cnblogs.com/moxiaotao/p/10829799.html
https://www.cnblogs.com/wlwl/p/11651409.html
https://www.cnblogs.com/austinspark-jessylu/p/8043726.html
https://blog.csdn.net/wuzhiwei549/article/details/80692278
什么是分布式鎖
概述
為了防止分布式系統中的多個進程之間相互干擾，我們需要一種分布式協調技術來對這些進程進行調度。而這個分布式協調技術的核心就是來實現這個分布式鎖。
為什么要使用分布式鎖
- 成員變量 A 存在 JVM1、JVM2、JVM3 三個 JVM 內存中
- 成員變量 A 同時都會在 JVM 分配一塊內存，三個請求發過來同時對這個變量操作，顯然結果是不對的
- 不是同時發過來，三個請求分別操作三個不同 JVM 內存區域的數據，變量 A 之間不存在共享，也不具有可見性，處理的結果也是不對的
注：該成員變量 A 是一個有狀態的對象
如果我們業務中確實存在這個場景的話，我們就需要一種方法解決這個問題，這就是分布式鎖要解決的問題
分布式鎖應該具備哪些條件
- 在分布式系統環境下，一個方法在同一時間只能被一個機器的一個線程執行
- 高可用的獲取鎖與釋放鎖
- 高性能的獲取鎖與釋放鎖
- 具備可重入特性（可理解為重新進入，由多於一個任務並發使用，而不必擔心數據錯誤）
- 具備鎖失效機制，防止死鎖
- 具備非阻塞鎖特性，即沒有獲取到鎖將直接返回獲取鎖失敗
分布式鎖的實現有哪些
- Memcached：利用 Memcached 的
add命令。此命令是原子性操作，只有在key不存在的情況下，才能add成功，也就意味着線程得到了鎖。 - Redis：和 Memcached 的方式類似，利用 Redis 的
setnx命令。此命令同樣是原子性操作，只有在key不存在的情況下，才能set成功。 - Zookeeper：利用 Zookeeper 的順序臨時節點，來實現分布式鎖和等待隊列。Zookeeper 設計的初衷，就是為了實現分布式鎖服務的。
- Chubby：Google 公司實現的粗粒度分布式鎖服務，底層利用了 Paxos 一致性算法。
通過 Redis 分布式鎖的實現理解基本概念
分布式鎖實現的三個核心要素：
加鎖
最簡單的方法是使用 setnx 命令。key 是鎖的唯一標識，按業務來決定命名。比如想要給一種商品的秒殺活動加鎖，可以給 key 命名為 “lock_sale_商品ID” 。而 value 設置成什么呢？我們可以姑且設置成 1。加鎖的偽代碼如下：
setnx（lock_sale_商品ID，1）
當一個線程執行 setnx 返回 1，說明 key 原本不存在，該線程成功得到了鎖；當一個線程執行 setnx 返回 0，說明 key 已經存在，該線程搶鎖失敗。
解鎖
有加鎖就得有解鎖。當得到鎖的線程執行完任務，需要釋放鎖，以便其他線程可以進入。釋放鎖的最簡單方式是執行 del 指令，偽代碼如下：
del（lock_sale_商品ID） 釋放鎖之后，其他線程就可以繼續執行 setnx 命令來獲得鎖。
鎖超時
鎖超時是什么意思呢？如果一個得到鎖的線程在執行任務的過程中掛掉，來不及顯式地釋放鎖，這塊資源將會永遠被鎖住（死鎖），別的線程再也別想進來。所以，setnx 的 key 必須設置一個超時時間，以保證即使沒有被顯式釋放，這把鎖也要在一定時間后自動釋放。setnx 不支持超時參數，所以需要額外的指令，偽代碼如下：
expire（lock_sale_商品ID， 30）
綜合偽代碼如下：
if（setnx（lock_sale_商品ID，1） == 1）{ expire（lock_sale_商品ID，30） try { do something ...... } finally { del（lock_sale_商品ID） } } 存在什么問題
以上偽代碼中存在三個致命問題
setnx 和 expire 的非原子性
設想一個極端場景，當某線程執行 setnx，成功得到了鎖：
setnx 剛執行成功，還未來得及執行 expire 指令，節點 1 掛掉了。
這樣一來，這把鎖就沒有設置過期時間，變成死鎖，別的線程再也無法獲得鎖了。
怎么解決呢？setnx 指令本身是不支持傳入超時時間的，set 指令增加了可選參數，偽代碼如下：
set（lock_sale_商品ID，1，30，NX）
這樣就可以取代 setnx 指令。
del 導致誤刪
又是一個極端場景，假如某線程成功得到了鎖，並且設置的超時時間是 30 秒。
如果某些原因導致線程 A 執行的很慢很慢，過了 30 秒都沒執行完，這時候鎖過期自動釋放，線程 B 得到了鎖。
隨后，線程 A 執行完了任務，線程 A 接着執行 del 指令來釋放鎖。但這時候線程 B 還沒執行完，線程A實際上 刪除的是線程 B 加的鎖。
怎么避免這種情況呢？可以在 del 釋放鎖之前做一個判斷，驗證當前的鎖是不是自己加的鎖。至於具體的實現，可以在加鎖的時候把當前的線程 ID 當做 value，並在刪除之前驗證 key 對應的 value 是不是自己線程的 ID。
加鎖：
String threadId = Thread.currentThread().getId() set（key，threadId ，30，NX） 解鎖：
if（threadId .equals(redisClient.get(key))）{ del(key) } 但是，這樣做又隱含了一個新的問題，判斷和釋放鎖是兩個獨立操作，不是原子性。
出現並發的可能性
還是剛才第二點所描述的場景，雖然我們避免了線程 A 誤刪掉 key 的情況，但是同一時間有 A，B 兩個線程在訪問代碼塊，仍然是不完美的。怎么辦呢？我們可以讓獲得鎖的線程開啟一個守護線程，用來給快要過期的鎖“續航”。
當過去了 29 秒，線程 A 還沒執行完，這時候守護線程會執行 expire 指令，為這把鎖“續命”20 秒。守護線程從第 29 秒開始執行，每 20 秒執行一次。
當線程 A 執行完任務，會顯式關掉守護線程。
另一種情況，如果節點 1 忽然斷電，由於線程 A 和守護線程在同一個進程，守護線程也會停下。這把鎖到了超時的時候，沒人給它續命，也就自動釋放了。
作者：擼帝
鏈接：https://www.jianshu.com/p/a1ebab8ce78a
來源：簡書
著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。
Redis分布式鎖的正確實現方式
前言
分布式鎖一般有三種實現方式：1. 數據庫樂觀鎖；2. 基於Redis的分布式鎖；3. 基於ZooKeeper的分布式鎖。本篇博客將介紹第二種方式，基於Redis實現分布式鎖。雖然網上已經有各種介紹Redis分布式鎖實現的博客，然而他們的實現卻有着各種各樣的問題，為了避免誤人子弟，本篇博客將詳細介紹如何正確地實現Redis分布式鎖。
可靠性
首先，為了確保分布式鎖可用，我們至少要確保鎖的實現同時滿足以下四個條件：
- 互斥性。在任意時刻，只有一個客戶端能持有鎖。
- 不會發生死鎖。即使有一個客戶端在持有鎖的期間崩潰而沒有主動解鎖，也能保證后續其他客戶端能加鎖。
- 具有容錯性。只要大部分的Redis節點正常運行，客戶端就可以加鎖和解鎖。
- 解鈴還須系鈴人。加鎖和解鎖必須是同一個客戶端，客戶端自己不能把別人加的鎖給解了。
代碼實現
組件依賴
首先我們要通過Maven引入Jedis開源組件，在pom.xml文件加入下面的代碼：
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
加鎖代碼
正確姿勢
Talk is cheap, show me the code。先展示代碼，再帶大家慢慢解釋為什么這樣實現：
public class RedisTool {
private static final String LOCK_SUCCESS = "OK";
private static final String SET_IF_NOT_EXIST = "NX";
private static final String SET_WITH_EXPIRE_TIME = "PX";
/**
* 嘗試獲取分布式鎖
* @param jedis Redis客戶端
* @param lockKey 鎖
* @param requestId 請求標識
* @param expireTime 超期時間
* @return 是否獲取成功
*/
public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
if (LOCK_SUCCESS.equals(result)) {
return true;
}
return false;
}
}
可以看到，我們加鎖就一行代碼：jedis.set(String key, String value, String nxxx, String expx, int time)，這個set()方法一共有五個形參：
-
第一個為key，我們使用key來當鎖，因為key是唯一的。
-
第二個為value，我們傳的是requestId，很多童鞋可能不明白，有key作為鎖不就夠了嗎，為什么還要用到value？原因就是我們在上面講到可靠性時，分布式鎖要滿足第四個條件解鈴還須系鈴人，通過給value賦值為requestId，我們就知道這把鎖是哪個請求加的了，在解鎖的時候就可以有依據。requestId可以使用
UUID.randomUUID().toString()方法生成。 -
第三個為nxxx，這個參數我們填的是NX，意思是SET IF NOT EXIST，即當key不存在時，我們進行set操作；若key已經存在，則不做任何操作；
-
第四個為expx，這個參數我們傳的是PX，意思是我們要給這個key加一個過期的設置，具體時間由第五個參數決定。
-
第五個為time，與第四個參數相呼應，代表key的過期時間。
總的來說，執行上面的set()方法就只會導致兩種結果：1. 當前沒有鎖（key不存在），那么就進行加鎖操作，並對鎖設置個有效期，同時value表示加鎖的客戶端。2. 已有鎖存在，不做任何操作。
心細的童鞋就會發現了，我們的加鎖代碼滿足我們可靠性里描述的三個條件。首先，set()加入了NX參數，可以保證如果已有key存在，則函數不會調用成功，也就是只有一個客戶端能持有鎖，滿足互斥性。其次，由於我們對鎖設置了過期時間，即使鎖的持有者后續發生崩潰而沒有解鎖，鎖也會因為到了過期時間而自動解鎖（即key被刪除），不會發生死鎖。最后，因為我們將value賦值為requestId，代表加鎖的客戶端請求標識，那么在客戶端在解鎖的時候就可以進行校驗是否是同一個客戶端。由於我們只考慮Redis單機部署的場景，所以容錯性我們暫不考慮。
錯誤示例1
比較常見的錯誤示例就是使用jedis.setnx()和jedis.expire()組合實現加鎖，代碼如下：
public static void wrongGetLock1(Jedis jedis, String lockKey, String requestId, int expireTime) {
Long result = jedis.setnx(lockKey, requestId);
if (result == 1) {
// 若在這里程序突然崩潰，則無法設置過期時間，將發生死鎖
jedis.expire(lockKey, expireTime);
}
}
setnx()方法作用就是SET IF NOT EXIST，expire()方法就是給鎖加一個過期時間。乍一看好像和前面的set()方法結果一樣，然而由於這是兩條Redis命令，不具有原子性，如果程序在執行完setnx()之后突然崩潰，導致鎖沒有設置過期時間。那么將會發生死鎖。網上之所以有人這樣實現，是因為低版本的jedis並不支持多參數的set()方法。
錯誤示例2
public static boolean wrongGetLock2(Jedis jedis, String lockKey, int expireTime) {
long expires = System.currentTimeMillis() + expireTime;
String expiresStr = String.valueOf(expires);
// 如果當前鎖不存在，返回加鎖成功
if (jedis.setnx(lockKey, expiresStr) == 1) {
return true;
}
// 如果鎖存在，獲取鎖的過期時間
String currentValueStr = jedis.get(lockKey);
if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
// 鎖已過期，獲取上一個鎖的過期時間，並設置現在鎖的過期時間
String oldValueStr = jedis.getSet(lockKey, expiresStr);
if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
// 考慮多線程並發的情況，只有一個線程的設置值和當前值相同，它才有權利加鎖
return true;
}
}
// 其他情況，一律返回加鎖失敗
return false;
}
這一種錯誤示例就比較難以發現問題，而且實現也比較復雜。實現思路：使用jedis.setnx()命令實現加鎖，其中key是鎖，value是鎖的過期時間。執行過程：1. 通過setnx()方法嘗試加鎖，如果當前鎖不存在，返回加鎖成功。2. 如果鎖已經存在則獲取鎖的過期時間，和當前時間比較，如果鎖已經過期，則設置新的過期時間，返回加鎖成功。代碼如下：
那么這段代碼問題在哪里？1. 由於是客戶端自己生成過期時間，所以需要強制要求分布式下每個客戶端的時間必須同步。 2. 當鎖過期的時候，如果多個客戶端同時執行jedis.getSet()方法，那么雖然最終只有一個客戶端可以加鎖，但是這個客戶端的鎖的過期時間可能被其他客戶端覆蓋。3. 鎖不具備擁有者標識，即任何客戶端都可以解鎖。
解鎖代碼
正確姿勢
還是先展示代碼，再帶大家慢慢解釋為什么這樣實現：
public class RedisTool {
private static final Long RELEASE_SUCCESS = 1L;
/**
* 釋放分布式鎖
* @param jedis Redis客戶端
* @param lockKey 鎖
* @param requestId 請求標識
* @return 是否釋放成功
*/
public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
if (RELEASE_SUCCESS.equals(result)) {
return true;
}
return false;
}
}
可以看到，我們解鎖只需要兩行代碼就搞定了！第一行代碼，我們寫了一個簡單的Lua腳本代碼，上一次見到這個編程語言還是在《黑客與畫家》里，沒想到這次居然用上了。第二行代碼，我們將Lua代碼傳到jedis.eval()方法里，並使參數KEYS[1]賦值為lockKey，ARGV[1]賦值為requestId。eval()方法是將Lua代碼交給Redis服務端執行。
那么這段Lua代碼的功能是什么呢？其實很簡單，首先獲取鎖對應的value值，檢查是否與requestId相等，如果相等則刪除鎖（解鎖）。那么為什么要使用Lua語言來實現呢？因為要確保上述操作是原子性的。關於非原子性會帶來什么問題，可以閱讀【解鎖代碼-錯誤示例2】 。那么為什么執行eval()方法可以確保原子性，源於Redis的特性，下面是官網對eval命令的部分解釋：
簡單來說，就是在eval命令執行Lua代碼的時候，Lua代碼將被當成一個命令去執行，並且直到eval命令執行完成，Redis才會執行其他命令。
錯誤示例1
最常見的解鎖代碼就是直接使用jedis.del()方法刪除鎖，這種不先判斷鎖的擁有者而直接解鎖的方式，會導致任何客戶端都可以隨時進行解鎖，即使這把鎖不是它的。
public static void wrongReleaseLock1(Jedis jedis, String lockKey) {
jedis.del(lockKey);
}
錯誤示例2
這種解鎖代碼乍一看也是沒問題，甚至我之前也差點這樣實現，與正確姿勢差不多，唯一區別的是分成兩條命令去執行，代碼如下：
public static void wrongReleaseLock2(Jedis jedis, String lockKey, String requestId) {
// 判斷加鎖與解鎖是不是同一個客戶端
if (requestId.equals(jedis.get(lockKey))) {
// 若在此時，這把鎖突然不是這個客戶端的，則會誤解鎖
jedis.del(lockKey);
}
}
如代碼注釋，問題在於如果調用jedis.del()方法的時候，這把鎖已經不屬於當前客戶端的時候會解除他人加的鎖。那么是否真的有這種場景？答案是肯定的，比如客戶端A加鎖，一段時間之后客戶端A解鎖，在執行jedis.del()之前，鎖突然過期了，此時客戶端B嘗試加鎖成功，然后客戶端A再執行del()方法，則將客戶端B的鎖給解除了。
總結
本文主要介紹了如何使用Java代碼正確實現Redis分布式鎖，對於加鎖和解鎖也分別給出了兩個比較經典的錯誤示例。其實想要通過Redis實現分布式鎖並不難，只要保證能滿足可靠性里的四個條件。互聯網雖然給我們帶來了方便，只要有問題就可以google，然而網上的答案一定是對的嗎？其實不然，所以我們更應該時刻保持着質疑精神，多想多驗證。
如果你的項目中Redis是多機部署的，那么可以嘗試使用Redisson實現分布式鎖，這是Redis官方提供的Java組件，鏈接在參考閱讀章節已經給出。
3種Redis分布式鎖的對比
我們通常使用的synchronized或者Lock都是線程鎖，對同一個JVM進程內的多個線程有效。因為鎖的本質 是內存中存放一個標記，記錄獲取鎖的線程是誰，這個標記對每個線程都可見。然而我們啟動的多個訂單服務，就是多個JVM，內存中的鎖顯然是不共享的，每個JVM進程都有自己的 鎖，自然無法保證線程的互斥了，這個時候我們就需要使用到分布式鎖了。常用的有三種解決方案:1.基於數據庫實現 2.基於zookeeper的臨時序列化節點實現 3.redis實現。本文我們介紹的就是redis的實現方式。
實現分布式鎖要滿足3點：多進程可見，互斥，可重入。
1） 多進程可見
redis本身就是基於JVM之外的，因此滿足多進程可見的要求。
2） 互斥
即同一時間只能有一個進程獲取鎖標記，我們可以通過redis的setnx實現，只有第一次執行的才會成功並返回1，其它情況返回0。
釋放鎖
釋放鎖其實只需要把鎖的key刪除即可，使用del xxx指令。不過，如果在我們執行del之前，服務突然宕機，那么鎖就永遠無法刪除了。所以我們可以通過setex 命令設置過期時間即可。
import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
/**
* 第一種分布式鎖
*/
@Component
public class RedisService {
private final Logger log = LoggerFactory.getLogger(this.getClass());
@Autowired
JedisPool jedisPool;
// 獲取鎖之前的超時時間(獲取鎖的等待重試時間)
private long acquireTimeout = 5000;
// 獲取鎖之后的超時時間(防止死鎖)
private int timeOut = 10000;
/**
* 獲取分布式鎖
* @return 鎖標識
*/
public boolean getRedisLock(String lockName,String val) {
Jedis jedis = null;
try {
jedis = jedisPool.getResource();
// 1.計算獲取鎖的時間
Long endTime = System.currentTimeMillis() + acquireTimeout;
// 2.嘗試獲取鎖
while (System.currentTimeMillis() < endTime) {
// 3. 獲取鎖成功就設置過期時間
if (jedis.setnx(lockName, val) == 1) {
jedis.expire(lockName, timeOut/1000);
return true;
}
}
} catch (Exception e) {
log.error(e.getMessage());
} finally {
returnResource(jedis);
}
return false;
}
/**
* 釋放分布式鎖
* @param lockName 鎖名稱
*/
public void unRedisLock(String lockName) {
Jedis jedis = null;
try {
jedis = jedisPool.getResource();
// 釋放鎖
jedis.del(lockName);
} catch (Exception e) {
log.error(e.getMessage());
} finally {
returnResource(jedis);
}
}
// ===============================================
public String get(String key) {
Jedis jedis = null;
String value = null;
try {
jedis = jedisPool.getResource();
value = jedis.get(key);
log.info(value);
} catch (Exception e) {
log.error(e.getMessage());
} finally {
returnResource(jedis);
}
return value;
}
public void set(String key, String value) {
Jedis jedis = null;
try {
jedis = jedisPool.getResource();
jedis.set(key, value);
} catch (Exception e) {
log.error(e.getMessage());
} finally {
returnResource(jedis);
}
}
/**
* 關閉連接
*/
public void returnResource(Jedis jedis) {
try {
if(jedis!=null) jedis.close();
} catch (Exception e) {
}
}
}
上面的分布式鎖實現了，但是這時候還可能出現另外2個問題：
一：獲取鎖時
setnx獲取鎖成功了，還沒來得及setex服務就宕機了，由於這種非原子性的操作，死鎖又發生了。其實redis提供了 nx 與 ex連用的命令。
二：釋放鎖時
1. 3個進程：A和B和C，在執行任務，並爭搶鎖，此時A獲取了鎖，並設置自動過期時間為10s
2. A開始執行業務，因為某種原因，業務阻塞，耗時超過了10秒，此時鎖自動釋放了
3. B恰好此時開始嘗試獲取鎖，因為鎖已經自動釋放，成功獲取鎖
4. A此時業務執行完畢，執行釋放鎖邏輯（刪除key），於是B的鎖被釋放了，而B其實還在執行業務
5. 此時進程C嘗試獲取鎖，也成功了，因為A把B的鎖刪除了。
問題出現了：B和C同時獲取了鎖，違反了互斥性！如何解決這個問題呢？我們應該在刪除鎖之前，判斷這個鎖是否是自己設置的鎖，如果不是（例如自己 的鎖已經超時釋放），那么就不要刪除了。所以我們可以在set 鎖時，存入當前線程的唯一標識！刪除鎖前，判斷下里面的值是不是與自己標識釋放一 致，如果不一致，說明不是自己的鎖，就不要刪除了。
/**
* 第二種分布式鎖
*/
public class RedisTool {
private static final String LOCK_SUCCESS = "OK";
private static final Long RELEASE_SUCCESS = 1L;
/**
* 嘗試獲取分布式鎖
* @param jedis Redis客戶端
* @param lockKey 鎖
* @param requestId 請求標識
* @param expireTime 超期時間
* @return 是否獲取成功
*/
public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
if (LOCK_SUCCESS.equals(result)) {
return true;
}
return false;
}
/**
* 釋放分布式鎖
* @param jedis Redis客戶端
* @param lockKey 鎖
* @param requestId 請求標識
* @return 是否釋放成功
*/
public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {
if (jedis.get(lockKey).equals(requestId)) {
System.out.println("釋放鎖..." + Thread.currentThread().getName() + ",identifierValue:" + requestId);
jedis.del(lockKey);
return true;
}
return false;
}
}
按照上面方式實現分布式鎖之后，就可以輕松解決大部分問題了。網上很多博客也都是這么實現的，但是仍然有些場景是不滿足的，例如一個方法獲取到鎖之后，可能在方法內調這個方法此時就獲取不到鎖了。這個時候我們就需要把鎖改進成可重入式鎖了。
3） 重入鎖：
也叫做遞歸鎖，指的是在同一線程內，外層函數獲得鎖之后，內層遞歸函數仍然可以獲取到該鎖。換一種說法：同一個線程再次進入同步代碼時，可以使用自己已獲取到的鎖。可重入鎖可以避免因同一線程中多次獲取鎖而導致死鎖發生。像synchronized就是一個重入鎖，它是通過moniter函數記錄當前線程信息來實現的。實現可重入鎖需要考慮兩點：
獲取鎖：首先嘗試獲取鎖，如果獲取失敗，判斷這個鎖是否是自己的，如果是則允許再次獲取， 而且必須記錄重復獲取鎖的次數。
釋放鎖：釋放鎖不能直接刪除了，因為鎖是可重入的，如果鎖進入了多次，在內層直接刪除鎖， 導致外部的業務在沒有鎖的情況下執行，會有安全問題。因此必須獲取鎖時累計重入的次數，釋放時則減去重入次數，如果減到0，則可以刪除鎖。
下面我們假設鎖的key為“ lock ”，hashKey是當前線程的id：“ threadId ”，鎖自動釋放時間假設為20
獲取鎖的步驟：
1、判斷lock是否存在 EXISTS lock
2、不存在，則自己獲取鎖，記錄重入層數為1.
2、存在，說明有人獲取鎖了，下面判斷是不是自己的鎖,即判斷當前線程id作為hashKey是否存在：HEXISTS lock threadId
3、不存在，說明鎖已經有了，且不是自己獲取的，鎖獲取失敗.
3、存在，說明是自己獲取的鎖，重入次數+1： HINCRBY lock threadId 1 ，最后更新鎖自動釋放時間， EXPIRE lock 20
釋放鎖的步驟：
1、判斷當前線程id作為hashKey是否存在： HEXISTS lock threadId
2、不存在，說明鎖已經失效，不用管了
2、存在，說明鎖還在，重入次數減1： HINCRBY lock threadId -1 ，
3、獲取新的重入次數，判斷重入次數是否為0，為0說明鎖全部釋放，刪除key： DEL lock
因此，存儲在鎖中的信息就必須包含：key、線程標識、重入次數。不能再使用簡單的key-value結構， 這里推薦使用hash結構。
獲取鎖的腳本(注釋刪掉,不然運行報錯)
local key = KEYS[1]; -- 第1個參數,鎖的key
local threadId = ARGV[1]; -- 第2個參數,線程唯一標識
local releaseTime = ARGV[2]; -- 第3個參數,鎖的自動釋放時間
if(redis.call('exists', key) == 0) then -- 判斷鎖是否已存在
redis.call('hset', key, threadId, '1'); -- 不存在, 則獲取鎖
redis.call('expire', key, releaseTime); -- 設置有效期
return 1; -- 返回結果
end;
if(redis.call('hexists', key, threadId) == 1) then -- 鎖已經存在，判斷threadId是否是自己
redis.call('hincrby', key, threadId, '1'); -- 如果是自己，則重入次數+1
redis.call('expire', key, releaseTime); -- 設置有效期
return 1; -- 返回結果
end;
return 0; -- 代碼走到這里,說明獲取鎖的不是自己，獲取鎖失敗
釋放鎖的腳本(注釋刪掉,不然運行報錯)
local key = KEYS[1]; -- 第1個參數,鎖的key
local threadId = ARGV[1]; -- 第2個參數,線程唯一標識
if (redis.call('HEXISTS', key, threadId) == 0) then -- 判斷當前鎖是否還是被自己持有
return nil; -- 如果已經不是自己，則直接返回
end;
local count = redis.call('HINCRBY', key, threadId, -1); -- 是自己的鎖，則重入次數-1
if (count == 0) then -- 判斷是否重入次數是否已經為0
redis.call('DEL', key); -- 等於0說明可以釋放鎖，直接刪除
return nil;
end;
完整代碼
import java.util.Collections;
import java.util.UUID;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
/**
* Redis可重入鎖
*/
public class RedisLock {
private static final StringRedisTemplate redisTemplate = SpringUtil.getBean(StringRedisTemplate.class);
private static final DefaultRedisScript<Long> LOCK_SCRIPT;
private static final DefaultRedisScript<Object> UNLOCK_SCRIPT;
static {
// 加載釋放鎖的腳本
LOCK_SCRIPT = new DefaultRedisScript<>();
LOCK_SCRIPT.setScriptSource(new ResourceScriptSource(new ClassPathResource("lock.lua")));
LOCK_SCRIPT.setResultType(Long.class);
// 加載釋放鎖的腳本
UNLOCK_SCRIPT = new DefaultRedisScript<>();
UNLOCK_SCRIPT.setScriptSource(new ResourceScriptSource(new ClassPathResource("unlock.lua")));
}
/**
* 獲取鎖
* @param lockName 鎖名稱
* @param releaseTime 超時時間(單位:秒)
* @return key 解鎖標識
*/
public static String tryLock(String lockName,String releaseTime) {
// 存入的線程信息的前綴，防止與其它JVM中線程信息沖突
String key = UUID.randomUUID().toString();
// 執行腳本
Long result = redisTemplate.execute(
LOCK_SCRIPT,
Collections.singletonList(lockName),
key + Thread.currentThread().getId(), releaseTime);
// 判斷結果
if(result != null && result.intValue() == 1) {
return key;
}else {
return null;
}
}
/**
* 釋放鎖
* @param lockName 鎖名稱
* @param key 解鎖標識
*/
public static void unlock(String lockName,String key) {
// 執行腳本
redisTemplate.execute(
UNLOCK_SCRIPT,
Collections.singletonList(lockName),
key + Thread.currentThread().getId(), null);
}
}
至此，一個比較完善的redis鎖就開發完成了。
分布式鎖的幾種實現方式
目前幾乎很多大型網站及應用都是分布式部署的，分布式場景中的數據一致性問題一直是一個比較重要的話題。分布式的CAP理論告訴我們“任何一個分布式系統都無法同時滿足一致性（Consistency）、可用性（Availability）和分區容錯性（Partition tolerance），最多只能同時滿足兩項。”所以，很多系統在設計之初就要對這三者做出取舍。在互聯網領域的絕大多數的場景中，都需要犧牲強一致性來換取系統的高可用性，系統往往只需要保證“最終一致性”，只要這個最終時間是在用戶可以接受的范圍內即可。
在很多場景中，我們為了保證數據的最終一致性，需要很多的技術方案來支持，比如分布式事務、分布式鎖等。有的時候，我們需要保證一個方法在同一時間內只能被同一個線程執行。在單機環境中，Java中其實提供了很多並發處理相關的API，但是這些API在分布式場景中就無能為力了。也就是說單純的Java Api並不能提供分布式鎖的能力。所以針對分布式鎖的實現目前有多種方案。
針對分布式鎖的實現，目前比較常用的有以下幾種方案：
基於數據庫實現分布式鎖 基於緩存（redis，memcached，tair）實現分布式鎖 基於Zookeeper實現分布式鎖
在分析這幾種實現方案之前我們先來想一下，我們需要的分布式鎖應該是怎么樣的？（這里以方法鎖為例，資源鎖同理）
可以保證在分布式部署的應用集群中，同一個方法在同一時間只能被一台機器上的一個線程執行。
這把鎖要是一把可重入鎖（避免死鎖）
這把鎖最好是一把阻塞鎖（根據業務需求考慮要不要這條）
有高可用的獲取鎖和釋放鎖功能
獲取鎖和釋放鎖的性能要好
基於數據庫實現分布式鎖
基於數據庫表
要實現分布式鎖，最簡單的方式可能就是直接創建一張鎖表，然后通過操作該表中的數據來實現了。
當我們要鎖住某個方法或資源時，我們就在該表中增加一條記錄，想要釋放鎖的時候就刪除這條記錄。
創建這樣一張數據庫表：
當我們想要鎖住某個方法時，執行以下SQL：
因為我們對method_name做了唯一性約束，這里如果有多個請求同時提交到數據庫的話，數據庫會保證只有一個操作可以成功，那么我們就可以認為操作成功的那個線程獲得了該方法的鎖，可以執行方法體內容。
當方法執行完畢之后，想要釋放鎖的話，需要執行以下Sql:
上面這種簡單的實現有以下幾個問題：
1、這把鎖強依賴數據庫的可用性，數據庫是一個單點，一旦數據庫掛掉，會導致業務系統不可用。
2、這把鎖沒有失效時間，一旦解鎖操作失敗，就會導致鎖記錄一直在數據庫中，其他線程無法再獲得到鎖。
3、這把鎖只能是非阻塞的，因為數據的insert操作，一旦插入失敗就會直接報錯。沒有獲得鎖的線程並不會進入排隊隊列，要想再次獲得鎖就要再次觸發獲得鎖操作。
4、這把鎖是非重入的，同一個線程在沒有釋放鎖之前無法再次獲得該鎖。因為數據中數據已經存在了。
當然，我們也可以有其他方式解決上面的問題。
- 數據庫是單點？搞兩個數據庫，數據之前雙向同步。一旦掛掉快速切換到備庫上。
- 沒有失效時間？只要做一個定時任務，每隔一定時間把數據庫中的超時數據清理一遍。
- 非阻塞的？搞一個while循環，直到insert成功再返回成功。
- 非重入的？在數據庫表中加個字段，記錄當前獲得鎖的機器的主機信息和線程信息，那么下次再獲取鎖的時候先查詢數據庫，如果當前機器的主機信息和線程信息在數據庫可以查到的話，直接把鎖分配給他就可以了。
基於數據庫排他鎖
除了可以通過增刪操作數據表中的記錄以外，其實還可以借助數據中自帶的鎖來實現分布式的鎖。
我們還用剛剛創建的那張數據庫表。可以通過數據庫的排他鎖來實現分布式鎖。 基於MySql的InnoDB引擎，可以使用以下方法來實現加鎖操作：
在查詢語句后面增加for update，數據庫會在查詢過程中給數據庫表增加排他鎖（這里再多提一句，InnoDB引擎在加鎖的時候，只有通過索引進行檢索的時候才會使用行級鎖，否則會使用表級鎖。這里我們希望使用行級鎖，就要給method_name添加索引，值得注意的是，這個索引一定要創建成唯一索引，否則會出現多個重載方法之間無法同時被訪問的問題。重載方法的話建議把參數類型也加上。）。當某條記錄被加上排他鎖之后，其他線程無法再在該行記錄上增加排他鎖。
我們可以認為獲得排它鎖的線程即可獲得分布式鎖，當獲取到鎖之后，可以執行方法的業務邏輯，執行完方法之后，再通過以下方法解鎖：
通過connection.commit()操作來釋放鎖。
這種方法可以有效的解決上面提到的無法釋放鎖和阻塞鎖的問題。
- 阻塞鎖？
for update語句會在執行成功后立即返回，在執行失敗時一直處於阻塞狀態，直到成功。 - 鎖定之后服務宕機，無法釋放？使用這種方式，服務宕機之后數據庫會自己把鎖釋放掉。
但是還是無法直接解決數據庫單點和可重入問題。
這里還可能存在另外一個問題，雖然我們對method_name 使用了唯一索引，並且顯示使用for update來使用行級鎖。但是，MySql會對查詢進行優化，即便在條件中使用了索引字段，但是否使用索引來檢索數據是由 MySQL 通過判斷不同執行計划的代價來決定的，如果 MySQL 認為全表掃效率更高，比如對一些很小的表，它就不會使用索引，這種情況下 InnoDB 將使用表鎖，而不是行鎖。如果發生這種情況就悲劇了。。。
還有一個問題，就是我們要使用排他鎖來進行分布式鎖的lock，那么一個排他鎖長時間不提交，就會占用數據庫連接。一旦類似的連接變得多了，就可能把數據庫連接池撐爆
總結
總結一下使用數據庫來實現分布式鎖的方式，這兩種方式都是依賴數據庫的一張表，一種是通過表中的記錄的存在情況確定當前是否有鎖存在，另外一種是通過數據庫的排他鎖來實現分布式鎖。
數據庫實現分布式鎖的優點
直接借助數據庫，容易理解。
數據庫實現分布式鎖的缺點
會有各種各樣的問題，在解決問題的過程中會使整個方案變得越來越復雜。
操作數據庫需要一定的開銷，性能問題需要考慮。
使用數據庫的行級鎖並不一定靠譜，尤其是當我們的鎖表並不大的時候。
基於緩存實現分布式鎖
相比較於基於數據庫實現分布式鎖的方案來說，基於緩存來實現在性能方面會表現的更好一點。而且很多緩存是可以集群部署的，可以解決單點問題。
目前有很多成熟的緩存產品，包括Redis，memcached以及我們公司內部的Tair。
這里以Tair為例來分析下使用緩存實現分布式鎖的方案。關於Redis和memcached在網絡上有很多相關的文章，並且也有一些成熟的框架及算法可以直接使用。
基於Tair的實現分布式鎖其實和Redis類似，其中主要的實現方式是使用TairManager.put方法來實現。
以上實現方式同樣存在幾個問題：
1、這把鎖沒有失效時間，一旦解鎖操作失敗，就會導致鎖記錄一直在tair中，其他線程無法再獲得到鎖。
2、這把鎖只能是非阻塞的，無論成功還是失敗都直接返回。
3、這把鎖是非重入的，一個線程獲得鎖之后，在釋放鎖之前，無法再次獲得該鎖，因為使用到的key在tair中已經存在。無法再執行put操作。
當然，同樣有方式可以解決。
- 沒有失效時間？tair的put方法支持傳入失效時間，到達時間之后數據會自動刪除。
- 非阻塞？while重復執行。
- 非可重入？在一個線程獲取到鎖之后，把當前主機信息和線程信息保存起來，下次再獲取之前先檢查自己是不是當前鎖的擁有者。
但是，失效時間我設置多長時間為好？如何設置的失效時間太短，方法沒等執行完，鎖就自動釋放了，那么就會產生並發問題。如果設置的時間太長，其他獲取鎖的線程就可能要平白的多等一段時間。這個問題使用數據庫實現分布式鎖同樣存在
總結
可以使用緩存來代替數據庫來實現分布式鎖，這個可以提供更好的性能，同時，很多緩存服務都是集群部署的，可以避免單點問題。並且很多緩存服務都提供了可以用來實現分布式鎖的方法，比如Tair的put方法，redis的setnx方法等。並且，這些緩存服務也都提供了對數據的過期自動刪除的支持，可以直接設置超時時間來控制鎖的釋放。
使用緩存實現分布式鎖的優點
性能好，實現起來較為方便。
使用緩存實現分布式鎖的缺點
通過超時時間來控制鎖的失效時間並不是十分的靠譜。
基於Zookeeper實現分布式鎖
基於zookeeper臨時有序節點可以實現的分布式鎖。
讓我們來回顧一下Zookeeper節點的概念：
Zookeeper的數據存儲結構就像一棵樹，這棵樹由節點組成，這種節點叫做Znode。
Znode分為四種類型：
1.持久節點 （PERSISTENT）
默認的節點類型。創建節點的客戶端與zookeeper斷開連接后，該節點依舊存在 。
2.持久節點順序節點（PERSISTENT_SEQUENTIAL）
所謂順序節點，就是在創建節點時，Zookeeper根據創建的時間順序給該節點名稱進行編號：
3.臨時節點（EPHEMERAL）
和持久節點相反，當創建節點的客戶端與zookeeper斷開連接后，臨時節點會被刪除：
4.臨時順序節點（EPHEMERAL_SEQUENTIAL）
顧名思義，臨時順序節點結合和臨時節點和順序節點的特點：在創建節點時，Zookeeper根據創建的時間順序給該節點名稱進行編號；當創建節點的客戶端與zookeeper斷開連接后，臨時節點會被刪除。
Zookeeper分布式鎖的原理
Zookeeper分布式鎖恰恰應用了臨時順序節點。具體如何實現呢？讓我們來看一看詳細步驟：
獲取鎖
首先，在Zookeeper當中創建一個持久節點ParentLock。當第一個客戶端想要獲得鎖時，需要在ParentLock這個節點下面創建一個臨時順序節點 Lock1。
之后，Client1查找ParentLock下面所有的臨時順序節點並排序，判斷自己所創建的節點Lock1是不是順序最靠前的一個。如果是第一個節點，則成功獲得鎖。
這時候，如果再有一個客戶端 Client2 前來獲取鎖，則在ParentLock下載再創建一個臨時順序節點Lock2。
Client2查找ParentLock下面所有的臨時順序節點並排序，判斷自己所創建的節點Lock2是不是順序最靠前的一個，結果發現節點Lock2並不是最小的。
於是，Client2向排序僅比它靠前的節點Lock1注冊Watcher，用於監聽Lock1節點是否存在。這意味着Client2搶鎖失敗，進入了等待狀態。
這時候，如果又有一個客戶端Client3前來獲取鎖，則在ParentLock下載再創建一個臨時順序節點Lock3。
Client3查找ParentLock下面所有的臨時順序節點並排序，判斷自己所創建的節點Lock3是不是順序最靠前的一個，結果同樣發現節點Lock3並不是最小的。
於是，Client3向排序僅比它靠前的節點Lock2注冊Watcher，用於監聽Lock2節點是否存在。這意味着Client3同樣搶鎖失敗，進入了等待狀態。
這樣一來，Client1得到了鎖，Client2監聽了Lock1，Client3監聽了Lock2。這恰恰形成了一個等待隊列，很像是Java當中ReentrantLock所依賴的
釋放鎖
釋放鎖分為兩種情況：
1.任務完成，客戶端顯示釋放
當任務完成時，Client1會顯示調用刪除節點Lock1的指令。
2.任務執行過程中，客戶端崩潰
獲得鎖的Client1在任務執行過程中，如果Duang的一聲崩潰，則會斷開與Zookeeper服務端的鏈接。根據臨時節點的特性，相關聯的節點Lock1會隨之自動刪除。
由於Client2一直監聽着Lock1的存在狀態，當Lock1節點被刪除，Client2會立刻收到通知。這時候Client2會再次查詢ParentLock下面的所有節點，確認自己創建的節點Lock2是不是目前最小的節點。如果是最小，則Client2順理成章獲得了鎖。
同理，如果Client2也因為任務完成或者節點崩潰而刪除了節點Lock2，那么Client3就會接到通知。
最終，Client3成功得到了鎖。
大致思想即為：每個客戶端對某個方法加鎖時，在zookeeper上的與該方法對應的指定節點的目錄下，生成一個唯一的瞬時有序節點。 判斷是否獲取鎖的方式很簡單，只需要判斷有序節點中序號最小的一個。 當釋放鎖的時候，只需將這個瞬時節點刪除即可。同時，其可以避免服務宕機導致的鎖無法釋放，而產生的死鎖問題。
來看下Zookeeper能不能解決前面提到的問題。
-
鎖無法釋放？使用Zookeeper可以有效的解決鎖無法釋放的問題，因為在創建鎖的時候，客戶端會在ZK中創建一個臨時節點，一旦客戶端獲取到鎖之后突然掛掉（Session連接斷開），那么這個臨時節點就會自動刪除掉。其他客戶端就可以再次獲得鎖。
-
非阻塞鎖？使用Zookeeper可以實現阻塞的鎖，客戶端可以通過在ZK中創建順序節點，並且在節點上綁定監聽器，一旦節點有變化，Zookeeper會通知客戶端，客戶端可以檢查自己創建的節點是不是當前所有節點中序號最小的，如果是，那么自己就獲取到鎖，便可以執行業務邏輯了。
-
不可重入？使用Zookeeper也可以有效的解決不可重入的問題，客戶端在創建節點的時候，把當前客戶端的主機信息和線程信息直接寫入到節點中，下次想要獲取鎖的時候和當前最小的節點中的數據比對一下就可以了。如果和自己的信息一樣，那么自己直接獲取到鎖，如果不一樣就再創建一個臨時的順序節點，參與排隊。
-
單點問題？使用Zookeeper可以有效的解決單點問題，ZK是集群部署的，只要集群中有半數以上的機器存活，就可以對外提供服務。
可以直接使用zookeeper第三方庫Curator客戶端，這個客戶端中封裝了一個可重入的鎖服務。
Curator提供的InterProcessMutex是分布式鎖的實現。acquire方法用戶獲取鎖，release方法用於釋放鎖。
使用ZK實現的分布式鎖好像完全符合了本文開頭我們對一個分布式鎖的所有期望。但是，其實並不是，Zookeeper實現的分布式鎖其實存在一個缺點，那就是性能上可能並沒有緩存服務那么高。因為每次在創建鎖和釋放鎖的過程中，都要動態創建、銷毀瞬時節點來實現鎖功能。ZK中創建和刪除節點只能通過Leader服務器來執行，然后將數據同不到所有的Follower機器上。
其實，使用Zookeeper也有可能帶來並發問題，只是並不常見而已。考慮這樣的情況，由於網絡抖動，客戶端可ZK集群的session連接斷了，那么zk以為客戶端掛了，就會刪除臨時節點，這時候其他客戶端就可以獲取到分布式鎖了。就可能產生並發問題。這個問題不常見是因為zk有重試機制，一旦zk集群檢測不到客戶端的心跳，就會重試，Curator客戶端支持多種重試策略。多次重試之后還不行的話才會刪除臨時節點。（所以，選擇一個合適的重試策略也比較重要，要在鎖的粒度和並發之間找一個平衡。）
總結
使用Zookeeper實現分布式鎖的優點
有效的解決單點問題，不可重入問題，非阻塞問題以及鎖無法釋放的問題。實現起來較為簡單。
使用Zookeeper實現分布式鎖的缺點
性能上不如使用緩存實現分布式鎖。 需要對ZK的原理有所了解。
三種方案的比較
上面幾種方式，哪種方式都無法做到完美。就像CAP一樣，在復雜性、可靠性、性能等方面無法同時滿足，所以，根據不同的應用場景選擇最適合自己的才是王道。
從理解的難易程度角度（從低到高）
數據庫 > 緩存 > Zookeeper
從實現的復雜性角度（從低到高）
Zookeeper >= 緩存 > 數據庫
從性能角度（從高到低）
緩存 > Zookeeper >= 數據庫
從可靠性角度（從高到低）
Zookeeper > 緩存 > 數據庫