Redisson分布式鎖的原理及分析

更新時(shí)間：2026年05月27日 14:38:01 作者：無語堵上西樓

分布式鎖的關(guān)鍵問題包括續(xù)期、釋放、可重入和互斥,Redisson提供完善的解決方案,通過WatchDog機(jī)制自動(dòng)續(xù)期,支持分布式鎖的重入和高效釋放,避免死鎖,提升資源利用率

分布式鎖就要考慮鎖的續(xù)期、釋放、可重入、互斥等問題。Redisson這個(gè)客戶端是目前最完美的一種方案，它在內(nèi)部可以對(duì)鎖進(jìn)行自動(dòng)續(xù)期，程序執(zhí)行結(jié)束、發(fā)生異常或者整個(gè)應(yīng)用掛掉都可以釋放鎖，可重入和互斥也都處理的很好。

有Redisson了，我們沒必要自己手寫分布式鎖了，手寫的分布式鎖不如Redisson考慮的全面的。

Redisson分布式鎖方案優(yōu)點(diǎn)

Redisson 通過 Watch Dog（看門狗）機(jī)制很好的解決了鎖的續(xù)期問題。
通過 Redisson 實(shí)現(xiàn)分布式可重入鎖，比原生的SET mylock userId NX PX milliseconds + lua 實(shí)現(xiàn)的效果更好。
在進(jìn)程等待申請(qǐng)鎖的實(shí)現(xiàn)上也做了一些優(yōu)化，減少了無效的鎖申請(qǐng)，提升了資源的利用率。

分布式鎖要注意的問題

安全屬性（Safety property）: 獨(dú)享（相互排斥）

在任意一個(gè)時(shí)刻，只有一個(gè)客戶端持有鎖。

活性A(Liveness property A): 無死鎖

即便持有鎖的客戶端崩潰（crashed)或者網(wǎng)絡(luò)被分裂（gets partitioned)，鎖仍然可以被獲取。

活性B(Liveness property B): 容錯(cuò)

只要大部分Redis節(jié)點(diǎn)都活著，客戶端就可以獲取和釋放鎖

為什么Redisson不用setnx實(shí)現(xiàn)分布式鎖？

Redisson沒有使用setnx命令實(shí)現(xiàn)分布式鎖，因?yàn)殡m然setnx命令能夠?qū)崿F(xiàn)分布式鎖，但存在以下幾個(gè)問題：

鎖過期時(shí)間不能自動(dòng)續(xù)約。使用setnx命令實(shí)現(xiàn)分布式鎖時(shí)，如果獲取鎖的客戶端執(zhí)行時(shí)間過長，導(dǎo)致鎖過期，其它客戶端就有可能獲取到這個(gè)鎖。因此需要加入自動(dòng)續(xù)約機(jī)制，在鎖的持有者自身沒有釋放鎖的情況下，對(duì)鎖進(jìn)行續(xù)約以保證該鎖持續(xù)生效。
不支持可重入。如果某個(gè)線程已經(jīng)持有了一個(gè)鎖，再次對(duì)這個(gè)鎖進(jìn)行加鎖時(shí)，setnx命令會(huì)認(rèn)為這個(gè)鍵已經(jīng)存在，無法再次進(jìn)行加鎖。而支持可重入的鎖允許同一線程多次加鎖，且要求解鎖次數(shù)與加鎖次數(shù)相等。
不支持鎖的釋放。setnx命令只能通過手動(dòng)設(shè)置過期時(shí)間或者等待過期時(shí)間釋放鎖。如果某個(gè)線程異常退出或者未能及時(shí)釋放鎖，就有可能導(dǎo)致死鎖的發(fā)生。而支持釋放鎖的鎖允許設(shè)置鎖的自動(dòng)釋放時(shí)間或者手動(dòng)釋放鎖。

Redisson提供的分布式鎖可以解決以上三個(gè)問題，并提供了更為完善的分布式鎖功能，使得使用Redisson實(shí)現(xiàn)分布式鎖更加方便和穩(wěn)定。

例如：使用Lua腳本來保證原子性，使用Redis的watch機(jī)制來實(shí)現(xiàn)分布式鎖的釋放，使用watchdog機(jī)制實(shí)現(xiàn)分布式鎖的續(xù)期。

原理分析

用法

RLock lock = redisson.getLock("myLock");
lock.lock();
try {
    // do sth.
} finally {
    lock.unlock();
}

獲取鎖，若獲取不成功則訂閱釋放鎖的消息，在收到釋放鎖的消息前阻塞，收到釋放鎖的消息后再去循環(huán)獲取鎖。

代碼調(diào)用流程

lock()          //org.redisson.RedissonLock.java#lock
    lock(-1, null, false)
        tryAcquire(-1, leaseTime, unit, threadId)
            tryAcquireAsync(waitTime, leaseTime, unit, threadId)
                tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);  //這是最重要的方法

lock(-1, null, false)

總結(jié)：獲取鎖，若獲取不成功則訂閱釋放鎖的消息，在收到釋放鎖的消息前阻塞，收到釋放鎖的消息后再去循環(huán)獲取鎖。

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
        long threadId = Thread.currentThread().getId();
		// 獲取鎖
        Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
        // 獲取成功
        if (ttl == null) {
            return;
        }
		// 異步訂閱redis channel
        RFuture<RedissonLockEntry> future = subscribe(threadId);
        if (interruptibly) {
            commandExecutor.syncSubscriptionInterrupted(future);
        } else {
            commandExecutor.syncSubscription(future);
        }
        try {
            while (true) {
                ttl = tryAcquire(-1, leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    break;
                }
                // waiting for message
                if (ttl >= 0) {
                    try {
                        future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        if (interruptibly) {
                            throw e;
                        }
                        future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    }
                } else {
                    if (interruptibly) {
                        future.getNow().getLatch().acquire();
                    } else {
                        future.getNow().getLatch().acquireUninterruptibly();
                    }
                }
            }
        } finally {
			// 取消訂閱
            unsubscribe(future, threadId);
        }
//        get(lockAsync(leaseTime, unit));
    }

tryAcquire(leaseTime, unit, threadId)

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
    return get(tryAcquireAsync(leaseTime, unit, threadId));// 通過異步獲取鎖，但get(future)實(shí)現(xiàn)同步
}

tryAcquireAsync(waitTime, leaseTime, unit, threadId)

總結(jié)：用到了Netty的Future-listen模型：給Future一個(gè)Promise。

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
    if (leaseTime != -1) { //1 如果設(shè)置了超時(shí)時(shí)間，直接調(diào)用 tryLockInnerAsync
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    //2 如果leaseTime==-1，則默認(rèn)超時(shí)時(shí)間為30s
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS, TimeUnit.SECONDS, threadId, RedisCommands.EVAL_LONG);
    //3 監(jiān)聽Future，獲取Future返回值ttlRemaining(剩余超時(shí)時(shí)間)，獲取鎖成功，但是ttlRemaining，則刷新過期時(shí)間
    ttlRemainingFuture.addListener(new FutureListener<Long>() {
        @Override
        public void operationComplete(Future<Long> future) throws Exception {
            if (!future.isSuccess()) {
                return;
            }
            Long ttlRemaining = future.getNow();
            // lock acquired
            if (ttlRemaining == null) {
                scheduleExpirationRenewal(threadId);
            }
        }
    });
    return ttlRemainingFuture;
}

tryLockInnerAsync

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);
    return commandExecutor.evalWriteAsync(
        getName(),
        LongCodec.INSTANCE,
        command,
          "if (redis.call('exists', KEYS[1]) == 0) then " +
              "redis.call('hset', KEYS[1], ARGV[2], 1); " +
              "redis.call('pexpire', KEYS[1], ARGV[1]); " +
              "return nil; " +
          "end; " +
          "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
              "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
              "redis.call('pexpire', KEYS[1], ARGV[1]); " +
              "return nil; " +
          "end; " +
          "return redis.call('pttl', KEYS[1]);",
        Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

腳本入?yún)?/h3>

參數(shù)	示例值	含義
KEY個(gè)數(shù)	1	KEY個(gè)數(shù)。為了后面可重入做的計(jì)數(shù)統(tǒng)計(jì)。
KEYS[1]	“myLock”	加鎖的 key 名字
ARGV[1]	60000	持有鎖的有效時(shí)間：毫秒。默認(rèn) 30 秒
ARGV[2]	285475da-9152-4c83-822a-67ee2f116a79:52	唯一標(biāo)識(shí)：Redisson客戶端ID(UUID)+線程ID

看一下在 Redis 中的存儲(chǔ)結(jié)構(gòu)：

127.0.0.1:6379> HGETALL myLock
1) "285475da-9152-4c83-822a-67ee2f116a79:52"
2) "1"

腳本解讀

-- 若鎖不存在：新增鎖，設(shè)置鎖重入計(jì)數(shù)為1，設(shè)置鎖過期時(shí)間，返回
if (redis.call('exists', KEYS[1]) == 0) then
    redis.call('hset', KEYS[1], ARGV[2], 1);
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return nil;
end;
-- 若鎖存在，且唯一標(biāo)識(shí)也匹配：表明當(dāng)前加鎖請(qǐng)求為鎖重入請(qǐng)求，故鎖重入計(jì)數(shù)+1，再次設(shè)置鎖過期時(shí)間，返回
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
    redis.call('hincrby', KEYS[1], ARGV[2], 1);
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return nil;
end;
-- 若鎖存在，但唯一標(biāo)識(shí)不匹配：表明鎖是被其他線程占用，當(dāng)前線程無權(quán)解他人的鎖，直接返回鎖剩余過期時(shí)間
return redis.call('pttl', KEYS[1]);

互斥

如果客戶端 2 來嘗試加鎖，會(huì)如何呢？

首先，第一個(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)槟抢锇氖强蛻舳?1 的 ID。所以，客戶端 2 會(huì)執(zhí)行：

return redis.call('pttl', KEYS[1]);

返回的一個(gè)數(shù)字，這個(gè)數(shù)字代表了 myLock 這個(gè)鎖 key 的剩余生存時(shí)間。

看一下 Redissson tryLock 的主流程：

@Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        // 1.嘗試獲取鎖
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return true;
        }
        // 申請(qǐng)鎖的耗時(shí)如果大于等于最大等待時(shí)間，則申請(qǐng)鎖失敗.
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(threadId);
            return false;
        }
        current = System.currentTimeMillis();
        /**
         * 2.訂閱鎖釋放事件，并通過 await 方法阻塞等待鎖釋放，有效的解決了無效的鎖申請(qǐng)浪費(fèi)資源的問題：
         * 基于信息量，當(dāng)鎖被其它資源占用時(shí)，當(dāng)前線程通過 Redis 的 channel 訂閱鎖的釋放事件，
         * 一旦鎖釋放會(huì)發(fā)消息通知待等待的線程進(jìn)行競爭.
         *
         * 當(dāng) this.await 返回 false，說明等待時(shí)間已經(jīng)超出獲取鎖最大等待時(shí)間，取消訂閱并返回獲取鎖失敗.
         * 當(dāng) this.await 返回 true，進(jìn)入循環(huán)嘗試獲取鎖.
         */
        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        // await 方法內(nèi)部是用 CountDownLatch 來實(shí)現(xiàn)阻塞，獲取 subscribe 異步執(zhí)行的結(jié)果（應(yīng)用了 Netty 的 Future）
        if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.onComplete((res, e) -> {
                    if (e == null) {
                        unsubscribe(subscribeFuture, threadId);
                    }
                });
            }
            acquireFailed(threadId);
            return false;
        }
        try {
            // 計(jì)算獲取鎖的總耗時(shí)，如果大于等于最大等待時(shí)間，則獲取鎖失敗.
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
                acquireFailed(threadId);
                return false;
              }
            /**
             * 3.收到鎖釋放的信號(hào)后，在最大等待時(shí)間之內(nèi)，循環(huán)一次接著一次的嘗試獲取鎖
             * 獲取鎖成功，則立馬返回 true，
             * 若在最大等待時(shí)間之內(nèi)還沒獲取到鎖，則認(rèn)為獲取鎖失敗，返回 false 結(jié)束循環(huán)
             */
            while (true) {
                long currentTime = System.currentTimeMillis();
                // 4.再次嘗試獲取鎖
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    return true;
                }
                // 5.超過最大等待時(shí)間則返回 false 結(jié)束循環(huán)，獲取鎖失敗
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }
                /**
                 * 6.阻塞等待鎖（通過信號(hào)量(共享鎖)阻塞,等待解鎖消息）：
                 */
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    //如果剩余時(shí)間(ttl)小于wait time ,就在 ttl 時(shí)間內(nèi)，從Entry的信號(hào)量獲取
                    //一個(gè)許可(除非被中斷或者一直沒有可用的許可)。
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    //則就在wait time 時(shí)間范圍內(nèi)等待可以通過信號(hào)量
                    getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }
                // 更新剩余的等待時(shí)間(最大等待時(shí)間-已經(jīng)消耗的阻塞時(shí)間)
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }
            }
        } finally {
            // 7.無論是否獲得鎖,都要取消訂閱解鎖消息
            unsubscribe(subscribeFuture, threadId);
        }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
    }

流程分析

嘗試獲取鎖。若返回 null 則說明加鎖成功；若返回一個(gè)數(shù)值，則說明已經(jīng)存在該鎖，ttl 為鎖的剩余存活時(shí)間。
如果此時(shí)客戶端 2 進(jìn)程獲取鎖失敗，那么使用客戶端 2 的線程 id（其實(shí)本質(zhì)上就是進(jìn)程 id）通過 Redis 的 channel 訂閱鎖釋放的事件，。如果等待的過程中一直未等到鎖的釋放事件通知，當(dāng)超過最大等待時(shí)間則獲取鎖失敗，返回 false，也就是第 39 行代碼。如果等到了鎖的釋放事件的通知，則開始進(jìn)入一個(gè)不斷重試獲取鎖的循環(huán)。
循環(huán)中每次都先試著獲取鎖，并得到已存在的鎖的剩余存活時(shí)間。如果在重試中拿到了鎖，則直接返回。如果鎖當(dāng)前還是被占用的，那么等待釋放鎖的消息，具體實(shí)現(xiàn)使用了 JDK 的信號(hào)量 Semaphore 來阻塞線程，當(dāng)鎖釋放并發(fā)布釋放鎖的消息后，信號(hào)量的release()方法會(huì)被調(diào)用，此時(shí)被信號(hào)量阻塞的等待隊(duì)列中的一個(gè)線程就可以繼續(xù)嘗試獲取鎖了。

特別注意

以上過程存在一個(gè)細(xì)節(jié)，這里有必要說明一下，也是分布式鎖的一個(gè)關(guān)鍵點(diǎn)：當(dāng)鎖正在被占用時(shí)，等待獲取鎖的進(jìn)程并不是通過一個(gè) while(true) 死循環(huán)去獲取鎖，而是利用了 Redis 的發(fā)布訂閱機(jī)制，通過 await 方法阻塞等待鎖的進(jìn)程，有效的解決了無效的鎖申請(qǐng)浪費(fèi)資源的問題。

續(xù)期

客戶端 1 加鎖的鎖 key 默認(rèn)生存時(shí)間才 30 秒，如果超過了 30 秒，客戶端 1 還想一直持有這把鎖，怎么辦呢？

Redisson 提供了一個(gè)續(xù)期機(jī)制，只要客戶端 1 一旦加鎖成功，就會(huì)啟動(dòng)一個(gè) Watch Dog?？梢赃@樣來使用看門狗（leaseTime不設(shè)置，或者設(shè)置為-1）

lock.tryLock()
lock.tryLock(xxx, -1, xxx)

源碼

org.redisson.RedissonLock#tryAcquireAsync

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
    if (leaseTime != -1) {
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        if (e != null) {
            return;
        }
        // lock acquired
        if (ttlRemaining == null) {
            scheduleExpirationRenewal(threadId);
        }
    });
    return ttlRemainingFuture;
}

注意：從以上源碼我們看到 leaseTime 必須是 -1 才會(huì)開啟 Watch Dog 機(jī)制，也就是如果你想開啟 Watch Dog 機(jī)制必須使用默認(rèn)的加鎖時(shí)間為 30s。如果你自己自定義時(shí)間，超過這個(gè)時(shí)間，鎖就會(huì)自己釋放，并不會(huì)延長。

private void scheduleExpirationRenewal(long threadId) {
    ExpirationEntry entry = new ExpirationEntry();
    ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
    if (oldEntry != null) {
        oldEntry.addThreadId(threadId);
    } else {
        entry.addThreadId(threadId);
        renewExpiration();
    }
}
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
    return 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.<Object>singletonList(getName()),
        internalLockLeaseTime, getLockName(threadId));
}

Watch Dog 機(jī)制其實(shí)就是一個(gè)后臺(tái)定時(shí)任務(wù)線程。獲取鎖成功之后，會(huì)將持有鎖的線程放入到一個(gè) RedissonLock.EXPIRATION_RENEWAL_MAP里面，然后每隔 10 秒（internalLockLeaseTime / 3）檢查一下，如果客戶端 1 還持有鎖 key（判斷客戶端是否還持有 key，其實(shí)就是遍歷 EXPIRATION_RENEWAL_MAP 里面線程 id 然后根據(jù)線程 id 去 Redis 中查，如果存在就會(huì)延長 key 的時(shí)間），那么就會(huì)不斷的延長鎖 key 的生存時(shí)間。

注意：這里有一個(gè)細(xì)節(jié)：如果服務(wù)宕機(jī)了，Watch Dog 機(jī)制的線程也就沒有了，此時(shí)就不會(huì)延長 key 的過期時(shí)間，到了 30s 之后就會(huì)自動(dòng)過期了，其他線程就可以獲取到鎖。

可重入

Redisson 也是支持可重入鎖的，比如下面這種代碼：

@Override
public void lock() {
    RLock lock = redissonSingle.getLock("myLock");
    try {
        lock.lock();
        // 執(zhí)行業(yè)務(wù)
        doBusiness();
        lock.lock();
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        // 釋放鎖
        lock.unlock();
        lock.unlock();
        logger.info("任務(wù)執(zhí)行完畢, 釋放鎖!");
    }
}

我們再分析一下加鎖那段 lua 代碼：

if (redis.call('exists', KEYS[1]) == 0) then " +
   "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
   "redis.call('pexpire', KEYS[1], ARGV[1]); " +
   "return nil; " +
   "end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
    "return nil; " +
    "end; " +
"return redis.call('pttl', KEYS[1]);"

第一個(gè) if 判斷肯定不成立，exists myLock 會(huì)顯示鎖 key 已經(jīng)存在。第二個(gè) if 判斷會(huì)成立，因?yàn)?myLock 的 hash 數(shù)據(jù)結(jié)構(gòu)中包含的那個(gè) ID 即客戶端 1 的 ID，此時(shí)就會(huì)執(zhí)行可重入加鎖的邏輯，使用：hincrby myLock 285475da-9152-4c83-822a-67ee2f116a79:52 1對(duì)客戶端 1 的加鎖次數(shù)加 1。此時(shí) myLock 數(shù)據(jù)結(jié)構(gòu)變?yōu)橄旅孢@樣：

127.0.0.1:6379> HGETALL myLock
1) "285475da-9152-4c83-822a-67ee2f116a79:52"
2) "2"

到這里，小伙伴本就都明白了 hash 結(jié)構(gòu)的 key 是鎖的名稱，field 是客戶端 ID，value 是該客戶端加鎖的次數(shù)。

釋放

鎖的釋放利用了Redis的訂閱功能。

釋放鎖的步驟主要分三步：

刪除鎖（這里注意可重入鎖，在上面的腳本中有詳細(xì)分析）。
廣播釋放鎖的消息，通知阻塞等待的進(jìn)程（向通道名為 redisson_lock__channel publish 一條 UNLOCK_MESSAGE 信息）。
取消 Watch Dog 機(jī)制，即將 RedissonLock.EXPIRATION_RENEWAL_MAP 里面的線程 id 刪除，并且 cancel 掉 Netty 的那個(gè)定時(shí)任務(wù)線程。

源碼

執(zhí)行

lock.unlock()

就可以釋放分布式鎖。我們來看一下釋放鎖的流程代碼：

@Override
public RFuture<Void> unlockAsync(long threadId) {
    RPromise<Void> result = new RedissonPromise<Void>();
    // 1. 異步釋放鎖
    RFuture<Boolean> future = unlockInnerAsync(threadId);
    // 取消 Watch Dog 機(jī)制
    future.onComplete((opStatus, e) -> {
        cancelExpirationRenewal(threadId);
        if (e != null) {
            result.tryFailure(e);
            return;
        }
        if (opStatus == null) {
            IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                    + id + " thread-id: " + threadId);
            result.tryFailure(cause);
            return;
        }
        result.trySuccess(null);
    });
    return result;
}
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            // 判斷鎖 key 是否存在
            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                "return nil;" +
            "end; " +
            // 將該客戶端對(duì)應(yīng)的鎖的 hash 結(jié)構(gòu)的 value 值遞減為 0 后再進(jìn)行刪除
            // 然后再向通道名為 redisson_lock__channel publish 一條 UNLOCK_MESSAGE 信息
            "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
            "if (counter > 0) then " +
                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                "return 0; " +
            "else " +
                "redis.call('del', KEYS[1]); " +
                "redis.call('publish', KEYS[2], ARGV[1]); " +
                "return 1; "+
            "end; " +
            "return nil;",
            Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}