情景如下：

我們有一批任務需要由多個分布式線程處理，每個任務都有一個taskId，為了保證每個任務只被執行一次，在工作線程執行任務之前，先獲取該任務的鎖，鎖的key可以為taskId

方式1：set(key,value)方式

原理：在獲取鎖之前先查詢一下以該鎖為key對應的value存不存在，如果存在，則說明該鎖被其他客戶端獲取了，否則的話就嘗試獲取鎖，獲取鎖的方法很簡單，只要以該鎖為key，設置一個隨機的值就行了

代碼：

func getLock(taskID int) bool {// 若存在該key，說明其他線程已獲取到鎖，返回失敗// 若不存在該key，則先設置key，標記自己獲取到鎖，返回成功if(existKey(taskID)) {return false} else {setKey(taskID)return true} }

缺陷：該函數并不是原子性的，當一個線程執行existKey()時，檢測到某個鎖不存在，并在執行setKey()之前，其他線程可能也執行了existKey()，同樣檢測到該鎖不存在，也會緊接著執行setKey方法，這樣一來，同一把鎖就有可能被不同的線程獲取到了

方式2：setnx(key,value,timeout)方式?，是「SET if Not eXists」的縮寫，也就是只有不存在的時候才設置，可以利用它來實現鎖的效果

原理：如果 setnx()?返回1，說明該線程獲得鎖，SETNX將鍵 key 的值設置為value

? ? ? ? ? ? 如果 setnx() 返回0，說明其他線程已經獲得了鎖，可以在一個循環中不斷地嘗試 setnx()操作，以獲得鎖

代碼：

func getLock(taskID int) bool {// setnx返回1，說明自己已獲取到鎖，返回成功// setnx返回0，說明其他線程已經獲取到鎖，返回失敗if(SETNX key value,,time.Second*2) == 1{return true;}return false; }

缺陷：客戶端A獲取鎖的時候設置了key的過期時間為2秒，然后客戶端A在獲取到鎖之后，業務邏輯方法doSomething執行了3秒（大于2秒），當執行完業務邏輯方法的時候，客戶端A獲取的鎖已經被Redis過期機制自動釋放了，因此客戶端A在獲取鎖經過2秒之后，該鎖可能已經被其他客戶端獲取到了。當客戶端A執行完doSomething方法之后接下來就是執行releaseLock方法釋放鎖了，由于前面說了，該鎖可能已經被其他客戶端獲取到了，因此這個時候釋放鎖就有可能釋放的是其他客戶端獲取到的鎖

方式3：setnx(key,value,timeout)方式，value為一個隨機值

原理：既然方式二可能會出現釋放了別的客戶端申請到的鎖的問題，那么該如何進行改進呢？有一個很簡單的方法是，我們設置key的時候，將value設置為一個隨機值r，當釋放鎖，也就是刪除key的時候，不是直接刪除，而是先判斷該key對應的value是否等于先前設置的隨機值r，只有當兩者相等的時候才刪除該key，由于每個客戶端產生的隨機值是不一樣的，這樣一來就不會誤釋放別的客戶端申請的鎖了

代碼：

// 獲得鎖 func getLock(taskID int) bool {// setnx返回1，說明自己已獲取到鎖，返回成功// setnx返回0，說明其他線程已經獲取到鎖，返回失敗if(SETNX key value,,time.Second*2) == 1{return true;}return false; }// 釋放鎖 func releaseLock(taskID, value) {// 獲取key對應的值，若和之前設置的隨機值相等，則刪除if getKey(taskID) == value {deleteKey(taskID)} }

缺陷：releaseLock()函數不是原子性的，不是原子性操作意味著當一個客戶端A執行完getKey()并在執行deleteKey()之前，也就是在這2個函數執行之間，其他客戶端是可以執行其他命令的。考慮這樣一種情況，在客戶端A執行完getKey()，并且該key對應的值也等于先前的隨機值的時候，接下來客戶端A將會執行deleteKey()。假設由于網絡或其他原因，客戶端A執行getKey()之后過了1秒鐘才執行deleteKey()，那么在這1秒鐘里，該key有可能也會因為過期而被Redis清除了，這樣一來另一個客戶端，姑且稱之為客戶端B，就有可能在這期間獲取到鎖，然后接下來客戶端A就執行到deleteKey()了，如此一來就又出現誤釋放別的客戶端申請的鎖的問題了

方式4：setnx(key,value,timeout)方式，value為一個隨機值，刪除時執行lua代碼，來保證原子性

原理：既然方式三的問題是因為釋放鎖的方法不是原子操作導致的，那么我們只要保證釋放鎖的代碼是原子性的就能解決該問題了。有另外一種方式，就是Lua腳本。由于Lua腳本的原子性，在Redis執行該腳本的過程中，其他客戶端的命令都需要等待該Lua腳本執行完才能執行，所以不會出現方案三所說的問題。至此，使用Redis實現分布式鎖的方案就相對完善了

代碼：go語言版本

/** ***基于單節點redis 分布式鎖 **/ package redislockimport ("crypto/rand""encoding/base64""errors""github.com/garyburd/redigo/redis" )type RedisLock struct {lockKey stringvalue string }//保證原子性（redis是單線程），避免del刪除了其他client獲得的lock var delScript = redis.NewScript(1, ` if redis.call("get", KEYS[1]) == ARGV[1] thenreturn redis.call("del", KEYS[1]) elsereturn 0 end`)// 獲得鎖 func (this *RedisLock) Lock(rd *redis.Conn, timeout int) error {{ //隨機數b := make([]byte, 16)_, err := rand.Read(b)if err != nil {return err}this.value = base64.StdEncoding.EncodeToString(b)}lockReply, err := (*rd).Do("SET", this.lockKey, this.value, "ex", timeout, "nx")if err != nil {return errors.New("redis fail")}if lockReply == "OK" {return nil} else {return errors.New("lock fail")} }// 釋放鎖 func (this *RedisLock) Unlock(rd *redis.Conn) {delScript.Do(*rd, this.lockKey, this.value) }

結論：

上述分布式鎖的實現方案中，都是針對單節點Redis而言的，然而在實際的生產環境中，我們使用的通常是Redis集群，并且每個主節點還會有從節點。由于Redis的主從復制是異步的，因此上述方案在Redis集群的環境下也是有問題的。比如主節點剛設置了一個key用做鎖，還沒同步到從節點，此時主節點崩潰了，稍后從節點升級為主節點，自然沒有這個key，那么其他客戶端請求時就又請求到了鎖，造成混亂。

關于在Redis集群中如何優雅地實現分布式鎖，后續再寫文章詳述

總結

以上是生活随笔為你收集整理的redis 分布式锁的实现方式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。