分布式鎖在分布式應(yīng)用中應(yīng)用廣泛，想要搞懂一個新事物首先得了解它的由來，這樣才能更加的理解甚至可以舉一反三。

首先談到分布式鎖自然也就聯(lián)想到分布式應(yīng)用。

在我們將應(yīng)用拆分為分布式應(yīng)用之前的單機(jī)系統(tǒng)中，對一些并發(fā)場景讀取公共資源時如扣庫存，賣車票之類的需求可以簡單的使用同步或者是加鎖就可以實現(xiàn)。

但是應(yīng)用分布式了之后系統(tǒng)由以前的單進(jìn)程多線程的程序變?yōu)榱硕噙M(jìn)程多線程，這時使用以上的解決方案明顯就不夠了。

因此業(yè)界常用的解決方案通常是借助于一個第三方組件并利用它自身的排他性來達(dá)到多進(jìn)程的互斥。如：

基于 DB 的唯一索引。

基于 ZK 的臨時有序節(jié)點。

基于 Redis 的 NX EX 參數(shù)。

這里主要基于 Redis 進(jìn)行討論。

既然是選用了 Redis，那么它就得具有排他性才行。同時它最好也有鎖的一些基本特性：

高性能(加、解鎖時高性能)

可以使用阻塞鎖與非阻塞鎖。

不能出現(xiàn)死鎖。

可用性(不能出現(xiàn)節(jié)點 down 掉后加鎖失敗)。

這里利用 Redis set key 時的一個 NX 參數(shù)可以保證在這個 key 不存在的情況下寫入成功。并且再加上 EX 參數(shù)可以讓該 key 在超時之后自動刪除。

所以利用以上兩個特性可以保證在同一時刻只會有一個進(jìn)程獲得鎖，并且不會出現(xiàn)死鎖(最壞的情況就是超時自動刪除 key)。

實現(xiàn)代碼如下：

 private html' target='_blank'>static final String SET_IF_NOT_EXIST = NX  private static final String SET_WITH_EXPIRE_TIME = PX  public boolean tryLock(String key, String request) { String result = this.jedis.set(LOCK_PREFIX + key, request, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, 10 * TIME); if (LOCK_MSG.equals(result)){ return true ; }else { return false ; }

注意這里使用的 jedis 的

String set(String key, String value, String nxxx, String expx, long time);

api。

該命令可以保證 NX EX 的原子性。

一定不要把兩個命令(NX EX)分開執(zhí)行，如果在 NX 之后程序出現(xiàn)問題就有可能產(chǎn)生死鎖。

同時也可以實現(xiàn)一個阻塞鎖：

 //一直阻塞 public void lock(String key, String request) throws InterruptedException { for (;;){ String result = this.jedis.set(LOCK_PREFIX + key, request, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, 10 * TIME); if (LOCK_MSG.equals(result)){ break ; //防止一直消耗 CPU  Thread.sleep(DEFAULT_SLEEP_TIME) ; //自定義阻塞時間 public boolean lock(String key, String request,int blockTime) throws InterruptedException { while (blockTime = 0){ String result = this.jedis.set(LOCK_PREFIX + key, request, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, 10 * TIME); if (LOCK_MSG.equals(result)){ return true ; blockTime -= DEFAULT_SLEEP_TIME ; Thread.sleep(DEFAULT_SLEEP_TIME) ; return false ; }

解鎖也很簡單，其實就是把這個 key 刪掉就萬事大吉了，比如使用 del key 命令。

但現(xiàn)實往往沒有那么 easy。

如果進(jìn)程 A 獲取了鎖設(shè)置了超時時間，但是由于執(zhí)行周期較長導(dǎo)致到了超時時間之后鎖就自動釋放了。這時進(jìn)程 B 獲取了該鎖執(zhí)行很快就釋放鎖。這樣就會出現(xiàn)進(jìn)程 B 將進(jìn)程 A 的鎖釋放了。

所以最好的方式是在每次解鎖時都需要判斷鎖是否是自己的。

這時就需要結(jié)合加鎖機(jī)制一起實現(xiàn)了。

加鎖時需要傳遞一個參數(shù)，將該參數(shù)作為這個 key 的 value，這樣每次解鎖時判斷 value 是否相等即可。

所以解鎖代碼就不能是簡單的 del了。

 public boolean unlock(String key,String request){ //lua script String script = if redis.call( get , KEYS[1]) == ARGV[1] then return redis.call( del , KEYS[1]) else return 0 end  Object result = null ; if (jedis instanceof Jedis){ result = ((Jedis)this.jedis).eval(script, Collections.singletonList(LOCK_PREFIX + key), Collections.singletonList(request)); }else if (jedis instanceof JedisCluster){ result = ((JedisCluster)this.jedis).eval(script, Collections.singletonList(LOCK_PREFIX + key), Collections.singletonList(request)); }else { //throw new RuntimeException( instance is error ) ; return false ; if (UNLOCK_MSG.equals(result)){ return true ; }else { return false ; }

這里使用了一個 lua 腳本來判斷 value 是否相等，相等才執(zhí)行 del 命令。

使用 lua 也可以保證這里兩個操作的原子性。

因此上文提到的四個基本特性也能滿足了：

使用 Redis 可以保證性能。

阻塞鎖與非阻塞鎖見上文。

利用超時機(jī)制解決了死鎖。

Redis 支持集群部署提高了可用性。

我自己有擼了一個完整的實現(xiàn)，并且已經(jīng)用于了生產(chǎn)，有興趣的朋友可以開箱使用:

maven 依賴：

 dependency  groupId top.crossoverjie.opensource /groupId  artifactId distributed-redis-lock /artifactId  version 1.0.0 /version  /dependency 

配置 bean :

@Configurationpublic class RedisLockConfig { @Bean public RedisLock build(){ RedisLock redisLock = new RedisLock() ; HostAndPort hostAndPort = new HostAndPort( 127.0.0.1 ,7000) ; JedisCluster jedisCluster = new JedisCluster(hostAndPort) ; // Jedis 或 JedisCluster 都可以 redisLock.setJedisCluster(jedisCluster) ; return redisLock ;}

使用：

 @Autowired private RedisLock redisLock ; public void use() { String key = key  String request = UUID.randomUUID().toString(); try { boolean locktest = redisLock.tryLock(key, request); if (!locktest) { System.out.println( locked error  return;
 }

使用很簡單。這里主要是想利用 Spring 來幫我們管理 RedisLock 這個單例的 bean，所以在釋放鎖的時候需要手動(因為整個上下文只有一個 RedisLock 實例)的傳入 key 以及 request(api 看起來不是特別優(yōu)雅)。

也可以在每次使用鎖的時候 new 一個 RedisLock 傳入 key 以及 request，這樣倒是在解鎖時很方便。但是需要自行管理 RedisLock 的實例。各有優(yōu)劣吧。

在做這個項目的時候讓我不得不想提一下單測。

因為這個應(yīng)用是強(qiáng)依賴于第三方組件的(Redis)，但是在單測中我們需要排除掉這種依賴。比如其他伙伴 fork 了該項目想在本地跑一遍單測，結(jié)果運(yùn)行不起來：

有可能是 Redis 的 ip、端口和單測里的不一致。

Redis 自身可能也有問題。

也有可能是該同學(xué)的環(huán)境中并沒有 Redis。

所以最好是要把這些外部不穩(wěn)定的因素排除掉，單測只測我們寫好的代碼。

于是就可以引入單測利器 Mock 了。

它的想法很簡答，就是要把你所依賴的外部資源統(tǒng)統(tǒng)屏蔽掉。如：數(shù)據(jù)庫、外部接口、外部文件等等。

使用方式也挺簡單，可以參考該項目的單測：

 @Test public void tryLock() throws Exception { String key = test  String request = UUID.randomUUID().toString(); Mockito.when(jedisCluster.set(Mockito.anyString(), Mockito.anyString(), Mockito.anyString(), Mockito.anyString(), Mockito.anyLong())).thenReturn( OK  boolean locktest = redisLock.tryLock(key, request); System.out.println( locktest= + locktest); Assert.assertTrue(locktest); //check Mockito.verify(jedisCluster).set(Mockito.anyString(), Mockito.anyString(), Mockito.anyString(), Mockito.anyString(), Mockito.anyLong()); }

這里只是簡單演示下，可以的話下次仔細(xì)分析分析。

它的原理其實也挺簡單，debug 的話可以很直接的看出來：

這里我們所依賴的 JedisCluster 其實是一個 cglib 代理對象。所以也不難想到它是如何工作的。

比如這里我們需要用到 JedisCluster 的 set 函數(shù)并需要它的返回值。

Mock 就將該對象代理了，并在實際執(zhí)行 set 方法后給你返回了一個你自定義的值。

這樣我們就可以隨心所欲的測試了，完全把外部依賴所屏蔽了。

至此一個基于 Redis 的分布式鎖完成，但是依然有些問題。

如在 key 超時之后業(yè)務(wù)并沒有執(zhí)行完畢但卻自動釋放鎖了，這樣就會導(dǎo)致并發(fā)問題。

就算 Redis 是集群部署的，如果每個節(jié)點都只是 master 沒有 slave，那么 master 宕機(jī)時該節(jié)點上的所有 key 在那一時刻都相當(dāng)于是釋放鎖了，這樣也會出現(xiàn)并發(fā)問題。就算是有 slave 節(jié)點，但如果在數(shù)據(jù)同步到 salve 之前 master 宕機(jī)也是會出現(xiàn)上面的問題。

感興趣的朋友還可以參考 Redisson 的實現(xiàn)。

以上就是如何實現(xiàn)基于 Redis 的分布式鎖的詳細(xì)內(nèi)容，PHP教程

鄭重聲明：本文版權(quán)歸原作者所有，轉(zhuǎn)載文章僅為傳播更多信息之目的，如作者信息標(biāo)記有誤，請第一時間聯(lián)系我們修改或刪除，多謝。