一、基礎(chǔ)

0）Redisson版本說(shuō)明、案例

使用當(dāng)前（2022年12月初）最新的版本：3.18.1；

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.18.1</version>
</dependency>

案例

案例采用redis-cluster集群的方式；

public class Main {
    public static void main(String[] args) throws Exception {
        // 1.配置Redis-Cluster集群節(jié)點(diǎn)的ip和port 
        Config config = new Config();
        config.useClusterServers()
                .addNodeAddress("redis://127.0.0.1:7001")
                .addNodeAddress("redis://127.0.0.1:7002")
                .addNodeAddress("redis://127.0.0.1:7003")
                .addNodeAddress("redis://127.0.0.1:7004");
        // 2.創(chuàng)建Redisson的客戶(hù)端 
        RedissonClient redisson = Redisson.create(config);
        // 3.測(cè)試Redisson可重?鎖的加鎖、釋放鎖
        testLock(redisson);
    }

    private static void testLock(RedissonClient redisson) throws InterruptedException {
        // 1.獲取key為"anyLock"的鎖對(duì)象
        final RLock lock = redisson.getLock("test_lock");
        boolean locked = true;
        try {
            //2.1：加鎖 
            lock.lock();
            // 2.2：加鎖，并設(shè)置嘗試獲取鎖超時(shí)時(shí)間30s、鎖超時(shí)?動(dòng)釋放的時(shí)間10s 
//            locked = lock.tryLock(30, 10, TimeUnit.SECONDS);
            if (locked)
                System.out.println("加鎖成功！" + new Date());
            
            Thread.sleep(20 * 1000);
            System.out.println("鎖邏輯執(zhí)行完畢！" + new Date());

        } finally {
            // 3.釋放鎖 
            lock.unlock();
        }
    }
}

1）Redisson連接Redis的方式

redission支持4種連接redis方式，分別為單機(jī)、主從、Sentinel、Cluster 集群；在分布式鎖的實(shí)現(xiàn)上區(qū)別在于hash槽的獲取方式。

具體配置方式見(jiàn)Redisson的GitHub（https://github.com/redisson/redisson/wiki/2.-%E9%85%8D%E7%BD%AE%E6%96%B9%E6%B3%95#21-%E7%A8%8B%E5%BA%8F%E5%8C%96%E9%85%8D%E7%BD%AE%E6%96%B9%E6%B3%95）

2）用到的Redis命令

分布式鎖主要需要以下redis命令：

EXISTS key：當(dāng) key 存在，返回1；不存在，返回0。

GETSET key value：將給定 key 的值設(shè)為 value ，并返回 key 的舊值 (old value)；當(dāng) key 存在但不是字符串類(lèi)型時(shí)，返回一個(gè)錯(cuò)誤；當(dāng)key不存在時(shí)，返回nil。

GET key：返回 key 所關(guān)聯(lián)的字符串值，如果 key 不存在那么返回 nil。

DEL key [KEY …]：刪除給定的一個(gè)或多個(gè) key（不存在的 key 會(huì)被忽略），返回實(shí)際刪除的key的個(gè)數(shù)（integer）。

DEL key1 key2 key3

HSET key field value：給一個(gè)key 設(shè)置一個(gè){field=value}的組合值，如果key沒(méi)有就直接賦值并返回1；如果field已有，那么就更新value的值，并返回0。

HEXISTS key field：當(dāng)key中存儲(chǔ)著field的時(shí)候返回1，如果key或者field有一個(gè)不存在返回0。

HINCRBY key field increment：將存儲(chǔ)在key中的哈希（Hash）對(duì)象中的指定字段field的值加上增量increment；

如果鍵key不存在，一個(gè)保存了哈希對(duì)象{field=value}的key將被創(chuàng)建；如果字段field不存在，在進(jìn)行當(dāng)前操作前，feild將被創(chuàng)建，且對(duì)應(yīng)的值被置為0；返回值是increment。

PEXPIRE key milliseconds：設(shè)置存活時(shí)間，單位是毫秒。EXPIRE操作單位是秒。

PUBLISH channel message：向channel post一個(gè)message內(nèi)容的消息，返回接收消息的客戶(hù)端數(shù)。

3）用到的lua腳本語(yǔ)義

Redisson源碼中，執(zhí)行redis命令的是lua腳本，其中主要有如下幾個(gè)概念：

• redis.call()：執(zhí)行redis命令。
• KEYS[n]：指腳本中第n個(gè)參數(shù)，比如KEYS[1]指腳本中的第一個(gè)參數(shù)。
• ARGV[n]：指腳本中第n個(gè)參數(shù)的值，比如ARGV[1]指腳本中的第一個(gè)參數(shù)的值。
• 返回值中nil與false同一個(gè)意思。

在redis執(zhí)行l(wèi)ua腳本時(shí)，相當(dāng)于一個(gè)redis級(jí)別的鎖，不能執(zhí)行其他操作，類(lèi)似于原子操作，這也是redisson實(shí)現(xiàn)的一個(gè)關(guān)鍵點(diǎn)。

另外，如果lua腳本執(zhí)行過(guò)程中出現(xiàn)了異常或者redis服務(wù)器宕機(jī)了，會(huì)將腳本中已經(jīng)執(zhí)行的命令在AOF、RDB日志中刪除；即LUA腳本執(zhí)行報(bào)錯(cuò)會(huì)進(jìn)行回滾操作。

二、源碼分析

1、RLock

RLock接口主要繼承了Lock接口，并擴(kuò)展了部分方法，比如：tryLock(long waitTime, long leaseTime, TimeUnit unit)方法中加入的leaseTime參數(shù)，用來(lái)設(shè)置鎖的過(guò)期時(shí)間，如果超過(guò)leaseTime還沒(méi)有解鎖的話，redis就強(qiáng)制解鎖；leaseTime的默認(rèn)時(shí)間是30s。

獲取RLock對(duì)象

RLock lock = redissonClient.getLock("test_lock");

RLock對(duì)象表示?個(gè)鎖對(duì)象，我們要某一個(gè)key加鎖時(shí)，需要先獲取?個(gè)鎖對(duì)象。

這里并沒(méi)有具體請(qǐng)求Redis進(jìn)行加鎖的邏輯，而只是調(diào)用RedissonLock的構(gòu)造函數(shù)，設(shè)置一些變量。

2、加鎖流程

進(jìn)入到Rlock#lock()方法，先看主流程；關(guān)于競(jìng)爭(zhēng)鎖等待時(shí)間、鎖超時(shí)釋放時(shí)間的配置、使用，在流程中穿插著聊。

0）加鎖流程圖

1）加鎖到哪臺(tái)機(jī)器

lock()方法執(zhí)行鏈路：

走到這里，已經(jīng)可以看到加鎖的底層邏輯：LUA腳本。

而lua腳本只是??串字符串，作為evalWriteAsync()?法的?個(gè)參數(shù)?已；所以下?步進(jìn)到evalWriteAsync()?法中：

走到這里會(huì)調(diào)用ConnectionManager#getEntry(String)方法；

在創(chuàng)建RedissonClient時(shí)，筆者配置的是Redis-Cluster，而走到這里卻會(huì)進(jìn)入到MasterSlaveConnectionManager，實(shí)際上實(shí)例化的ConnectionManager就是RedisCluster模式下的ClusterConnectionManager，而ClusterConnectionManager繼承自MasterSlaveConnectionManager，并且ClusterConnectionManager沒(méi)有重寫(xiě)getEntry(String)方法，所以會(huì)進(jìn)入到MasterSlaveConnectionManager#getEntry(String)方法。

ConnectionManager#getEntry(String)方法會(huì)根據(jù)傳入的key名稱(chēng)找到相應(yīng)的Redis節(jié)點(diǎn)、目標(biāo)master。

Redis-Cluster集群中的數(shù)據(jù)分布式是 通過(guò)?個(gè)?個(gè)的hash slot來(lái)實(shí)現(xiàn)的，Redis-Cluster集群總共16384個(gè)hash slot，它們都 會(huì)被均勻分布到所有的master節(jié)點(diǎn)上；這里ClusterConnectionManager通過(guò)key名稱(chēng)計(jì)算出相應(yīng)的hash slot方式如下：

?先通過(guò)key計(jì)算出CRC16值，然后 CRC16值對(duì)16384進(jìn)?取模，進(jìn)?得到hash slot。

@Override
public int calcSlot(String key) {
    if (key == null) {
        return 0;
    }

    int start = key.indexOf('{');
    if (start != -1) {
        int end = key.indexOf('}');
        if (end != -1 && start + 1 < end) {
            key = key.substring(start + 1, end);
        }
    }

    int result = CRC16.crc16(key.getBytes()) % MAX_SLOT;
    log.debug("slot {} for {}", result, key);
    return result;
}

這?計(jì)算出key的hash slot之后，就可以通過(guò)hash slot 去看?看哪個(gè)master上有這個(gè)hash slot，如果某個(gè)master上有個(gè)這個(gè)hash slot，那么這個(gè) key當(dāng)然就會(huì)落到該master節(jié)點(diǎn)上，執(zhí)?加鎖指令也就應(yīng)該在該master上執(zhí)?。

下面進(jìn)入本文重點(diǎn)，可重入鎖的各種加鎖、釋放鎖。

2）Client第一次加鎖

在尋找應(yīng)該在哪臺(tái)Redis機(jī)器上加鎖時(shí)，在RedissonLock#tryLockInnerAsync()方法中我們看到了一堆LUA腳本：

LUA腳本參數(shù)解析：

• KEYS[1] 表示的是 getName() ，即鎖key的名稱(chēng)，比如案例中的 test_lock；
• ARGV[1] 表示的是 internalLockLeaseTime 默認(rèn)值是30s；
• ARGV[2] 表示的是 getLockName(threadId) ，唯一標(biāo)識(shí)當(dāng)前訪問(wèn)線程，使用鎖對(duì)象id+線程id（UUID:ThreadId）方式表示，用于區(qū)分不同服務(wù)器上的線程。
  • UUID用來(lái)唯?標(biāo)識(shí)?個(gè)客戶(hù)端，因?yàn)闀?huì)有多個(gè)客戶(hù)端的多個(gè)線程加鎖；
  • 結(jié)合起來(lái)的UUID:ThreadId 表示：具體哪個(gè)客戶(hù)端上的哪個(gè)線程過(guò)來(lái)加鎖，通 過(guò)這樣的組合?式唯?標(biāo)識(shí)?個(gè)線程。

LUA腳本邏輯：

• 如果鎖名稱(chēng)不存在；
  • 則向redis中添加一個(gè)key為test_lock的HASH結(jié)構(gòu)、添加一個(gè)field為線程id，值=1的鍵值對(duì){field:increment}，表示此線程的重入次數(shù)為1；
  • 設(shè)置test_lock的過(guò)期時(shí)間，防止當(dāng)前服務(wù)器出問(wèn)題后導(dǎo)致死鎖，然后return nil; end;返回nil，lua腳本執(zhí)行完畢；
• 如果鎖存在，檢測(cè)當(dāng)前線程是否持有鎖；
  • 如果是當(dāng)前線程持有鎖，hincrby將該線程重入的次數(shù)++；并重新設(shè)置鎖的過(guò)期時(shí)間；返回nil，lua腳本執(zhí)行完畢；
  • 如果不是當(dāng)前線程持有鎖，pttl返回鎖的過(guò)期時(shí)間，單位ms。

第一次加鎖時(shí)，key肯定不存在與master節(jié)點(diǎn)上；

會(huì)執(zhí)行下列LUA腳本對(duì)應(yīng)的Redis指令：

hset test_lock UUID:ThreadId 1 
pexpire test_lock 30000

此時(shí)，Redis中多一個(gè)Hash結(jié)構(gòu)的key（test_lock）：

test_lock : 
{
    UUID:ThreadId：1
}

這里的1使用來(lái)做鎖重入的。

pexpire指令為test_lock這個(gè)key設(shè)置過(guò)期時(shí)間為30s，即：30s后這個(gè)key會(huì)?動(dòng)過(guò)期被刪除，key對(duì)應(yīng)的鎖在那時(shí)也就被釋放了。

總體來(lái)看，加鎖的邏輯很簡(jiǎn)單：

在key對(duì)應(yīng)的hash數(shù)據(jù)結(jié)構(gòu)中記錄了? 下當(dāng)前是哪個(gè)客戶(hù)端的哪個(gè)線程過(guò)來(lái)加鎖了，然后設(shè)置了?下key的過(guò)期時(shí)間為30s。 3）加鎖成功之后的鎖續(xù)約

成功加鎖后，lua腳本返回nil，即null。

加鎖成功之后，tryLockInnerAsync()?法返回；再結(jié)合Java8的Stream，對(duì)加鎖結(jié)果進(jìn)一步處理；

因?yàn)榧渔i成功后返回的是nil，這是lua腳本的返回形式，體現(xiàn)到j(luò)ava代碼中的返回值為：null。
又由于RLock#lock()方法傳入的leaseTime是-1，所以進(jìn)入到scheduleExpirationRenewal(long)方法做鎖續(xù)約。

renewExpirationAsync()方法負(fù)責(zé)做具體的鎖續(xù)約：

protected CompletionStage<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]); " +
                    "return 1; " +
                    "end; " +
                    "return 0;",
            Collections.singletonList(getRawName()),
            internalLockLeaseTime, getLockName(threadId));
}

這里L(fēng)UA腳本的邏輯很簡(jiǎn)單：

• 判斷當(dāng)前key中，是否還被線程UUID:ThreadId持有鎖，持有則設(shè)置過(guò)期時(shí)間為30s（續(xù)命）。

鎖續(xù)約（看門(mén)狗機(jī)制）其實(shí)就是每次加鎖成功后，會(huì)?上開(kāi)啟?個(gè)后臺(tái)線程， 每隔10s檢查?下key是否存在，如果存在就為key續(xù)期30s。

• 這里的10s，取自配置的lockWatchdogTimeout參數(shù)，默認(rèn)為30 * 1000 ms；
• 所以?個(gè)key往往當(dāng)過(guò)期時(shí)間慢慢消逝到20s左右時(shí)就?會(huì)被定時(shí)任務(wù)重置為了30s，這樣就能保證：只要這個(gè)定時(shí)任務(wù)還在、這個(gè)key還在，就?直維持加鎖。

如果當(dāng)前持有鎖的線程被中斷了，會(huì)停止鎖續(xù)約，即殺死看門(mén)狗；

protected void cancelExpirationRenewal(Long threadId) {
    ExpirationEntry task = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    if (task == null) {
        return;
    }
    
    if (threadId != null) {
        task.removeThreadId(threadId);
    }

    if (threadId == null || task.hasNoThreads()) {
        Timeout timeout = task.getTimeout();
        if (timeout != null) {
            timeout.cancel();
        }
        EXPIRATION_RENEWAL_MAP.remove(getEntryName());
    }
}

所謂的停止鎖續(xù)約，實(shí)際就是將當(dāng)前線程的threadId從看門(mén)狗緩存中移除，后續(xù)在執(zhí)行鎖續(xù)約時(shí)，如果發(fā)現(xiàn)看門(mén)狗緩存中已經(jīng)沒(méi)有了當(dāng)前線程threadId，則直接退出鎖續(xù)約 并且 不再延時(shí)10s開(kāi)啟一個(gè)定時(shí)任務(wù)。

如果加鎖時(shí)指定了leaseTime > 0，則不會(huì)開(kāi)門(mén)狗機(jī)制，表示強(qiáng)制鎖leaseTime 毫秒后過(guò)期。一共有三種加鎖方式可以做到，如下：

• RLock#lock(long leaseTime, TimeUnit unit)
• RLock#tryLock(long waitTime, long leaseTime, TimeUnit unit)
• RLock#lockInterruptibly(long leaseTime, TimeUnit unit)

4）重入加鎖（相同線程多次加鎖）

再次回到加鎖的LUA腳本：

同一個(gè)線程對(duì)分布式鎖多次加鎖時(shí)，會(huì)走以下邏輯：

• 判斷當(dāng)前key是否被當(dāng)前線程持有，如果是則增加鎖重入的次數(shù)，并重新設(shè)置鎖的過(guò)期時(shí)間為30s；

對(duì)應(yīng)的Redis命令為：

hexists test_lock UUID:ThreadId
hincrby test_lock UUID:ThreadId 1
pexpire test_lock 30000

此時(shí)Redis中test_key對(duì)應(yīng)的數(shù)據(jù)結(jié)構(gòu)從

test_lock : 
{
    UUID:ThreadId：1
}

變成：

test_lock : 
{
    UUID:ThreadId：2
}

并將key的過(guò)期時(shí)間重新設(shè)置為30s。

鎖重入成功之后，后臺(tái)也會(huì)開(kāi)啟?個(gè)watchdog后臺(tái)線程做鎖續(xù)約，每隔10s檢查?下key，如果key存在就將key的過(guò)期時(shí)間重新設(shè)置為30s。

Redisson可重?加鎖的語(yǔ)義，實(shí)際是通過(guò)Hash結(jié)構(gòu)的key中某個(gè)線程（UUID:ThreadId）對(duì)應(yīng)的加鎖次數(shù)來(lái)表示的。

5）鎖競(jìng)爭(zhēng)（其他線程加鎖失敗）

再再次回到加鎖的LUA腳本：

如果分布式鎖已經(jīng)被其他線程持有，LUA腳本會(huì)執(zhí)行以下邏輯：

返回當(dāng)前key的剩余存活時(shí)間，因?yàn)椴皇欠祷豱il，也就代表著加鎖失?。?/p>

對(duì)應(yīng)的Redis的命令為：

pttl test_lock

針對(duì)加鎖方式的不同，加鎖失敗的邏輯也不同；可以分兩大類(lèi)：指定了加鎖失敗的等待時(shí)間waitTime和未指定waitTime。

• 未執(zhí)行加鎖失敗的等待時(shí)間waitTime：獲取分布式鎖失敗會(huì)一直重試，直到獲取鎖成功。比如下列加鎖方法：
  • Rlock#lock()：一直嘗試獲取分布式鎖，直到獲取鎖成功。
  • RLock#lockInterruptibly(long leaseTime, TimeUnit unit)
  • RLock#lock(long leaseTime, TimeUnit unit)
• 指定了加鎖失敗的等待時(shí)間waitTime：獲取分布式鎖會(huì)超時(shí)，超時(shí)之后返回加鎖失?。?ul>
• Rlock#tryLock(long waitTime, TimeUnit unit)：指定獲取鎖失敗的等待時(shí)間。在等待時(shí)間范圍之內(nèi)進(jìn)行重試，超時(shí)則返回加鎖失敗。
• Rlock#tryLock(long waitTime, long leaseTime, TimeUnit unit)：同樣是指定獲取鎖失敗的等待時(shí)間，并且強(qiáng)制指定鎖過(guò)期的時(shí)間（不開(kāi)啟看門(mén)狗）。在等待時(shí)間范圍之內(nèi)進(jìn)行重試，超時(shí)則返回加鎖失敗。

可以簡(jiǎn)單的概述為RLock接口下的tryLock()方法獲取鎖會(huì)失敗，lock()方法獲取鎖一定會(huì)成功。

1> 一直重試直到加鎖成功

以Rlock#lock()方法為例：

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    long threadId = Thread.currentThread().getId();
    Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        return;
    }

    CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
    pubSub.timeout(future);
    RedissonLockEntry entry;
    if (interruptibly) {
        entry = commandExecutor.getInterrupted(future);
    } else {
        entry = commandExecutor.get(future);
    }

    try {
        while (true) {
            // lock() 或 lockInterruptibly()為入口走到這里時(shí)。leaseTime為-1，表示會(huì)開(kāi)始開(kāi)門(mén)狗；如果leaseTime大于0，則不會(huì)開(kāi)啟開(kāi)門(mén)狗；
            ttl = tryAcquire(-1, leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                break;
            }

            // waiting for message
            if (ttl >= 0) {
                try {
                    // 因?yàn)镾emaphore的可用資源為0，所以這里就等價(jià)于Thread.sleep(ttl)；
                    entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    if (interruptibly) {
                        throw e;
                    }
                    entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                }
            } else {
                if (interruptibly) {
                    entry.getLatch().acquire();
                } else {
                    entry.getLatch().acquireUninterruptibly();
                }
            }
        }
    } finally {
        unsubscribe(entry, threadId);
    }
}

首先訂閱解鎖channel（命名格式：redisson_lock__channel:{keyName}），其他線程解鎖后，會(huì)發(fā)布解鎖的消息；這里收到消息會(huì)立即嘗試獲取鎖；訂閱解鎖channel的超時(shí)時(shí)間默認(rèn)為7.5s。也就說(shuō)獲取鎖失敗7.5s之內(nèi)，如果其他線程釋放鎖，當(dāng)前線程可以立即嘗試獲取到鎖。

獲取鎖失敗之后會(huì)進(jìn)??個(gè)while死循環(huán)中：

每休息鎖的存活時(shí)間ttl之后，就嘗試去獲取鎖，直到成功獲取到鎖才會(huì)跳出while死循環(huán)。

2> 等待鎖超時(shí)返回加鎖失敗

以Rlock#tryLock(long waitTime, TimeUnit unit)為例：

@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();
    Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        return true;
    }

    // 獲取鎖剩余的等待時(shí)長(zhǎng)
    time -= System.currentTimeMillis() - current;
    if (time <= 0) {
        // 獲取鎖超時(shí)，返回獲取分布式鎖失敗
        acquireFailed(waitTime, unit, threadId);
        return false;
    }
    
    current = System.currentTimeMillis();
    CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
    try {
        // 訂閱解鎖channel的超時(shí)時(shí)長(zhǎng)為 獲取鎖剩余的等待時(shí)長(zhǎng)
        subscribeFuture.get(time, TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
        if (!subscribeFuture.completeExceptionally(new RedisTimeoutException(
                "Unable to acquire subscription lock after " + time + "ms. " +
                        "Try to increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters."))) {
            subscribeFuture.whenComplete((res, ex) -> {
                if (ex == null) {
                    unsubscribe(res, threadId);
                }
            });
        }
        acquireFailed(waitTime, unit, threadId);
        return false;
    } catch (ExecutionException e) {
        acquireFailed(waitTime, unit, threadId);
        return false;
    }

    try {
        // 收到解鎖channel的消息之后，走到這里，再次判斷獲取鎖等待時(shí)長(zhǎng)是否超時(shí)
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
    
        // while循環(huán)中嘗試去獲取鎖
        while (true) {
            long currentTime = System.currentTimeMillis();
            ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                return true;
            }

            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }

            // waiting for message
            currentTime = System.currentTimeMillis();
            if (ttl >= 0 && ttl < time) {
                // 如果獲取鎖失敗后，鎖存活時(shí)長(zhǎng) 小于 剩余鎖等待時(shí)長(zhǎng)，則線程睡眠 鎖存活時(shí)長(zhǎng)
                commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
            } else {
                // 如果獲取鎖失敗后，鎖存活時(shí)間 大于等于 剩余鎖等待時(shí)長(zhǎng)，則線程睡眠 鎖等待時(shí)長(zhǎng)
                commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
            }

            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }
        }
    } finally {
        unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);
    }
}

加鎖存在超時(shí)時(shí)間 相比于 一直重試直到加鎖成功，只是多一個(gè)時(shí)間限制，具體差異體現(xiàn)在：訂閱解鎖channel的超時(shí)時(shí)長(zhǎng)、獲取鎖失敗后線程的睡眠時(shí)長(zhǎng)、重試獲取鎖次數(shù)的限制；

獲取分布式鎖失敗之后，立即判斷當(dāng)前獲取鎖是否超時(shí)，如果超時(shí)，則返回加鎖失??；

否者，訂閱解鎖channel（命名格式：redisson_lock__channel:{keyName}），其他線程解鎖后，會(huì)發(fā)布解鎖的消息；

訂閱解鎖channel的超時(shí)時(shí)間為 獲取鎖剩余的等待時(shí)長(zhǎng)。 在這個(gè)時(shí)間范圍之內(nèi)，如果其他線程釋放鎖，當(dāng)前線程收到解鎖channel的消息之后再次判斷獲取鎖是否超時(shí)，如果不超時(shí)，嘗試獲取鎖。

獲取鎖之后會(huì)進(jìn)??個(gè)while死循環(huán)中： 如果獲取鎖超時(shí)，則返回加鎖失??；

否者讓線程睡眠： 如果鎖存活時(shí)長(zhǎng)ttl 小于 剩余鎖等待時(shí)長(zhǎng)，則線程睡眠 鎖存活時(shí)長(zhǎng)；

如果鎖存活時(shí)間ttl 大于等于 剩余鎖等待時(shí)長(zhǎng)，則線程睡眠 鎖等待時(shí)長(zhǎng)；

線程睡眠完之后，判斷獲取鎖是否超時(shí)，不超時(shí)則嘗試去獲取鎖。

3、釋放鎖流程

1）Client主動(dòng)嘗試釋放鎖

進(jìn)入到Rlock#unlock()方法；

和加鎖的方式?樣，釋放鎖也是通過(guò)lua腳本來(lái)完成的；

LUA腳本參數(shù)解析：

• KEYS[1] 表示的是 getName() ，代表的是鎖名 test_lock；
• KEYS[2] 表示getChanelName() 表示的是發(fā)布訂閱過(guò)程中使用的Chanel；
• ARGV[1] 表示的是LockPubSub.unLockMessage，解鎖消息，實(shí)際代表的是數(shù)字 0，代表解鎖消息；
• ARGV[2] 表示的是internalLockLeaseTime 默認(rèn)的有效時(shí)間 30s；
• ARGV[3] 表示的是 getLockName(thread.currentThread().getId()) 代表的是 UUID:ThreadId 用鎖對(duì)象id+線程id， 表示當(dāng)前訪問(wèn)線程，用于區(qū)分不同服務(wù)器上的線程。

LUA腳本邏輯：

• 如果鎖名稱(chēng)不存在；
• 可能是因?yàn)殒i過(guò)期導(dǎo)致鎖不存在，也可能是并發(fā)解鎖。
• 則發(fā)布鎖解除的消息，返回1，lua腳本執(zhí)行完畢；
• 如果鎖存在，檢測(cè)當(dāng)前線程是否持有鎖；
• 如果是當(dāng)前線程持有鎖，定義變量counter，接收?qǐng)?zhí)行incrby將該線程重入的次數(shù)–的結(jié)果；
• 如果重入次數(shù)大于0，表示該線程還有其他任務(wù)需要執(zhí)行；重新設(shè)置鎖的過(guò)期時(shí)間；返回0，lua腳本執(zhí)行完畢；
• 否則表示該線程執(zhí)行結(jié)束，del刪除該鎖；并且publish發(fā)布該鎖解除的消息；返回1，lua腳本執(zhí)行完畢；
• 如果不是當(dāng)前線程持有鎖 或 其他情況，都返回nil，lua腳本執(zhí)行完畢。

腳本執(zhí)行結(jié)束之后，如果返回值不是0或1，即當(dāng)前線程去釋放其他線程的加鎖時(shí)，拋出異常。

通過(guò)LUA腳本釋放鎖成功之后，會(huì)將看門(mén)狗殺死；

2）Client主動(dòng)強(qiáng)制釋放鎖

forceUnlockAsync()方法被調(diào)用的地方很多，大多都是在清理資源時(shí)刪除鎖。

@Override
public RFuture<Boolean> forceUnlockAsync() {
    cancelExpirationRenewal(null);
    return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            "if (redis.call('del', KEYS[1]) == 1) then "
                    + "redis.call('publish', KEYS[2], ARGV[1]); "
                    + "return 1 "
                    + "else "
                    + "return 0 "
                    + "end",
            Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE);
}

LUA腳本邏輯：

邏輯比較簡(jiǎn)單粗暴：刪除鎖成功則并發(fā)布鎖被刪除的消息，返回1結(jié)束，否則返回0結(jié)束。

3）Client宕機(jī)，鎖超時(shí)釋放

如果Redisson客戶(hù)端剛加鎖成功，并且未指定releaseTime，后臺(tái)會(huì)啟動(dòng)一個(gè)定時(shí)任務(wù)watchdog每隔10s檢查key：key如果存在就為它?動(dòng)續(xù)命到30s；在watchdog定時(shí)任務(wù)存在的情況下，如果不是主動(dòng)釋放鎖，那么key將會(huì)?直的被watchdog這個(gè)定時(shí)任務(wù)維持加鎖。

但是如果客戶(hù)端宕機(jī)了，定時(shí)任務(wù)watchdog也就沒(méi)了，也就沒(méi)有鎖續(xù)約機(jī)制了，那么過(guò)完30s之后，key會(huì)?動(dòng)被刪除、key對(duì)應(yīng)的鎖也自動(dòng)被釋放了。

4）不啟動(dòng)鎖續(xù)約的超時(shí)釋放鎖

如果在加鎖時(shí)指定了leaseTime，加鎖成功之后，后臺(tái)并不會(huì)啟動(dòng)一個(gè)定時(shí)任務(wù)watchdog做鎖續(xù)約；key存活leaseTime 毫秒之后便會(huì)自動(dòng)被刪除、key對(duì)應(yīng)的鎖也就自動(dòng)被釋放了；無(wú)論當(dāng)前線程的業(yè)務(wù)邏輯是否執(zhí)行完畢。

比如使用如下方式加鎖：

• RLock#lock(long leaseTime, TimeUnit unit)
• RLock#tryLock(long waitTime, long leaseTime, TimeUnit unit)
• RLock#lockInterruptibly(long leaseTime, TimeUnit unit)

到此這篇關(guān)于Redisson如何實(shí)現(xiàn)分布式鎖、鎖續(xù)約的文章就介紹到這了,更多相關(guān)redisson分布式鎖、鎖續(xù)約內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論