開啟掘金成長之旅！這是我參與「掘金日新計劃 · 2 月更文挑戰」的第 天，點選檢視活動詳情
前言：
日常開發中，我們經常會使用到鎖，以保證某一段邏輯是線程安全的，同步的。 但是當今一般都是同一個服務部署到多台機器上，在這種情況下，如果用java中的鎖，将隻能保證在某台機器上的線程安全，而不能保證真正意義上的線程安全，那麼此時分布式鎖就上場了。
下邊我們将以層層遞進的方式，看看怎麼用redis實作 相對完美的 分布式鎖！
- 多說一句：一把合格的分布式鎖，至少應該保證以下特性： 1:互斥性 (保證鎖内代碼邏輯同步) 2:逾時自動釋放（防止死鎖，占用資源） 3:安全性（不能删除非自己加的鎖） 4:原子（放值和取值時保證原子） 5:高可用，高性能 6:支援可重入 7:支援自動續期 複制代碼
1. setnx + expire 實作一個簡陋的鎖
setnx 指令解釋：如果資料庫不存在給定的key， 則設定成功傳回true，如果存在則設定不成功，傳回false，
複制代碼 值的注意的是在一些極端情況下（比如鎖中代碼執行時間過長，或者沒有成功删除鎖對應的key,此時将會一直占用資源（也就是死鎖），其他線程擷取不到鎖，會出現很嚴重的問題）？是以我們給這個key加個過期時間以便給這個key加一個”限制”，使得資源及時釋放，僞代碼如下：
if ( setnx( key1, value1)==1 ){
expire(key1);
try {
//TODO 執行業務代碼...
} finally {
del(key);
}
}else{
//未擷取到鎖
}
複制代碼 1.1 setnx與expire非原子問題以及解決
but這樣存在一個問題（因為setnx和expire不是原子操作），也就代表，在某一時刻，可能setx執行了，但是expire沒執行。此時就又有可能出現死鎖問題。在業界用的最多的一種就是使用lua腳本解決該問題，即通過lua将setnx和expire兩個指令變成原子操作。如下：
if (redis.call('setnx',KEYS[1],ARGV[1]) < 1) then return 0; end; redis.call('expire',KEYS[1],tonumber(ARGV[2])); return 1;" 1 key value 100
複制代碼 從上腳本可以看出，調用setnx如果設定成功則緊接着調用expire指令設定過期時間。
1.2 B鎖 被A删情況以及解決
Ok,上邊的lua腳本是保證了原子，但是删除時候又出現問題了。
- 考慮如下場景：
- 線程A 搶到鎖，執行業務邏輯，但是超過了鎖的過期時間（比如：10秒）也沒執行完，此時由于到了過期時間，是以redis中會删除該key,此時線程B恰好過來搶到了鎖，在執行過程中，線程A 執行完了，此時就會進入finally中執行del(key)，于是線程A把線程B加的鎖給删除了。。。
- 實時上，A删B鎖這種場景不僅是加鎖解鎖邏輯的混亂，還會導緻并發，即線程A删除B加的鎖後，設想此時線程C過來并且搶到鎖，那麼線程c也會執行鎖中代碼快（假設此時線程B還沒結束），也就是說有線程B 線程C并發執行鎖中代碼，此時鎖已經不起作用，形同虛設。。。（士可忍孰不可忍）
為了避免這種嚴重問題，我們的解決方式是：在某個線程去設定鎖的時候，value給一個uuid,當删除key時候，我們先get出來然後和設定時候比對一下，如果等于那就說明是目前線程設定的鎖，執行del,否則就說明不是，則不進行del,進而避免A鎖B删這種現象發生。
but此時又有個問題，就是解鎖時候get操作和del操作不是原子的，那麼我們還可以使用lua來解決這個問題，如下：
if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end
複制代碼 ok到這裡是不是感覺沒問題了？答案是：no！
在實際場景中，我們可能對分布式鎖有其他的要求，比如一些特殊場景下需要支援：
- 可重入： （即線程a拿到鎖後，在持有鎖期間，再次去申請這把鎖是可以申請成功的（個人認為這種情況很少遇到，我能想到的有：1：遞歸邏輯。2：鎖中代碼邏輯再次請求和外部同key的鎖（ps:真實場景似乎沒遇到，我們這裡僅僅讨論理論上的可能性） ））
- 自動續期： 在某些場景下，需要保證鎖逾時後，要自動續期，以保證沒執行完的邏輯繼續執行。
等特性，我們這裡不再班門弄斧，直接來看下業内大拿：Redisson是怎麼做的。
2. redisson鎖實作
作為redis用戶端架構 redisson不止提供了開箱即用的分布式鎖api,還包括其他很多特性這裡不再展開，更多請關注：redisson官網 ，我們隻是關注redisson是如何實作的分布式鎖的。
首先我們要知道的是 redisson實作的分布式鎖api,基本上是一把合格的鎖，他保證了我們開篇說的那幾個特性，如下：
1:互斥性 (保證鎖内代碼邏輯同步)
2:逾時自動釋放（防止死鎖，占用資源）
3:安全性（不能删除非自己加的鎖）
4:原子（放值和取值時保證原子）
5:高可用，高性能
6:支援可重入
7:支援自動續期
複制代碼 要想分析redisson如何實作，那必然離不了源碼（注意：這裡我們隻是簡單過一遍加鎖，解鎖，重入性，自動續期等實作邏輯,其他略過）
2.1 加鎖
廢話不多說，我們直接來到加鎖腳本這，如下：
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then " + //先檢查下給定的key 存不存在，不存在進入下邊分支邏輯
//如果不存在，那麼往redis寫入一個hash結構的資料(hincrby指令如果檢測到沒有key時，會寫入)，
//其中key是調用者傳入的key,value是map類型，其中key是用戶端id(看源碼的話知道他是一個uuid每一個連接配接都保持唯一):threid ，
//value就是某個線程的擷取鎖的次數（這個操作是實作可重入的保障）
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +//給剛剛設定的key 設定過期時間
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + //如果給定的key存在,
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +//将該key中對應的value值的value值 +1 操作，以記錄某個key下 某個線程的重入次數
"redis.call('pexpire', KEYS[1], ARGV[1]); " +//重入後，更新過期時間為給定值
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);", //沒擷取到鎖，則傳回鎖的剩餘時間
Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}
protected String getLockName(long threadId) {
return id + ":" + threadId;//這裡的id是CommandAsyncExecutor實作類中的id,本質上和某一條連接配接一一對應，值是uuid
}
複制代碼 讀完上邊腳本代碼和注釋，我們知道redisson
- 通過lua腳本來保證多指令原子性，
- 通過hash結構來存儲鎖資訊 鎖 結構如下： { 調用方給定的key : {redisson用戶端id（是個uuid，每個連接配接唯一）+擷取到鎖的線程id : 該線程擷取到的鎖次數 } } 複制代碼
- 通過判斷目前線程是否持有鎖，持有的話 value值+1，來記錄某個線程擷取鎖的次數，進而間接實作了重入的特性
一張圖直覺看下加鎖流程：
2.2 解鎖
廢話不多說直接上代碼
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +//不存在key 則直接傳回，不做操作
"return nil;" +
"end; " +
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +//否則根據key和 目前連接配接id+目前線程id 将value值減一
"if (counter > 0) then " +//如果減一後value大于0 說明該線程重入過，
"redis.call('pexpire', KEYS[1], ARGV[2]); " +//将該線程持有的鎖的過期時間更新
"return 0; " +
"else " +//counter不大于0，則删除key 釋放鎖
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +//釋出事件，廣播給其他訂閱該事件的線程，通知他們，”可以嘗試搶鎖啦！“
"return 1; " +//傳回釋放鎖成功
"end; " +
"return nil;",
Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}
複制代碼 從上邊可以看到:
- 解鎖時，如果重入過n次，那麼就需要釋放n次，直到value值為0 才真正删除 key。
- 另外 在删除時候，我們可以看到先判斷存在與否，而這裡的判斷是根據給定key和連接配接id+線程id來的，也就是說，不會發生B鎖A删的情況出現。
- 同時在删除鎖後，釋出事件，告知其他等待線程可以搶鎖啦！
2.3 自動續期
我們來到 tryAcquireAsync 方法，如下：
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
RFuture<Long> ttlRemainingFuture;
if (leaseTime != -1) {
ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
} else {
ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
}
//加鎖成功
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// lock acquired
if (ttlRemaining == null) {
if (leaseTime != -1) {
internalLockLeaseTime = unit.toMillis(leaseTime);
} else {
//可以看到如果leaseTime == -1 （也就是沒有傳leaseTime參數）那麼就會自動續期
scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
續期代碼：
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +//該線程持有鎖，
"redis.call('pexpire', KEYS[1], ARGV[1]); " +// 重置過期時間為 internalLockLeaseTime 即 30秒（redisson代碼中預設的，但可以配置）
"return 1; " +
"end; " +
"return 0;",
Collections.singletonList(getRawName()),
internalLockLeaseTime, getLockName(threadId));
}
複制代碼 自動續期内部源碼不在展開，我們簡單介紹下主要流程：
隻要線程加鎖成功并且 沒傳leaseTime值（leaseTime我喜歡叫它租約時長，這個參數很重要！！！，在使用時候一定要注意，如果調用tryLock時，指定了租約時長，那麼redisson看門狗機制将不起作用）此時該值會預設為 -1 ，就會啟動一個timer線程（也有人稱為watch dog看門狗機制），它是一個定時任務（基于netty的Timer類實作的），會每隔10（internalLockLeaseTime（預設為30）/3 =10 ）秒檢查一下，如果線程A還持有鎖，那麼就會不斷的延長鎖key的生存時間。進而實作了 自動續期。
注意：一旦你開啟了自動續期（watch dog機制），那麼一定要合理控制鎖中邏輯的執行時間，避免執行時間過長（事實上，大多數的邏輯我們都應該盡可能的快速執行完畢）。
ok到這裡我們用一張圖，來直覺看下redisson整體的一個流程：
2.4 叢集情況下存在的問題以及RedLock
到這redisson我們就說完了，但是有人會說，使用redisson這個分布式鎖肯定就能保證完美無瑕嗎？
- 單機版 redisson 已經 yyds ，很完美了至少我個人覺得
- 叢集 redisson 還是有問題，鎖并不一定安全，不能真正保證互斥 考慮如下場景： 如果線程A在Redis的master節點上拿到了鎖，但是加鎖的key還沒同步到slave節點。恰好這時， master節點發生故障，一個slave節點就會更新為master節點。線程B就可以擷取同個key的鎖啦，但線程A 也已經拿到鎖了，此時兩個不同線程都拿到鎖，并發執行鎖中代碼邏輯，鎖的安全性，互斥性也就蕩然無存了！ 複制代碼
基于上述叢集存在的問題，Redis作者 antirez 大佬提出一種進階的分布式鎖算法：Redlock。核心思想如下：
不能隻在一個redis執行個體上建立鎖，應該是在多個redis執行個體上建立鎖( 意味着那就要部署多個master )，并且必須在 (n/2)+1 個master節點上都成功建立鎖，才能算這個整體的RedLock加鎖成功，避免說僅僅在一個redis執行個體上加鎖，然後同步到slave時帶來的問題。
而redisson是實作了RedLock了的，其實作也很簡單，即周遊所有的Redis用戶端，然後依次加鎖（也得看redis叢集是什麼模式的 哨兵應該是不支援RedLock的，因為slave不能由用戶端直接寫），最後統計成功的次數來判斷是否加鎖成功。
public class RedissonMultiLock implements RLock
複制代碼 該類RedissonMultiLock即是RedLock的實作。具體如何使用以及有哪些坑我們不去展開，（ps:個人感覺RedLock的思想有點投入與産出不成正比，如果業務一定保證穩定性，互斥，安全，且并發不算太高時候，也可以用zk實作的分布式鎖，沒必要這麼費時費力，還得搭建多個master）
到此redis分布式鎖就寫完了，我們一般工作中也不造輪子，直接用redisson就好了，但是一定要去了解裡邊的實作，否則很容易踩坑呀！