怎樣用Redis Nx方式實現(xiàn)分布式鎖？

本地鎖只能控制所在虛擬機(jī)中的線程同步執(zhí)行，現(xiàn)在要實現(xiàn)分布式環(huán)境下所有虛擬機(jī)中的線程去同步執(zhí)行就需要讓多個虛擬機(jī)去共用一個鎖，虛擬機(jī)可以分布式部署，鎖也可以分布式部署，如下圖：

虛擬機(jī)都去搶占同一個鎖，鎖是一個單獨的程序提供加鎖、解鎖服務(wù)，誰搶到鎖誰去查詢數(shù)據(jù)庫。

該鎖已不屬于某個虛擬機(jī)，而是分布式部署，由多個虛擬機(jī)所共享，這種鎖叫分布式鎖。

分布式鎖實現(xiàn)方案

實現(xiàn)分布式鎖的方案有很多，常用的如下：

1、基于數(shù)據(jù)庫實現(xiàn)分布鎖

利用數(shù)據(jù)庫主鍵唯一性的特點，或利用數(shù)據(jù)庫唯一索引的特點，多個線程同時去插入相同的記錄，誰插入成功誰就搶到鎖。

2、基于redis實現(xiàn)鎖

redis提供了分布式鎖的實現(xiàn)方案，比如：SETNX、set nx、redisson等。

拿SETNX舉例說明，SETNX命令的工作過程是去set一個不存在的key，多個線程去設(shè)置同一個key只會有一個線程設(shè)置成功，設(shè)置成功的的線程拿到鎖。

3、使用zookeeper實現(xiàn)

zookeeper是一個分布式協(xié)調(diào)服務(wù)，主要解決分布式程序之間的同步的問題。zookeeper的結(jié)構(gòu)類似的文件目錄，多線程向zookeeper創(chuàng)建一個子目錄(節(jié)點)只會有一個創(chuàng)建成功，利用此特點可以實現(xiàn)分布式鎖，誰創(chuàng)建該結(jié)點成功誰就獲得鎖。

Redis NX實現(xiàn)分布式鎖

redis實現(xiàn)分布式鎖的方案可以在redis.cn網(wǎng)站查閱，地址http://www.redis.cn/commands/set.html

使用命令： SET resource-name anystring NX EX max-lock-time 即可實現(xiàn)。

NX：表示key不存在才設(shè)置成功。

EX：設(shè)置過期時間

這里啟動三個ssh客戶端，連接redis: docker exec -it redis redis-cli

先認(rèn)證: auth redis

同時向三個客戶端發(fā)送測試命令如下：

表示設(shè)置lock001鎖，value為001，過期時間為30秒

Plain Text
SET lock001 001 NX EX 30

命令發(fā)送成功，觀察三個ssh客戶端發(fā)現(xiàn)只有一個設(shè)置成功，其它兩個設(shè)置失敗，設(shè)置成功的請求表示搶到了lock001鎖。

如何在代碼中使用Set nx去實現(xiàn)分布鎖呢?

使用spring-boot-starter-data-redis 提供的api即可實現(xiàn)set nx。添加依賴：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
    <version>2.6.2</version>
</dependency>

添加依賴后，在bean中注入restTemplate。我們先分析一段偽代碼如下：

if(緩存中有){

  返回緩存中的數(shù)據(jù)
}else{

  獲取分布式鎖
  if(獲取鎖成功）{
       try{
         查詢數(shù)據(jù)庫
      }finally{
         釋放鎖
      }
  }
 
}

1、獲取分布式鎖

使用redisTemplate.opsForValue().setIfAbsent(key,vaue)獲取鎖。

這里考慮一個問題，當(dāng)set nx一個key/value成功1后，這個key(就是鎖)需要設(shè)置過期時間嗎?

如果不設(shè)置過期時間當(dāng)獲取到了鎖卻沒有執(zhí)行finally這個鎖將會一直存在，其它線程無法獲取這個鎖。所以執(zhí)行set nx時要指定過期時間，即使用如下的命令。

SET resource-name anystring NX EX max-lock-time

具體調(diào)用的方法是：redisTemplate.opsForValue().setIfAbsent(K var1, V var2, long var3, TimeUnit var5)

2、如何釋放鎖

釋放鎖分為兩種情況：key到期自動釋放，手動刪除。

1)key到期自動釋放的方法

因為鎖設(shè)置了過期時間，key到期會自動釋放，但是會存在一個問題就是 查詢數(shù)據(jù)庫等操作還沒有執(zhí)行完時key到期了，此時其它線程就搶到鎖了，最終重復(fù)查詢數(shù)據(jù)庫執(zhí)行了重復(fù)的業(yè)務(wù)操作。

怎么解決這個問題?

可以將key的到期時間設(shè)置的長一些，足以執(zhí)行完成查詢數(shù)據(jù)庫并設(shè)置緩存等相關(guān)操作。

如果這樣效率會低一些，另外這個時間值也不好把控。

2)手動刪除鎖

如果是采用手動刪除鎖可能和key到期自動刪除有所沖突，造成刪除了別人的鎖。

比如：當(dāng)查詢數(shù)據(jù)庫等業(yè)務(wù)還沒有執(zhí)行完時key過期了，此時其它線程占用了鎖，當(dāng)上一個線程執(zhí)行查詢數(shù)據(jù)庫等業(yè)務(wù)操作完成后手動刪除鎖就把其它線程的鎖給刪除了。

要解決這個問題可以采用刪除鎖之前判斷是不是自己設(shè)置的鎖，偽代碼如下：

if(緩存中有){

  返回緩存中的數(shù)據(jù)
}else{

  獲取分布式鎖: set lock 01 NX
  if(獲取鎖成功）{
       try{
         查詢數(shù)據(jù)庫
      }finally{
         if(redis.call("get","lock")=="01"){
            釋放鎖: redis.call("del","lock")
         }
         
      }
  }
 
}

以上代碼第11行到13行非原子性，也會導(dǎo)致刪除其它線程的鎖。查看文檔上的說明：http://www.redis.cn/commands/set.html

上述優(yōu)化方法會避免下述場景：a客戶端獲得的鎖(鍵key)已經(jīng)由于過期時間到了被redis服務(wù)器刪除，但是這個時候a客戶端還去執(zhí)行DEL命令。而b客戶端已經(jīng)在a設(shè)置過期時間之后重新獲取了這個同樣key的鎖，那么a執(zhí)行DEL就會釋放了b客戶端加好的鎖。

解鎖腳本的一個例子將類似于以下：

if redis.cal1("get",KEYS[1]) == ARGV[1]
then
    return redis. call("del",KEYS[1])
else
    return 0
end

在調(diào)用setnx命令設(shè)置key/value時，每個線程設(shè)置不一樣的value值，這樣當(dāng)線程去刪除鎖時可以先根據(jù)key查詢出來判斷是不是自己當(dāng)時設(shè)置的vlaue，如果是則刪除。

這整個操作是原子的，實現(xiàn)方法就是去執(zhí)行上邊的lua腳本。

Lua 是一個小巧的腳本語言，redis在2.6版本就支持通過執(zhí)行Lua腳本保證多個命令的原子性。

什么是原子性?

這些指令要么全成功要么全失敗。

以上就是使用Redis Nx方式實現(xiàn)分布式鎖，為了避免刪除別的線程設(shè)置的鎖需要使用redis去執(zhí)行Lua腳本的方式去實現(xiàn)，這樣就具有原子性，但是過期時間的值設(shè)置不存在不精確的問題。