一、前言分布式鎖在實際工作中的應(yīng)用還是比較多的其實現(xiàn)方式也有很多種常見的有基于數(shù)據(jù)庫鎖、基于zookeeper、基于redis的今天我們來講下基于redis實現(xiàn)的分布式鎖。redisson是一個redis客戶端框架提供了分布式鎖的功能特性這里我們通過解析redisson的源碼來分析它是如何基于redis來實現(xiàn)分布式鎖的二、源碼解析2.1 樣例代碼下面是一個分布式鎖的簡單樣例代碼// 初始化配置創(chuàng)建Redisson客戶端 Config confignew Config();config.setCodec(new JsonJacksonCodec()).useSingleServer().setAddress(redis://192.168.10.131:6379);RedissonClient clientRedisson.create(config);// 獲取分布式鎖 RLock lockclient.getLock(myLock);lock.lock();System.out.println(Thread.currentThread().getId(): 獲取到分布式鎖);try{Thread.sleep(60*1000);}catch(Exception e){e.printStackTrace();}finally{// 解鎖 lock.unlock();}上面的樣例代碼比較簡單通過redisson客戶端獲取一個分布式鎖該分布式鎖的key為myLock睡眠60秒之后釋放鎖。這里比較重要的是lock()方法該方法是獲取鎖的具體步驟所以接下來詳細(xì)解析一下該方法2.2 整體流程獲取鎖的流程圖如下具體流程為第一次嘗試獲取鎖如果獲取到鎖直接返回。如果未獲取到鎖返回鎖的剩余過期時間ttl當(dāng)未獲取到鎖時訂閱頻道redisson_lock__channel:{myLock}訂閱該頻道的作用是當(dāng)該分布式鎖被其他擁有者所釋放時會往該訂閱頻道發(fā)送一個解鎖消息UNLOCK_MESSAGE這時當(dāng)前等待該分布式鎖的線程會中斷等待并再次嘗試獲取鎖開啟死循環(huán)嘗試獲取鎖如果未獲取到鎖拿到鎖的剩余過期時間并等待該鎖的剩余過期時間中間過程中如果訂閱頻道有解鎖消息UNLOCK_MESSAGE會提前中斷等待繼續(xù)循環(huán)直到獲取鎖退出循環(huán)獲取到鎖之后取消訂閱頻道源碼如下private void lock(long leaseTime, TimeUnit unit, boolean interruptibly)throws InterruptedException{long threadIdThread.currentThread().getId();//1、第一次嘗試獲取鎖ttl為null表示獲取到鎖直接return Long ttltryAcquire(-1, leaseTime, unit, threadId);if(ttlnull){return;}//2、訂閱頻道redisson_lock__channel:{myLock}CompletableFutureRedissonLockEntryfuturesubscribe(threadId);pubSub.timeout(future);RedissonLockEntry entry;if(interruptibly){entrycommandExecutor.getInterrupted(future);}else{entrycommandExecutor.get(future);}try{//3、開啟循環(huán)while(true){// 再次嘗試獲取鎖ttl為null表示獲取到鎖退出循環(huán) ttltryAcquire(-1, leaseTime, unit, threadId);if(ttlnull){break;}// 如果ttl大于等于0if(ttl0){try{// 等待ttl時間 或者 接收到解鎖消息 entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);}catch(InterruptedException e){...}}else{// 如果ttl小于0說明該鎖未設(shè)置過期時間等待接收解鎖消息if(interruptibly){entry.getLatch().acquire();}else{entry.getLatch().acquireUninterruptibly();}}}}finally{// 退出訂閱頻道redisson_lock__channel:{myLock}unsubscribe(entry, threadId);}}2.3 鎖的獲取那么如何表示當(dāng)前線程獲取到鎖redisson中的分布式鎖實質(zhì)上是個hash結(jié)構(gòu)的數(shù)據(jù)假設(shè)鎖的名稱為myLock那么當(dāng)某個線程獲取到鎖之后會在這個hash結(jié)構(gòu)里設(shè)置一個hashkey其為【連接管理器id】 : 【線程id】如下圖redisson通過執(zhí)行l(wèi)ua腳本來獲取鎖lua腳本如下// 如果鎖不存在則成功獲取到鎖設(shè)置鎖的過期時間并返回nilif(redis.call(exists, KEYS[1])0)thenredis.call(hincrby, KEYS[1], ARGV[2],1);redis.call(pexpire, KEYS[1], ARGV[1]);returnnil;end;// 如果鎖已存在判斷是否是當(dāng)前線程已經(jīng)獲取到如果是對應(yīng)的值加1if(redis.call(hexists, KEYS[1], ARGV[2])1)thenredis.call(hincrby, KEYS[1], ARGV[2],1);redis.call(pexpire, KEYS[1], ARGV[1]);returnnil;end;// 否則表示未獲取到鎖返回鎖的過期時間returnredis.call(pttl, KEYS[1]);該lua腳本的主要作用是如果鎖不存在則成功獲取到鎖設(shè)置鎖的過期時間lockWatchdogTimeout默認(rèn)是30秒并返回nil如果鎖已存在判斷是否是當(dāng)前線程已經(jīng)獲取到如果是對應(yīng)的值加1否則表示未獲取到鎖返回鎖的過期時間這里的第一步為什么要設(shè)置鎖的過期時間其實是為了當(dāng)鎖的擁有者掛了之后避免鎖一直存在導(dǎo)致其他應(yīng)用永遠(yuǎn)無法獲取到鎖。2.4 鎖續(xù)期那么既然鎖設(shè)置了過期時間那很自然地想到如果在鎖過期的這段時間內(nèi)擁有鎖的線程還未執(zhí)行完業(yè)務(wù)邏輯這時鎖自動過期導(dǎo)致其他應(yīng)用也獲取到了鎖從而產(chǎn)生邏輯錯誤。所以引入了鎖續(xù)期。當(dāng)獲取到鎖時redisson會啟動一個看門狗該看門狗每隔 lockWatchdogTimeout / 3秒續(xù)期一次鎖假設(shè)lockWatchdogTimeout默認(rèn)為30秒則每隔10秒續(xù)期鎖源碼如下private voidrenewExpiration(){... //1、創(chuàng)建一個10秒后執(zhí)行的延遲任務(wù) Timeout taskcommandExecutor.getConnectionManager().newTimeout(newTimerTask(){Override public void run(Timeouttimeout)throws Exception{... //2、執(zhí)行續(xù)期鎖的lua腳本 CompletionStageBooleanfuturerenewExpirationAsync(threadId);future.whenComplete((res, e)-{... //3、res為true代表鎖續(xù)期成功重新調(diào)用該方法繼續(xù)創(chuàng)建延遲任務(wù) // false表示鎖續(xù)期失敗if(res){renewExpiration();}else{cancelExpirationRenewal(null);}});}}, internalLockLeaseTime /3, TimeUnit.MILLISECONDS);}續(xù)期鎖的lua腳本如下// 如果鎖存在這個hashkey重新設(shè)置鎖的過期時間if(redis.call(hexists, KEYS[1], ARGV[2])1)thenredis.call(pexpire, KEYS[1], ARGV[1]);return1;end;return0;到這里redisson實現(xiàn)分布式鎖的源碼解析就結(jié)束了。三、總結(jié)redisson的源碼中大量使用了異步編程這導(dǎo)致閱讀源碼的難度系數(shù)較高這里我也只是大概整理了一下有問題的同學(xué)可以互相討論一下或自行查閱源碼。 redisson分布式鎖的源碼解析-元一軟件
一、前言分布式鎖在實際工作中的應(yīng)用還是比較多的其實現(xiàn)方式也有很多種常見的有基于數(shù)據(jù)庫鎖、基于zookeeper、基于redis的今天我們來講下基于redis實現(xiàn)的分布式鎖。redisson是一個redis客戶端框架提供了分布式鎖的功能特性這里我們通過解析redisson的源碼來分析它是如何基于redis來實現(xiàn)分布式鎖的二、源碼解析2.1 樣例代碼下面是一個分布式鎖的簡單樣例代碼// 初始化配置創(chuàng)建Redisson客戶端 Config confignew Config();config.setCodec(new JsonJacksonCodec()).useSingleServer().setAddress(redis://192.168.10.131:6379);RedissonClient clientRedisson.create(config);// 獲取分布式鎖 RLock lockclient.getLock(myLock);lock.lock();System.out.println(Thread.currentThread().getId(): 獲取到分布式鎖);try{Thread.sleep(60*1000);}catch(Exception e){e.printStackTrace();}finally{// 解鎖 lock.unlock();}上面的樣例代碼比較簡單通過redisson客戶端獲取一個分布式鎖該分布式鎖的key為myLock睡眠60秒之后釋放鎖。這里比較重要的是lock()方法該方法是獲取鎖的具體步驟所以接下來詳細(xì)解析一下該方法2.2 整體流程獲取鎖的流程圖如下具體流程為第一次嘗試獲取鎖如果獲取到鎖直接返回。如果未獲取到鎖返回鎖的剩余過期時間ttl當(dāng)未獲取到鎖時訂閱頻道redisson_lock__channel:{myLock}訂閱該頻道的作用是當(dāng)該分布式鎖被其他擁有者所釋放時會往該訂閱頻道發(fā)送一個解鎖消息UNLOCK_MESSAGE這時當(dāng)前等待該分布式鎖的線程會中斷等待并再次嘗試獲取鎖開啟死循環(huán)嘗試獲取鎖如果未獲取到鎖拿到鎖的剩余過期時間并等待該鎖的剩余過期時間中間過程中如果訂閱頻道有解鎖消息UNLOCK_MESSAGE會提前中斷等待繼續(xù)循環(huán)直到獲取鎖退出循環(huán)獲取到鎖之后取消訂閱頻道源碼如下private void lock(long leaseTime, TimeUnit unit, boolean interruptibly)throws InterruptedException{long threadIdThread.currentThread().getId();//1、第一次嘗試獲取鎖ttl為null表示獲取到鎖直接return Long ttltryAcquire(-1, leaseTime, unit, threadId);if(ttlnull){return;}//2、訂閱頻道redisson_lock__channel:{myLock}CompletableFutureRedissonLockEntryfuturesubscribe(threadId);pubSub.timeout(future);RedissonLockEntry entry;if(interruptibly){entrycommandExecutor.getInterrupted(future);}else{entrycommandExecutor.get(future);}try{//3、開啟循環(huán)while(true){// 再次嘗試獲取鎖ttl為null表示獲取到鎖退出循環(huán) ttltryAcquire(-1, leaseTime, unit, threadId);if(ttlnull){break;}// 如果ttl大于等于0if(ttl0){try{// 等待ttl時間 或者 接收到解鎖消息 entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);}catch(InterruptedException e){...}}else{// 如果ttl小于0說明該鎖未設(shè)置過期時間等待接收解鎖消息if(interruptibly){entry.getLatch().acquire();}else{entry.getLatch().acquireUninterruptibly();}}}}finally{// 退出訂閱頻道redisson_lock__channel:{myLock}unsubscribe(entry, threadId);}}2.3 鎖的獲取那么如何表示當(dāng)前線程獲取到鎖redisson中的分布式鎖實質(zhì)上是個hash結(jié)構(gòu)的數(shù)據(jù)假設(shè)鎖的名稱為myLock那么當(dāng)某個線程獲取到鎖之后會在這個hash結(jié)構(gòu)里設(shè)置一個hashkey其為【連接管理器id】 : 【線程id】如下圖redisson通過執(zhí)行l(wèi)ua腳本來獲取鎖lua腳本如下// 如果鎖不存在則成功獲取到鎖設(shè)置鎖的過期時間并返回nilif(redis.call(exists, KEYS[1])0)thenredis.call(hincrby, KEYS[1], ARGV[2],1);redis.call(pexpire, KEYS[1], ARGV[1]);returnnil;end;// 如果鎖已存在判斷是否是當(dāng)前線程已經(jīng)獲取到如果是對應(yīng)的值加1if(redis.call(hexists, KEYS[1], ARGV[2])1)thenredis.call(hincrby, KEYS[1], ARGV[2],1);redis.call(pexpire, KEYS[1], ARGV[1]);returnnil;end;// 否則表示未獲取到鎖返回鎖的過期時間returnredis.call(pttl, KEYS[1]);該lua腳本的主要作用是如果鎖不存在則成功獲取到鎖設(shè)置鎖的過期時間lockWatchdogTimeout默認(rèn)是30秒并返回nil如果鎖已存在判斷是否是當(dāng)前線程已經(jīng)獲取到如果是對應(yīng)的值加1否則表示未獲取到鎖返回鎖的過期時間這里的第一步為什么要設(shè)置鎖的過期時間其實是為了當(dāng)鎖的擁有者掛了之后避免鎖一直存在導(dǎo)致其他應(yīng)用永遠(yuǎn)無法獲取到鎖。2.4 鎖續(xù)期那么既然鎖設(shè)置了過期時間那很自然地想到如果在鎖過期的這段時間內(nèi)擁有鎖的線程還未執(zhí)行完業(yè)務(wù)邏輯這時鎖自動過期導(dǎo)致其他應(yīng)用也獲取到了鎖從而產(chǎn)生邏輯錯誤。所以引入了鎖續(xù)期。當(dāng)獲取到鎖時redisson會啟動一個看門狗該看門狗每隔 lockWatchdogTimeout / 3秒續(xù)期一次鎖假設(shè)lockWatchdogTimeout默認(rèn)為30秒則每隔10秒續(xù)期鎖源碼如下private voidrenewExpiration(){... //1、創(chuàng)建一個10秒后執(zhí)行的延遲任務(wù) Timeout taskcommandExecutor.getConnectionManager().newTimeout(newTimerTask(){Override public void run(Timeouttimeout)throws Exception{... //2、執(zhí)行續(xù)期鎖的lua腳本 CompletionStageBooleanfuturerenewExpirationAsync(threadId);future.whenComplete((res, e)-{... //3、res為true代表鎖續(xù)期成功重新調(diào)用該方法繼續(xù)創(chuàng)建延遲任務(wù) // false表示鎖續(xù)期失敗if(res){renewExpiration();}else{cancelExpirationRenewal(null);}});}}, internalLockLeaseTime /3, TimeUnit.MILLISECONDS);}續(xù)期鎖的lua腳本如下// 如果鎖存在這個hashkey重新設(shè)置鎖的過期時間if(redis.call(hexists, KEYS[1], ARGV[2])1)thenredis.call(pexpire, KEYS[1], ARGV[1]);return1;end;return0;到這里redisson實現(xiàn)分布式鎖的源碼解析就結(jié)束了。三、總結(jié)redisson的源碼中大量使用了異步編程這導(dǎo)致閱讀源碼的難度系數(shù)較高這里我也只是大概整理了一下有問題的同學(xué)可以互相討論一下或自行查閱源碼。