Redis 實現(xiàn)分布式鎖真的安全嗎？

什么是分布式鎖

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

定義：

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

特點：

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

實現(xiàn)方式：

• 數(shù)據(jù)庫鎖
• 基于Redis的分布式鎖
• 基于Zookeeper的分布式鎖

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

代碼示例

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

參數(shù)解釋：

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

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

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

?*?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 已經(jīng)拿到了鎖，當(dāng)前時間：%s",?Thread.currentThread().getName(),?System.currentTimeMillis()?/?1000));
????}
????return?lock;
}

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

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

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;
}

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

Redisson實現(xiàn)分布式鎖

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

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

執(zhí)行步驟：

1、第一步，客戶端獲取當(dāng)前時間。

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

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

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

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

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

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

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

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

代碼示例：

首先引入Redisson依賴的Jar包

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

Redisson 支持3種方式連接redis，分別為單機(jī)、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è)務(wù)處理
????????doBusiness();
????}
}catch(exception?e){
}finally{
????lock.unlock();
}

項目源碼地址

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

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