Redis 實現分布式鎖真的安全嗎？

什么是分布式鎖

隨著互聯網業(yè)務快速發(fā)展，軟件架構開始向分布式集群演化。由于分布式系統的多線程分布在不同的服務器上，為了跨JVM控制全局共享資源的訪問，于是誕生了分布式鎖。

定義：

分布式鎖是控制分布式系統之間同步訪問共享資源的一種方式。在分布式系統中，常常需要協調他們的動作，如果不同的系統或是同一個系統的不同服務器之間共享了一個或一組資源，那么訪問這些資源的時候，往往需要互斥來防止彼此干擾來保證一致性，在這種情況下，便需要使用到分布式鎖。

特點：

• 互斥性。任意時刻，只有一個客戶端能持有鎖。
• 鎖超時。和本地鎖一樣支持鎖超時，防止死鎖
• 高可用。加鎖和解鎖要保證性能，同時也需要保證高可用防止分布式鎖失效，可以增加降級。
• 支持阻塞和非阻塞。和ReentrantLock 一樣支持 lock 和 trylock 以及 tryLock(long timeOut)。

實現方式：

• 數據庫鎖
• 基于Redis的分布式鎖
• 基于Zookeeper的分布式鎖

考慮到性能要求，一般采用redis來實現分布式鎖。另外，在實際的業(yè)務應用中，如果你想要提升分布式鎖的可靠性，可以通過Redlock 算法來實現。

代碼示例

通過redis原子命令 set key value [NX|XX] [EX seconds | PX milliseconds] 來是實現加鎖操作。

參數解釋：

• EX seconds：設置失效時長，單位秒
• PX milliseconds：表示這個 key 的存活時間，稱作鎖過期時間，單位毫秒。當資源被鎖定超過這個時間時，鎖將自動釋放。
• NX：key不存在時設置value，成功返回OK，失敗返回(nil)
• XX：key存在時設置value，成功返回OK，失敗返回(nil)
• value：必須是全局唯一的值。這個隨機數在釋放鎖時保證釋放鎖操作的安全性。

原理：只有在某個 key 不存在的情況下才能設置（set）成功該 key。于是，這就可以讓多個線程并發(fā)去設置同一個 key，只有一個線程能設置成功。而其它的線程因為之前有人把 key 設置成功了，而導致失?。ㄒ簿褪谦@得鎖失敗）。

/**
?*?獲取鎖
?*?

?*?true：成功獲取鎖
?*?false：本次請求沒有拿到鎖
?*/
public?boolean?lock(String?key,?String?value,?long?expireTime)?{
????key?=?prefixKey?+?key;

????boolean?lock?=?false;
????try?{
????????lock?=?redisTemplate.opsForValue().setIfAbsent(key,?value,?expireTime,?TimeUnit.MILLISECONDS);
????}?catch?(Exception?e)?{
????????e.printStackTrace();
????????lock?=?false;
????}

????if?(lock)?{
????????System.out.println(String.format("%s 已經拿到了鎖，當前時間：%s",?Thread.currentThread().getName(),?System.currentTimeMillis()?/?1000));
????}
????return?lock;
}

分布式鎖使用結束后需要手動來釋放鎖。可以直接通過 del 命令刪除key即可，但是從高可用性上講，如果業(yè)務的執(zhí)行時間超過了鎖釋放的時間，導致 redis 中的key 自動超時過期，鎖被動釋放。然后被其他線程競爭獲取了鎖，此時之前的線程再釋放的就是別人的鎖，會引發(fā)混亂。

為了避免該問題，我們通過lua腳本，在釋放鎖時，先進行值比較判斷，只能釋放自己的鎖！??！

public?boolean?unLock(String?key,?String?value)?{
????key?=?prefixKey?+?key;
????Long?result?=?-1L;
????String?luaScript?=
????????????"if?redis.call('get',?KEYS[1])?==?ARGV[1]?then?"?+
?????????????"??return?redis.call('del',?KEYS[1])?"?+
?????????????"else?"?+
?????????????"??return?0?"?+
?????????????"end";

????DefaultRedisScript?redisScript?=?new?DefaultRedisScript(luaScript,?Long.class);
????try?{
????????//?del?成功返回?1
????????result?=?(Long)?redisTemplate.execute(redisScript,?Lists.list(key),?value);
????????//?System.out.println(result);
????}?catch?(Exception?e)?{
????????e.printStackTrace();

????}
????return?result?==?1???true?:?false;
}

在這種場景（主從結構）中存在明顯的競態(tài): 客戶端A從master獲取到鎖， 在master將鎖同步到slave之前，master宕掉了。slave節(jié)點被晉升為新的master節(jié)點， 客戶端B取得了同一個資源被客戶端A已經獲取到的另外一個鎖。「安全失效」！

Redisson實現分布式鎖

為了避免 Redis 實例故障而導致的鎖無法工作的問題，Redis 的開發(fā)者 Antirez 提出了分布式鎖算法Redlock。

Redlock 算法的基本思路，是讓客戶端和多個獨立的 Redis 實例依次請求加鎖，如果客戶端能夠和半數以上的實例成功地完成加鎖操作，那么我們就認為，客戶端成功地獲得分布式鎖了，否則加鎖失敗。這樣一來，即使有單個 Redis 實例發(fā)生故障，因為鎖變量在其它實例上也有保存，所以，客戶端仍然可以正常地進行鎖操作。

執(zhí)行步驟：

1、第一步，客戶端獲取當前時間。

2、第二步，客戶端按順序依次向 N 個 Redis 實例執(zhí)行加鎖操作。

這里的加鎖操作和在單實例上執(zhí)行的加鎖操作一樣，使用 SET 命令，帶上 NX，EX/PX 選項，以及帶上客戶端的唯一標識。當然，如果某個 Redis 實例發(fā)生故障了，為了保證在這種情況下，Redlock 算法能夠繼續(xù)運行，我們需要給加鎖操作設置一個超時時間。

如果客戶端在和一個 Redis 實例請求加鎖時，一直到超時都沒有成功，那么此時，客戶端會和下一個 Redis 實例繼續(xù)請求加鎖。加鎖操作的超時時間需要遠遠地小于鎖的有效時間，一般為幾十毫秒。

3、第三步，一旦客戶端完成了和所有 Redis 實例的加鎖操作，客戶端就要計算整個加鎖過程的總耗時。只有在滿足下面的這兩個條件時，才能認為是加鎖成功。

• 條件一：客戶端從超過半數（大于等于 N/2+1）的 Redis 實例上成功獲取到了鎖；
• 條件二：客戶端獲取鎖的總耗時沒有超過鎖的有效時間。

在滿足了這兩個條件后，我們需要重新計算這把鎖的有效時間，計算的結果是鎖的最初有效時間減去客戶端為獲取鎖的總耗時。如果鎖的有效時間已經來不及完成共享數據的操作了，我們可以釋放鎖，以免出現還沒完成數據操作，鎖就過期了的情況。

當然，如果客戶端在和所有實例執(zhí)行完加鎖操作后，沒能同時滿足這兩個條件，那么，客戶端向所有 Redis 節(jié)點發(fā)起釋放鎖的操作。

在 Redlock 算法中，釋放鎖的操作和在單實例上釋放鎖的操作一樣，只要執(zhí)行釋放鎖的 Lua 腳本就可以了。這樣一來，只要 N 個 Redis 實例中的半數以上實例能正常工作，就能保證分布式鎖的正常工作了。

代碼示例：

首先引入Redisson依賴的Jar包

????org.redisson
????redisson
????3.9.1

Redisson 支持3種方式連接redis，分別為單機、Sentinel 哨兵、Cluster 集群，項目中使用的連接方式是 Sentinel。

Sentinel配置，首先創(chuàng)建RedissonClient客戶端實例

Config?config?=?new?Config();
config.useSentinelServers().addSentinelAddress("127.0.0.1:6479",?"127.0.0.1:6489").setMasterName("master").setPassword("password").setDatabase(0);
RedissonClient?redisson?=?Redisson.create(config);

加鎖、釋放鎖

RLock?lock?=?redisson.getLock("test_lock");
try{
????boolean?isLock=lock.tryLock();
????if(isLock){
????????//?模擬業(yè)務處理
????????doBusiness();
????}
}catch(exception?e){
}finally{
????lock.unlock();
}

項目源碼地址

https://github.com/aalansehaiyang/spring-boot-bulking??

模塊：spring-boot-bulking-redis-lock