對不起,網(wǎng)上找的Redis分布式鎖都有漏洞!

基于 Redis 的分布式鎖對大家來說并不陌生，可是你的分布式鎖有失敗的時候嗎？在失敗的時候可曾懷疑過你在用的分布式鎖真的靠譜嗎？以下是結合自己的踩坑經(jīng)驗總結的一些經(jīng)驗之談。

你真的需要分布式鎖嗎？

• 提高效率。比如多個節(jié)點計算同一批任務，如果某個任務已經(jīng)有節(jié)點在計算了，那其他節(jié)點就不用重復計算了，以免浪費計算資源。不過重復計算也沒事，不會造成其他更大的損失。也就是允許偶爾的失敗。

• 保證正確性。這種情況對鎖的要求就很高了，如果重復計算，會對正確性造成影響。這種不允許失敗。

從一個簡單的分布式鎖實現(xiàn)說起

最簡單的實現(xiàn)

• 加鎖和解鎖的鎖必須是同一個，常見的解決方案是給每個鎖一個鑰匙（唯一 ID），加鎖時生成，解鎖時判斷。

• 不能讓一個資源永久加鎖。常見的解決方案是給一個鎖的過期時間。當然了還有其他方案，后面再說。

public?static?boolean?tryLock(String?key,?String?uniqueId,?int?seconds)?{
????return?"OK".equals(jedis.set(key,?uniqueId,?"NX",?"EX",?seconds));
}

https://redis.io/commands/set

public?static?boolean?releaseLock(String?key,?String?uniqueId)?{
????String?luaScript?=?"if?redis.call('get',?KEYS[1])?==?ARGV[1]?then?"?+
????????????"return?redis.call('del',?KEYS[1])?else?return?0?end";
????return?jedis.eval(
????????luaScript,?
????????Collections.singletonList(key),?
????????Collections.singletonList(uniqueId)
????).equals(1L);
}

靠譜嗎？

• 單點問題。上面的實現(xiàn)只要一個 Master 節(jié)點就能搞定，這里的單點指的是單 Master，就算是個集群，如果加鎖成功后，鎖從 Master 復制到 Slave 的時候掛了，也是會出現(xiàn)同一資源被多個 Client 加鎖的。

• 執(zhí)行時間超過了鎖的過期時間。上面寫到為了不出現(xiàn)一直上鎖的情況，加了一個兜底的過期時間，時間到了鎖自動釋放，但是，如果在這期間任務并沒有做完怎么辦？由于 GC 或者網(wǎng)絡延遲導致的任務時間變長，很難保證任務一定能在鎖的過期時間內(nèi)完成。

Redlock?算法

分布式鎖的坑

高并發(fā)場景下的問題

節(jié)點宕機

• 假設有 5 個 Redis 的節(jié)點：A、B、C、D、E，沒有做持久化。

• Client1 從 A、B、C 這3 個節(jié)點獲取鎖成功，那么 client1 獲取鎖成功。

• 節(jié)點 C 掛了。

• Client2 從 C、D、E 獲取鎖成功，client2 也獲取鎖成功，那么在同一時刻 Client1 和 Client2 同時獲取鎖，Redlock 被玩壞了。

任務執(zhí)行時間超過鎖的 TTL

• Client1 獲取到鎖。

• Client1 開始任務，然后發(fā)生了 STW 的 GC，時間超過了鎖的過期時間。

• Client2 獲取到鎖，開始了任務。

• Client1 的 GC 結束，繼續(xù)任務，這個時候 Client1 和 Client2 都認為自己獲取了鎖，都會處理任務，從而發(fā)生錯誤。

?public?class?RedissonLock?extends?RedissonExpirable?implements?RLock?{
????...
????@Override
????public?void?lock()?{
????????try?{
????????????lockInterruptibly();
????????}?catch?(InterruptedException?e)?{
????????????Thread.currentThread().interrupt();
????????}
????}

????@Override
????public?void?lock(long?leaseTime,?TimeUnit?unit)?{
????????try?{
????????????lockInterruptibly(leaseTime,?unit);
????????}?catch?(InterruptedException?e)?{
????????????Thread.currentThread().interrupt();
????????}
????}
????...
?}

public?class?RedissonLock?extends?RedissonExpirable?implements?RLock?{

????...

????public?static?final?long?LOCK_EXPIRATION_INTERVAL_SECONDS?=?30;
????protected?long?internalLockLeaseTime?=?TimeUnit.SECONDS.toMillis(LOCK_EXPIRATION_INTERVAL_SECONDS);


????@Override
????public?void?lock()?{
????????try?{
????????????lockInterruptibly();
????????}?catch?(InterruptedException?e)?{
????????????Thread.currentThread().interrupt();
????????}
????}

????@Override
????public?void?lockInterruptibly()?throws?InterruptedException?{
????????lockInterruptibly(-1,?null);
????}

????@Override
????public?void?lockInterruptibly(long?leaseTime,?TimeUnit?unit)?throws?InterruptedException?{
????????long?threadId?=?Thread.currentThread().getId();
????????Long?ttl?=?tryAcquire(leaseTime,?unit,?threadId);
????????//?lock?acquired
????????if?(ttl?==?null)?{
????????????return;
????????}

????????RFuture?future?=?subscribe(threadId);
????????commandExecutor.syncSubscription(future);

????????try?{
????????????while?(true)?{
????????????????ttl?=?tryAcquire(leaseTime,?unit,?threadId);
????????????????//?lock?acquired
????????????????if?(ttl?==?null)?{
????????????????????break;
????????????????}

????????????????//?waiting?for?message
????????????????if?(ttl?>=?0)?{
????????????????????getEntry(threadId).getLatch().tryAcquire(ttl,?TimeUnit.MILLISECONDS);
????????????????}?else?{
????????????????????getEntry(threadId).getLatch().acquire();
????????????????}
????????????}
????????}?finally?{
????????????unsubscribe(future,?threadId);
????????}
//????????get(lockAsync(leaseTime,?unit));
????}

????private?Long?tryAcquire(long?leaseTime,?TimeUnit?unit,?long?threadId)?{
????????return?get(tryAcquireAsync(leaseTime,?unit,?threadId));
????}

????private??RFuture?tryAcquireAsync(long?leaseTime,?TimeUnit?unit,?final?long?threadId)?{
????????if?(leaseTime?!=?-1)?{
????????????return?tryLockInnerAsync(leaseTime,?unit,?threadId,?RedisCommands.EVAL_LONG);
????????}
????????RFuture?ttlRemainingFuture?=?tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS,?TimeUnit.SECONDS,?threadId,?RedisCommands.EVAL_LONG);
????????ttlRemainingFuture.addListener(new?FutureListener()?{
????????????@Override
????????????public?void?operationComplete(Future?future)?throws?Exception?{
????????????????if?(!future.isSuccess())?{
????????????????????return;
????????????????}

????????????????Long?ttlRemaining?=?future.getNow();
????????????????//?lock?acquired
????????????????if?(ttlRemaining?==?null)?{
????????????????????scheduleExpirationRenewal(threadId);
????????????????}
????????????}
????????});
????????return?ttlRemainingFuture;
????}

????private?void?scheduleExpirationRenewal(final?long?threadId)?{
????????if?(expirationRenewalMap.containsKey(getEntryName()))?{
????????????return;
????????}

????????Timeout?task?=?commandExecutor.getConnectionManager().newTimeout(new?TimerTask()?{
????????????@Override
????????????public?void?run(Timeout?timeout)?throws?Exception?{

????????????????RFuture?future?=?commandExecutor.evalWriteAsync(getName(),?LongCodec.INSTANCE,?RedisCommands.EVAL_BOOLEAN,
????????????????????????"if?(redis.call('hexists',?KEYS[1],?ARGV[2])?==?1)?then?"?+
????????????????????????????"redis.call('pexpire',?KEYS[1],?ARGV[1]);?"?+
????????????????????????????"return?1;?"?+
????????????????????????"end;?"?+
????????????????????????"return?0;",
??????????????????????????Collections.