Redis：Redisson分布式鎖的使用（生產(chǎn)環(huán)境下）（推薦使用）

關(guān)鍵詞

• 基于NIO的Netty框架，生產(chǎn)環(huán)境使用分布式鎖
• redisson加鎖：lua腳本加鎖（其他客戶端自旋）
• 自動(dòng)延時(shí)機(jī)制：?jiǎn)?dòng)watch dog，后臺(tái)線程，每隔10秒檢查一下客戶端1還持有鎖key，會(huì)不斷的延長(zhǎng)鎖key的生存時(shí)間
• 可重入鎖機(jī)制：第二個(gè)if判斷 ，myLock :{“8743c9c0-0795-4907-87fd-6c719a6b4586:1”:2 }
• 釋放鎖：無(wú)鎖直接返回；有鎖不是我加的，返回；有鎖是我加的，執(zhí)行hincrby -1,當(dāng)重入鎖減完才執(zhí)行del操作
• Redis使用同一個(gè)Lua解釋器來(lái)執(zhí)行所有命令，Redis保證以一種原子性的方式來(lái)執(zhí)行腳本：當(dāng)lua腳本在執(zhí)行的時(shí)候，不會(huì)有其他腳本和命令同時(shí)執(zhí)行，這種語(yǔ)義類(lèi)似于 MULTI/EXEC。從別的客戶端的視角來(lái)看，一個(gè)
• lua腳本要么不可見(jiàn)，要么已經(jīng)執(zhí)行完

一、 Redisson使用

Redisson是架設(shè)在Redis基礎(chǔ)上的一個(gè)Java駐內(nèi)存數(shù)據(jù)網(wǎng)格（In-Memory Data Grid）。

Redisson在基于NIO的Netty框架上，生產(chǎn)環(huán)境使用分布式鎖。

加入jar包的依賴

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>2.7.0</version>
</dependency>

配置Redisson

public class RedissonManager {
  private static Config config = new Config();
  //聲明redisso對(duì)象
  private static Redisson redisson = null;
  
   //實(shí)例化redisson
	static{
	  config.useClusterServers()
	  // 集群狀態(tài)掃描間隔時(shí)間，單位是毫秒
	 .setScanInterval(2000)
	  //cluster方式至少6個(gè)節(jié)點(diǎn)(3主3從，3主做sharding，3從用來(lái)保證主宕機(jī)后可以高可用)
	 .addNodeAddress("redis://127.0.0.1:6379" )
	 .addNodeAddress("redis://127.0.0.1:6380")
	 .addNodeAddress("redis://127.0.0.1:6381")
	 .addNodeAddress("redis://127.0.0.1:6382")
	 .addNodeAddress("redis://127.0.0.1:6383")
	 .addNodeAddress("redis://127.0.0.1:6384");
	 
	  //得到redisson對(duì)象
	  redisson = (Redisson) Redisson.create(config);
	}
	
	  //獲取redisson對(duì)象的方法
	  public static Redisson getRedisson(){
	    return redisson;
	 }
}

鎖的獲取和釋放

public class DistributedRedisLock {
  //從配置類(lèi)中獲取redisson對(duì)象
  private static Redisson redisson = RedissonManager.getRedisson();
  private static final String LOCK_TITLE = "redisLock_";
  
  //加鎖
  public static boolean acquire(String lockName){
    //聲明key對(duì)象
    String key = LOCK_TITLE + lockName;
    //獲取鎖對(duì)象
    RLock mylock = redisson.getLock(key);
    //加鎖，并且設(shè)置鎖過(guò)期時(shí)間3秒，防止死鎖的產(chǎn)生  uuid+threadId
    mylock.lock(2,3,TimeUtil.SECOND);
    //加鎖成功
    return  true;
  }
  
  //鎖的釋放
  public static void release(String lockName){
    //必須是和加鎖時(shí)的同一個(gè)key
    String key = LOCK_TITLE + lockName;
    //獲取所對(duì)象
    RLock mylock = redisson.getLock(key);
    //釋放鎖（解鎖）
    mylock.unlock();
  
 

業(yè)務(wù)邏輯中使用分布式鎖

public String discount() throws IOException{
    String key = "lock001";
    //加鎖
    DistributedRedisLock.acquire(key);
    //執(zhí)行具體業(yè)務(wù)邏輯
    dosoming
    //釋放鎖
    DistributedRedisLock.release(key);
    //返回結(jié)果
    return soming;
 }

二、Redisson分布式鎖的實(shí)現(xiàn)原理

2.1 加鎖機(jī)制

如果該客戶端面對(duì)的是一個(gè)redis cluster集群，他首先會(huì)根據(jù)hash節(jié)點(diǎn)選擇一臺(tái)機(jī)器。

發(fā)送lua腳本到redis服務(wù)器上，腳本如下:

//exists',KEYS[1])==0 不存在，沒(méi)鎖
"if (redis.call('exists',KEYS[1])==0) then "+       --看有沒(méi)有鎖
  // 命令：hset,1：第一回
	"redis.call('hset',KEYS[1],ARGV[2],1) ; "+       --無(wú)鎖 加鎖 
	// 配置鎖的生命周期 
	"redis.call('pexpire',KEYS[1],ARGV[1]) ; "+      
	"return nil; end ;" +

//可重入操作，判斷是不是我加的鎖
"if (redis.call('hexists',KEYS[1],ARGV[2]) ==1 ) then "+  --我加的鎖
   //hincrby 在原來(lái)的鎖上加1
	"redis.call('hincrby',KEYS[1],ARGV[2],1) ; "+  --重入鎖
	"redis.call('pexpire',KEYS[1],ARGV[1]) ; "+  
	"return nil; end ;" +

//否則，鎖存在，返回鎖的有效期，決定下次執(zhí)行腳本時(shí)間
"return redis.call('pttl',KEYS[1]) ;"  --不能加鎖，返回鎖的時(shí)間

lua的作用：保證這段復(fù)雜業(yè)務(wù)邏輯執(zhí)行的原子性。

lua的解釋?zhuān)?/p>

• KEYS[1]) ： 加鎖的key
• ARGV[1] ： key的生存時(shí)間，默認(rèn)為30秒
• ARGV[2] ： 加鎖的客戶端ID (UUID.randomUUID()） + “:” + threadId)

第一段if判斷語(yǔ)句，就是用“exists myLock”命令判斷一下，如果你要加鎖的那個(gè)鎖key不存在的話，你就進(jìn)行加鎖。

如何加鎖呢？很簡(jiǎn)單，用下面的命令：

hset myLock

8743c9c0-0795-4907-87fd-6c719a6b4586:1 1

通過(guò)這個(gè)命令設(shè)置一個(gè)hash數(shù)據(jù)結(jié)構(gòu)，這行命令執(zhí)行后，會(huì)出現(xiàn)一個(gè)類(lèi)似下面的數(shù)據(jù)結(jié)構(gòu)：

myLock :{“8743c9c0-0795-4907-87fd-6c719a6b4586:1”:1 }

上述就代表“8743c9c0-0795-4907-87fd-6c719a6b4586:1”這個(gè)客戶端對(duì)“myLock”這個(gè)鎖key完成了加鎖。

接著會(huì)執(zhí)行“pexpire myLock 30000”命令，設(shè)置myLock這個(gè)鎖key的生存時(shí)間是30秒。

鎖互斥機(jī)制

那么在這個(gè)時(shí)候，如果客戶端2來(lái)嘗試加鎖，執(zhí)行了同樣的一段lua腳本，會(huì)咋樣呢？

很簡(jiǎn)單，第一個(gè)if判斷會(huì)執(zhí)行“exists myLock”，發(fā)現(xiàn)myLock這個(gè)鎖key已經(jīng)存在了。

接著第二個(gè)if判斷，判斷一下，myLock鎖key的hash數(shù)據(jù)結(jié)構(gòu)中，是否包含客戶端2的ID，但是明顯不是的，因?yàn)槟抢锇氖强蛻舳?的ID。

所以，客戶端2會(huì)獲取到pttl myLock返回的一個(gè)數(shù)字，這個(gè)數(shù)字代表了myLock這個(gè)鎖key的剩余生存時(shí)間。比如還剩15000毫秒的生存時(shí)間。

此時(shí)客戶端2會(huì)進(jìn)入一個(gè)while循環(huán)，不停的嘗試加鎖。

自動(dòng)延時(shí)機(jī)制

只要客戶端1一旦加鎖成功，就會(huì)啟動(dòng)一個(gè)watch dog看門(mén)狗，他是一個(gè)后臺(tái)線程，會(huì)每隔10秒檢查一下，如果客戶端1還持有鎖key，那么就會(huì)不斷的延長(zhǎng)鎖key的生存時(shí)間。

可重入鎖機(jī)制

第一個(gè)if判斷 肯定不成立，“exists myLock”會(huì)顯示鎖key已經(jīng)存在了。

第二個(gè)if判斷 會(huì)成立，因?yàn)閙yLock的hash數(shù)據(jù)結(jié)構(gòu)中包含的那個(gè)ID，就是客戶端1的那個(gè)ID，也就是“8743c9c0-0795-4907-87fd-6c719a6b4586:1”

此時(shí)就會(huì)執(zhí)行可重入加鎖的邏輯，他會(huì)用：incrby myLock 8743c9c0-0795-4907-87fd-6c71a6b4586:1 1

通過(guò)這個(gè)命令，對(duì)客戶端1的加鎖次數(shù)，累加1。數(shù)據(jù)結(jié)構(gòu)會(huì)變成：myLock :{“8743c9c0-0795-4907-87fd-6c719a6b4586:1”:2 }

2.2 釋放鎖機(jī)制

執(zhí)行l(wèi)ua腳本如下：

# 如果key已經(jīng)不存在，說(shuō)明已經(jīng)被解鎖，直接發(fā)布（publish）redis消息（無(wú)鎖，直接返回）
"if (redis.call('exists', KEYS[1]) == 0) then " +
            "redis.call('publish', KEYS[2], ARGV[1]); " +
            "return 1; " +
          "end;" +
# key和field不匹配，說(shuō)明當(dāng)前客戶端線程沒(méi)有持有鎖，不能主動(dòng)解鎖。 不是我加的鎖 不能解鎖 （有鎖不是我加的，返回）
          "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
            "return nil;" +
          "end; " +
# 將value減1 （有鎖是我加的，進(jìn)行hincrby -1 ）
          "local counter = redis.call('hincrby', KEYS[1], ARGV[3],-1); " +
# 如果counter>0說(shuō)明鎖在重入，不能刪除key
          "if (counter > 0) then " +
            "redis.call('pexpire', KEYS[1], ARGV[2]); " +
            "return 0; " +
# 刪除key并且publish 解鎖消息
					# 可重入鎖減完了，進(jìn)行del操作
          "else " + 
            "redis.call('del', KEYS[1]); " + #刪除鎖
            "redis.call('publish', KEYS[2], ARGV[1]); " +
            "return 1; "+
             "end; " +
             "return nil;",

• – KEYS[1] ：需要加鎖的key，這里需要是字符串類(lèi)型。
• – KEYS[2] ：redis消息的ChannelName，一個(gè)分布式鎖對(duì)應(yīng)唯一的一個(gè)channelName: “redisson_lockchannel{” + getName() + “}”
• – ARGV[1] ：reids消息體，這里只需要一個(gè)字節(jié)的標(biāo)記就可以，主要標(biāo)記redis的key已經(jīng)解鎖，再結(jié)合 redis的Subscribe，能喚醒其他訂閱解鎖消息的客戶端線程申請(qǐng)鎖。
• – ARGV[2] ：鎖的超時(shí)時(shí)間，防止死鎖
• – ARGV[3] ：鎖的唯一標(biāo)識(shí)，也就是剛才介紹的 id（UUID.randomUUID()） + “:” + threadId

如果執(zhí)行l(wèi)ock.unlock()，就可以釋放分布式鎖，此時(shí)的業(yè)務(wù)邏輯也是非常簡(jiǎn)單的。

其實(shí)說(shuō)白了，就是每次都對(duì)myLock數(shù)據(jù)結(jié)構(gòu)中的那個(gè)加鎖次數(shù)減1。

如果發(fā)現(xiàn)加鎖次數(shù)是0了，說(shuō)明這個(gè)客戶端已經(jīng)不再持有鎖了，此時(shí)就會(huì)用：

• “del myLock”命令，從redis里刪除這個(gè)key。
• 然后呢，另外的客戶端2就可以嘗試完成加鎖了。

分布式鎖特性

• 互斥性
• 任意時(shí)刻，只能有一個(gè)客戶端獲取鎖，不能同時(shí)有兩個(gè)客戶端獲取到鎖。
• 同一性
• 鎖只能被持有該鎖的客戶端刪除，不能由其它客戶端刪除。
• 可重入性
• 持有某個(gè)鎖的客戶端可繼續(xù)對(duì)該鎖加鎖，實(shí)現(xiàn)鎖的續(xù)租
• 容錯(cuò)性
• 鎖失效后（超過(guò)生命周期）自動(dòng)釋放鎖（key失效），其他客戶端可以繼續(xù)獲得該鎖，防止死鎖

分布式鎖的實(shí)際應(yīng)用

• 數(shù)據(jù)并發(fā)競(jìng)爭(zhēng)
• 利用分布式鎖可以將處理串行化，前面已經(jīng)講過(guò)了。
• 防止庫(kù)存超賣(mài)

訂單1下單前會(huì)先查看庫(kù)存，庫(kù)存為10，所以下單5本可以成功；

訂單2下單前會(huì)先查看庫(kù)存，庫(kù)存為10，所以下單8本可以成功；

訂單1和訂單2 同時(shí)操作，共下單13本，但庫(kù)存只有10本，顯然庫(kù)存不夠了，這種情況稱為庫(kù)存超賣(mài)。

可以采用分布式鎖解決這個(gè)問(wèn)題。

訂單1和訂單2都從Redis中獲得分布式鎖(setnx)，誰(shuí)能獲得鎖誰(shuí)進(jìn)行下單操作，這樣就把訂單系統(tǒng)下單的順序串行化了，就不會(huì)出現(xiàn)超賣(mài)的情況了。

偽碼如下：

//加鎖并設(shè)置有效期
if(redis.lock("RDL",200)){
  //判斷庫(kù)存
  if (orderNum<getCount()){
  //加鎖成功 ,可以下單
  order(5);
  //釋放鎖
  redis,unlock("RDL");
 }  
}

注意此種方法會(huì)降低處理效率，這樣不適合秒殺的場(chǎng)景，秒殺可以使用CAS和Redis隊(duì)列的方式。

總結(jié)

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

最新評(píng)論