快捷導(dǎo)航

redis中RedissonLock如何實(shí)現(xiàn)等待鎖的

更新時(shí)間：2021年11月11日 15:41:54 作者：Fun_

本文主要介紹了redis中RedissonLock如何實(shí)現(xiàn)等待鎖的，文中通過(guò)示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

前言

經(jīng)常會(huì)有到這樣的需求，就是在一個(gè)查詢接口，第一次查詢的時(shí)候，如果沒(méi)有查詢到就要執(zhí)行初始化方法，初始化數(shù)據(jù)出來(lái)，之后的查詢就可以直接查詢庫(kù)里的數(shù)據(jù)了。這樣設(shè)計(jì)的目的是，如果需要初始化的數(shù)據(jù)特別大，無(wú)法再一次調(diào)用方法里處理完，或者說(shuō)數(shù)據(jù)并不是每條都需要初始化，這種情況下，優(yōu)先查詢的數(shù)據(jù)優(yōu)先初始化。

問(wèn)題

這種方案隨之而來(lái)就會(huì)引發(fā)一個(gè)問(wèn)題。查詢接口眾所周知是個(gè)自然冪等的，不需要我們額外去做冪等處理。但是在方案中，這個(gè)查詢就不單單是個(gè)查詢了。沒(méi)有查詢到就要執(zhí)行初始化方法，本質(zhì)上是個(gè)插入邏輯。這就需要我們自己去做冪等了。

方案

單臺(tái)服務(wù)，我們可以用Java的鎖來(lái)實(shí)現(xiàn)冪等，每條數(shù)據(jù)的主鍵id來(lái)當(dāng)鎖。但在現(xiàn)在基本上都是分布式服務(wù)，如同上篇文章說(shuō)的，我們可以用分布式鎖RedissonLock來(lái)實(shí)現(xiàn)。
并發(fā)第一次請(qǐng)求時(shí)，競(jìng)爭(zhēng)RedissonLock，誰(shuí)獲得了鎖，誰(shuí)就執(zhí)行初始化方法，沒(méi)有競(jìng)爭(zhēng)到鎖的請(qǐng)求，可以設(shè)置一個(gè)等待時(shí)間，等待鎖釋放。鎖釋放了，就可以先查詢數(shù)據(jù)有沒(méi)有初始化好，完成了就直接查庫(kù)。這里，就要提一下RedissonLock是如何實(shí)現(xiàn)等待的？

tryLock

RedissonLock在加鎖方法提供了一個(gè)api，提供了一個(gè)參數(shù)waitTime即等待時(shí)間。

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit)

在waitTime時(shí)間內(nèi)會(huì)訂閱消息，這里用的是redis本身的發(fā)布訂閱功能。

RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
    if (!subscribeFuture.cancel(false)) {
        subscribeFuture.onComplete((res, e) -> {
            if (e == null) {
                unsubscribe(subscribeFuture, threadId);
            }
        });
    }
    acquireFailed(threadId);
    return false;
}

這樣，在釋放鎖的時(shí)候，同時(shí)發(fā)布消息出來(lái)。所有監(jiān)聽(tīng)該鎖的線程都將收到通知，那么，這些線程將再次去競(jìng)爭(zhēng)加鎖，從而達(dá)到我們需要的冪等功能。我們?cè)诳纯瘁尫沛i的邏輯，是不是發(fā)布消息了？

unlockInnerAsync

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                "return nil;" +
            "end; " +
            "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));

}

可以看出在unlockInnerAsync方法里，執(zhí)行了lua腳本，在腳本里，我們很容易能看到執(zhí)行了publish命令。