分散式鎖

分散式系統 Jedis Lua · 發表 2019-01-31 14:33:00

摘要：單機方案比較多，synchronized和juc很豐富分散式鎖互斥性:在任意時刻，只有一個客戶端能持有鎖不會發生死鎖:即有一個客戶端在持有鎖的期間崩潰而沒有主動解鎖，也能保證後續其他客戶端能加鎖文章來源：https://www....

單機

方案比較多，synchronized和juc很豐富

分散式鎖

互斥性:在任意時刻，只有一個客戶端能持有鎖
不會發生死鎖:即有一個客戶端在持有鎖的期間崩潰而沒有主動解鎖，也能保證後續其他客戶端能加鎖

文章來源：https://www.cnblogs.com/guozp/p/10341337.html

常見方案

基於資料庫
基於分散式快取（redis、tair等）
基於zk
要基於你的業務場景選擇合適方案

資料庫(mysql)

基於資料庫的ACID以及MVCC(多版本併發控制機)，MVCC是通過儲存資料在某個時間點的快照來實現的，不同儲存引擎的MVCC實現是不同的,典型的有樂觀併發控制和悲觀併發控制

基於悲觀鎖（for update）

select * from table where *** for update
基於樂觀鎖（version）

樂觀鎖是基於資料的版本號實現的,表增加一個欄位version，每次讀取的時候，將version取出，更新的時候，比較version是否一致，一致，處理完後把version加1；不一致，本次未拿到鎖
- 表定義（根據需求增加）
  
  id resource status expire version
  
  1 1 2 2019-01-01 12:00:00 1
  
  2 2 2 2019-01-01 12:00:01 1
- 含義
  - resource：代表資源
  - status:鎖定狀態
  - expire:過期時間,根據需求看是否需要增加使用
- 執行流程：
  1. 執行查詢操作獲取當前資料的資料版本號,例如：select id, resource, state,version from table where state=1 and id=1;
  2. 執行更新：update table set state=2, version=上次+1 where resource=1 and state=1 and version=1
  3. 上述執行影響1行，加鎖成功，影響0行，自己加鎖失敗，其它人已經加鎖鎖定

id	resource	status	expire	version
1	1	2	2019-01-01 12:00:00	1
2	2	2	2019-01-01 12:00:01	1

tair

Tair沒有直接提供分散式鎖的api，但是可以藉助提供的其他api實現分散式鎖。

incr/decr(不可重入鎖)
- 原理：通過計數api的上下限值約束來實現(增加/減少計數。可設定最大值和最小值)
- api:
  1. 增加計數（加鎖）：
    Result<Integer> incr(int namespace, Serializable key, int value, int defaultValue, int expireTime, int lowBound, int upperBound)
  2. 減少計數（釋放鎖）：
```
Result<Integer> decr(int namespace, Serializable key, int value, int defaultValue, int expireTime, int lowBound, int upperBound)
```
  1. 關鍵引數解釋：
  defaultValue: 第一次呼叫incr時的key的count初始值，第一次返回的值為defaultValue + value, decr第一次返回defaultValue - valuelowBound: 最小值upperBound: 最大值
使用
1. 執行緒一呼叫incr加鎖,加鎖後，key的值變成1，而key的上限值為1，其他執行緒再呼叫該介面時會報錯COUNTER_OUT_OF_RANGE
2. 待執行緒一使用完成後，呼叫decr解鎖,此時key已經有值1，返回 1-1=0，解鎖成功。多次呼叫會失敗，因為範圍是0~1。
3. 通過0、1的來回變化，達到分散式鎖的目的，當key為1時獲取到鎖，為0時釋放鎖

Get/Put

原理：使用put的version校驗實現

api

ResultCode put(int namespace, Serializable key, Serializable value, int version, int expireTime)`

一定要設定過期引數expireTime，否則鎖執行過程中程序crash，鎖不會釋放，會長期佔有，影響業務，加上後，業務至少可以自行恢復

關鍵引數解釋：

version - 為了解決併發更新同一個資料而設定的引數。當version為0時，表示強制更新
這裡注意：
此處version，除了0、1外的任何數字都可以，傳入0，tair會強制覆蓋；而傳入1，第一個client寫入會成功，但是新寫入時服務端的version以0開始計數啊，所以此時version也是1，所以下一個到來的client寫入也會成功，這樣造成了衝突。

實現

這裡針對網路等問題做了重試，同時改造支援可重入鎖，不可重入鎖，目前這裡可重入沒有做計數以及重新設定過期時間，使用的各位可以根據實際情況進行改造

@Override
public boolean tryLock(String lockKey, int expireTime, boolean reentrant) {
if (expireTime <= 0) {
expireTime = DEFAULT_EXPIRE_TIME;
}
int retryGet = 0;
try {
Result<DataEntry> result = tairManager.get(NAMESPACE, lockKey);
while (retryGet++ < LOCK_GET_MAX_RETRY && result != null && isError(result.getRc())) {
result = tairManager.get(NAMESPACE, lockKey);
}
if (result == null) {
log.error("tryLock error, maybe Tair service is unavailable");
return false;
}
if (ResultCode.DATANOTEXSITS.equals(result.getRc())) {
// version 2表示為空，若不是為空，則返回version error
ResultCode code = tairManager.put(NAMESPACE, lockKey, getLockValue(), DEFAULT_VERSION, expireTime);
if (ResultCode.SUCCESS.equals(code)) {
return true;
} else if (retryPut.get() < LOCK_PUT_MAX_RETRY && isError(code)) {
retryPut.set(retryPut.get() + 1);
return tryLock(lockKey, expireTime);
}
} else if (reentrant && result.getValue() != null && getLockValue().equals(result.getValue().getValue())) {
return true;
}
} catch (Exception e) {
log.error("try lock is error, msg is {}", e);
} finally {
retryPut.remove();
}
return false;
}

@Override
public void unlock(String lockKey) {
unlock(lockKey, false);
}

@Override
public boolean unlock(String lockKey, boolean reentrant) {
if (!reentrant) {
ResultCode invalid = tairManager.invalid(NAMESPACE, lockKey);
return invalid != null && invalid.isSuccess();
}
Result<DataEntry> result = tairManager.get(NAMESPACE, lockKey);
if (result != null && result.isSuccess() && result.getValue() != null) {
String value = result.getValue().getValue().toString();
if (getLockValue().equals(value)) {
ResultCode rc = tairManager.invalid(NAMESPACE, lockKey);
if (rc != null && rc.isSuccess()) {
return true;
} else {
log.error("unlock failed, tairLockManager.invalidValue fail, key is {}, ResultCode is {}",
lockKey, rc);
return false;
}
} else {
log.warn("unlock failed,value is not equal lockValue, key is {}, lockValue is {}, value is {}",
lockKey, getLockValue(), value);
return false;
}
}
return false;
}

@Override
public boolean lockStatus(String lockKey) {
Result<DataEntry> result = tairManager.get(NAMESPACE, lockKey);
if (result != null && result.isSuccess() && result.getValue() != null) {
return true;
}
return false;
}

private boolean isError(ResultCode code) {
return code == null || ResultCode.CONNERROR.equals(code) || ResultCode.TIMEOUT.equals(code) || ResultCode.UNKNOW
.equals(code);
}

private String getLockValue() {
return NetUtils.getLocalIp() + "_" + Thread.currentThread().getName();
}

redis

正確的加鎖邏輯
- API：
  1. 加鎖
    SET key value [EX seconds] [PX milliseconds] [NX|XX]
  2. 釋放鎖
    EVAL script numkeys key [key ...] arg [arg ...]
- 關鍵引數解釋
  
  加鎖
```
``
EX second ：設定鍵的過期時間為 second 秒。 SET key value EX second 效果等同於 SETEX key second value
PX millisecond ：設定鍵的過期時間為 millisecond 毫秒。SET key value PXmillisecond 效果等同於 PSETEX key millisecond value 
NX ：只在鍵不存在時，才對鍵進行設定操作。 SET key value NX 效果等同於 SETNX key value
XX ：只在鍵已經存在時，才對鍵進行設定操作。
```
```
> 釋放
```
  script 引數是一段 Lua 5.1 指令碼程式，它會被執行在 Redis 伺服器上下文中，這段指令碼不必(也不應該)定義為一個 Lua 函式。
  numkeys 引數用於指定鍵名引數的個數。
  鍵名引數 key [key ...] 從 EVAL 的第三個引數開始算起，表示在指令碼中所用到的那些 Redis 鍵(key)，這些鍵名引數可以在 Lua 中通過全域性變數 KEYS 陣列，用 1 為基址的形式訪問( KEYS[1] ， KEYS[2] ，以此類推)。
  在命令的最後，那些不是鍵名引數的附加引數 arg [arg ...] ，可以在 Lua 中通過全域性變數 ARGV 陣列訪問，訪問的形式和 KEYS 變數類似( ARGV[1] 、 ARGV[2] ，諸如此類)。
  
```
- 實現
```
/**
*1. 當前沒有鎖（key不存在），那麼就進行加鎖操作，並對鎖設定個有效期，同時value表示加鎖的客戶端。2. 已有鎖存在，不做任何操作
**/
public boolean tryLock(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;

}

public boolean unlock(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;
}
```
  - 首先，set()加入了NX引數，可以保證如果key已存在，則函式不會呼叫成功，即只有一個客戶端能持有鎖。其次，由於我們對鎖設定了過期時間，即使鎖的持有者後續發生crash而沒有解鎖，鎖也會因為到了過期時間而自動解鎖（即key被刪除），不會發生死鎖。最後，因為我們將value賦值為requestId，代表加鎖的客戶端請求標識，那麼在客戶端在解鎖的時候就可以進行校驗是否是同一個客戶端。
  - 釋放鎖，這段Lua程式碼的功能：首先獲取鎖對應的value值，檢查是否與requestId相等，如果相等則刪除鎖（解鎖）。那麼為什麼要使用Lua語言來實現呢？因為lua可以確保上述操作是原子性的 。

tair的rdb引擎目前不支援上述命令，所以需要寫成兩行命令（或許新版本支援了，因為我使用的的還是舊版本,所以rdb的實現方式：

支援可重入鎖，不可重入鎖，目前這裡可重入沒有做計數以及重新設定過期時間，使用的各位可以根據實際情況進行改造

/**
* rdb 不支援多引數，所以使用兩個命令
*
* @param lockKey
* @param expireTime 超時時間
* @param reentrant是否可重入,重入後會延長時間
* @return
*/
@Override
public boolean tryLock(String lockKey, int expireTime, boolean reentrant) {
if (expireTime <= 0) {
expireTime = DEFAULT_EXPIRE_TIME;
}
boolean result = redisRepo.setNx(lockKey, getLockValue(), expireTime);
if (!reentrant) {
return result;
}
String value = redisRepo.get(lockKey);
if (getLockValue().equals(value)) {
result = redisRepo.setNx(lockKey, getLockValue(), expireTime);
}
return result;
}

/**
* 版本不支援lua，所以使用兩個命令
*
* @param lockKey
* @param reentrant 是否可以釋放其它人建立的鎖
* @return
*/
@Override
public boolean unlock(String lockKey, boolean reentrant) {
if (!reentrant) {
return redisRepo.delKeys(lockKey) > 0;
}
long result = 0;
String value = redisRepo.get(lockKey);
if (getLockValue().equals(value)) {
result = redisRepo.delKeys(lockKey);
}
return result > 0;
}

@Override
public boolean lockStatus(String lockKey) {
String value = redisRepo.get(lockKey);
return StringUtils.isNotBlank(value);
}

private String getLockValue() {
return NetUtils.getLocalIp() + "_" + Thread.currentThread().getName();
}

錯誤的加鎖示例
1. setnx()方法作用就是SET IF NOT EXIST，expire()方法就是給鎖加一個過期時間。乍一看好像和前面的set()方法結果一樣，但是由於這是兩條Redis命令，不具有原子性，如果程式在執行完setnx()之後crash，由於鎖沒有設定過期時間，將會發生死鎖
```
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);
}

}
```
2. 1. 通過setnx()方法嘗試加鎖，如果當前鎖不存在，返回加鎖成功。
  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.getSet()方法，那麼雖然最終只有一個客戶端可以加鎖，但是這個客戶端的鎖的過期時間可能被其他客戶端覆蓋。
 * 鎖不具備擁有者標識，即任何客戶端都可以解鎖（看個人業務）
```

錯誤的鎖釋放示例

使用jedis.del()方法刪除鎖，這種不先判斷鎖的擁有者而直接解鎖的方式，會導致任何客戶端都可以隨時進行解鎖

``
public static void wrongReleaseLock1(Jedis jedis, String lockKey) {
jedis.del(lockKey);
}

```

以下程式碼分成兩條命令去執行，如果呼叫jedis.del()的時候，鎖已經不屬於當前客戶端的時,會解除他人加的鎖

public static void wrongReleaseLock2(Jedis jedis, String lockKey, String requestId) {

// 判斷加鎖與解鎖是不是同一個客戶端
if (requestId.equals(jedis.get(lockKey))) {
// 若在此時，這把鎖過期不屬於這個客戶端的，則會誤解鎖
jedis.del(lockKey);
} 
}

redis官方鎖

Redis的官方曾提出了一個容錯的分散式鎖演算法：RedLock，只要有超過一半的快取伺服器能夠正常工作，系統就可以保證分散式鎖的可用性。詳情參考

zk

有機會或者留言需要的在寫吧,略略略

文章來源：https://www.cnblogs.com/guozp/p/10341337.html

方案比較（從低到高）

從理解的難易程度角度：資料庫 > 快取 > Zookeeper
從實現的複雜性角度：Zookeeper >= 快取 > 資料庫
從效能角度：快取 > Zookeeper >= 資料庫
從可靠性角度：Zookeeper > 快取 > 資料庫

分散式鎖

單機

分散式鎖

常見方案

資料庫(mysql)

tair

redis

redis官方鎖

zk

方案比較（從低到高）

您可能也會喜歡…