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

首先談到分布式鎖自然也就聯想到分布式應用。

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

但是應用分布式了之后系統由以前的單進程多線程的程序變為了多進程多線程，這時使用以上的解決方案明顯就不夠了。

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

基于 DB 的唯一索引。

基于 ZK 的臨時有序節點。

基于 Redis 的 NX EX 參數。

這里主要基于 Redis 進行討論。

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

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

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

不能出現死鎖。

可用性(不能出現節點 down 掉后加鎖失敗)。

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

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

實現代碼如下：

 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)分開執行，如果在 NX 之后程序出現問題就有可能產生死鎖。

同時也可以實現一個阻塞鎖：

 //一直阻塞 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 命令。

但現實往往沒有那么 easy。

如果進程 A 獲取了鎖設置了超時時間，但是由于執行周期較長導致到了超時時間之后鎖就自動釋放了。這時進程 B 獲取了該鎖執行很快就釋放鎖。這樣就會出現進程 B 將進程 A 的鎖釋放了。

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

這時就需要結合加鎖機制一起實現了。

加鎖時需要傳遞一個參數，將該參數作為這個 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 是否相等，相等才執行 del 命令。

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

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

使用 Redis 可以保證性能。

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

利用超時機制解決了死鎖。

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

我自己有擼了一個完整的實現，并且已經用于了生產，有興趣的朋友可以開箱使用:

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 看起來不是特別優雅)。

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

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

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

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

Redis 自身可能也有問題。

也有可能是該同學的環境中并沒有 Redis。

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

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

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

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

 @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()); }

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

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

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

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

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

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

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

如在 key 超時之后業務并沒有執行完畢但卻自動釋放鎖了，這樣就會導致并發問題。

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

感興趣的朋友還可以參考 Redisson 的實現。

以上就是如何實現基于 Redis 的分布式鎖的詳細內容，PHP教程

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