Redisson公平鎖的源碼解讀分享

更新時間：2022年11月08日 16:32:32 作者：王谷雨

Redisson是一個在Redis的基礎(chǔ)上實現(xiàn)的Java駐內(nèi)存數(shù)據(jù)網(wǎng)格（In-Memory?Data?Grid）。這篇文章主要通過源碼和大家聊聊Redisson公平鎖，需要的可以了解一下

前言

我在上一篇文章聊了Redisson的分布式鎖，這次繼續(xù)來聊聊Redisson的公平鎖。下面是官方原話：

它保證了當多個Redisson客戶端線程同時請求加鎖時，優(yōu)先分配給先發(fā)出請求的線程。所有請求線程會在一個隊列中排隊，當某個線程出現(xiàn)宕機時，Redisson會等待5秒后繼續(xù)下一個線程，也就是說如果前面有5個線程都處于等待狀態(tài)，那么后面的線程會等待至少25秒。

源碼版本：3.17.7

這是我 fork 的分支，添加了自己理解的中文注釋：https://github.com/xiaoguyu/redisson

公平鎖

先上官方例子：

RLock fairLock = redisson.getFairLock("anyLock");
// 嘗試加鎖，最多等待100秒，上鎖以后10秒自動解鎖
boolean res = fairLock.tryLock(100, 10, TimeUnit.SECONDS);
...
fairLock.unlock();

因為在Redisson中，公平鎖和普通可重入鎖的邏輯大體上一樣，我在上一篇文章都介紹了，這里就不再贅述。下面開始介紹合理邏輯。

加鎖

加鎖的 lua 腳本在 RedissonFairLock#tryLockInnerAsync方法中

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        long wait = threadWaitTime;
        if (waitTime > 0) {
            wait = unit.toMillis(waitTime);
        }
 
        long currentTime = System.currentTimeMillis();
        if (command == RedisCommands.EVAL_NULL_BOOLEAN) {
            ......
        }
 
        if (command == RedisCommands.EVAL_LONG) {
            return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
                    // remove stale threads
                    "while true do " +  // list為空，證明沒有人排隊，退出循環(huán)
                        "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" +
                        "if firstThreadId2 == false then " +
                            "break;" +
                        "end;" +
                        // 能到這里，證明有人排隊，拿出在排隊的第一個人的超時時間，如果超時了，則移除相應數(shù)據(jù)
                        "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" +
                        "if timeout <= tonumber(ARGV[4]) then " +
                            // remove the item from the queue and timeout set
                            // NOTE we do not alter any other timeout
                            "redis.call('zrem', KEYS[3], firstThreadId2);" +
                            "redis.call('lpop', KEYS[2]);" +
                        "else " +
                            "break;" +
                        "end;" +
                    "end;" +
 
                    // check if the lock can be acquired now
                    // 檢查是否可以獲取鎖。如果hash和list都不存在，或者線程隊列的第一個是當前線程，則可以獲取鎖
                    "if (redis.call('exists', KEYS[1]) == 0) " +
                        "and ((redis.call('exists', KEYS[2]) == 0) " +
                            "or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then " +
 
                        // remove this thread from the queue and timeout set
                        // 都獲取鎖了，當然要從線程隊列和時間隊列中移除
                        "redis.call('lpop', KEYS[2]);" +
                        "redis.call('zrem', KEYS[3], ARGV[2]);" +
 
                        // decrease timeouts for all waiting in the queue
                        // 刷新時間集合中的時間
                        "local keys = redis.call('zrange', KEYS[3], 0, -1);" +
                        "for i = 1, #keys, 1 do " +
                            "redis.call('zincrby', KEYS[3], -tonumber(ARGV[3]), keys[i]);" +
                        "end;" +
 
                        // acquire the lock and set the TTL for the lease
                        // 和公平鎖的設(shè)置一樣，值加1并且設(shè)置過期時間
                        "redis.call('hset', KEYS[1], ARGV[2], 1);" +
                        "redis.call('pexpire', KEYS[1], ARGV[1]);" +
                        "return nil;" +
                    "end;" +
 
                    // check if the lock is already held, and this is a re-entry
                    // 能到這里，證明前面拿不到鎖，但是也要做可重入鎖的處理
                    "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;" +
 
                    // the lock cannot be acquired
                    // check if the thread is already in the queue
                    // 時間集合中有值，證明線程已經(jīng)在隊列中，不需要往后執(zhí)行邏輯了
                    "local timeout = redis.call('zscore', KEYS[3], ARGV[2]);" +
                    "if timeout ~= false then " +
                        // the real timeout is the timeout of the prior thread
                        // in the queue, but this is approximately correct, and
                        // avoids having to traverse the queue
                        // 因為下面的timeout = ttl + tonumber(ARGV[3]) + tonumber(ARGV[4])
                        // 所以這里的ttl = timeout - tonumber(ARGV[3]) - tonumber(ARGV[4])
                        "return timeout - tonumber(ARGV[3]) - tonumber(ARGV[4]);" +
                    "end;" +
 
                    // add the thread to the queue at the end, and set its timeout in the timeout set to the timeout of
                    // the prior thread in the queue (or the timeout of the lock if the queue is empty) plus the
                    // threadWaitTime
                    "local lastThreadId = redis.call('lindex', KEYS[2], -1);" +
                    "local ttl;" +
                    // 如果最后一個線程不是當前線程，則從時間集合取出（舉例：線程1/2/3按順序獲取鎖，此時pttl得到的是線程1的鎖過期時間，zscore拿到的是線程2的鎖的過期時間，此時線程3應該以線程2的為準）
                    "if lastThreadId ~= false and lastThreadId ~= ARGV[2] then " +
                        "ttl = tonumber(redis.call('zscore', KEYS[3], lastThreadId)) - tonumber(ARGV[4]);" +
                    "else " +
                        // 否則直接獲取鎖的存活時間
                        "ttl = redis.call('pttl', KEYS[1]);" +
                    "end;" +
                    // 過期時間 = 鎖存活時間 + 等待時間 + 當前時間戳
                    "local timeout = ttl + tonumber(ARGV[3]) + tonumber(ARGV[4]);" +
                    // 如果添加到時間集合成功，則同時添加線程集合
                    "if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then " +
                        "redis.call('rpush', KEYS[2], ARGV[2]);" +
                    "end;" +
                    "return ttl;",
                    Arrays.asList(getRawName(), threadsQueueName, timeoutSetName),
                    unit.toMillis(leaseTime), getLockName(threadId), wait, currentTime);
        }
 
        throw new IllegalArgumentException();
    }

公平鎖總共用了Redis的三種數(shù)據(jù)類型，對應著 lua 腳本里面的keys1、2、3的參數(shù)：

KEYS[1]

鎖的名字，使用 Hash 數(shù)據(jù)類型，是可重入鎖的基礎(chǔ)，結(jié)構(gòu)為 {”threadId1”: 1, “thread2”: 1}，key為線程id，value是鎖的次數(shù)

KEYS[2]

線程隊列的名字，使用 List 數(shù)據(jù)類型，結(jié)構(gòu)為 [ “threadId1”, “threadId2” ]，按順序存放需要獲取鎖的線程的id

KEYS[3]

時間隊列的名字，使用 sorted set 數(shù)據(jù)類型，結(jié)構(gòu)為 {”threadId2”:123, “threadId1”:190}，key為線程id，value為獲取鎖的超時時間戳

我下面會用鎖、線程隊列、時間隊列來表示這3個數(shù)據(jù)結(jié)構(gòu)，需要注意下我的表述。

同樣的，介紹下參數(shù)：

ARGV[1]：leaseTime 鎖的持有時間
ARGV[2]：線程id（描述不太準確，暫時按這樣理解）
ARGV[3]：waitTime 嘗試獲取鎖的最大等待時間
ARGV[4]：currentTime 當前時間戳

接下來，我們一段一段分析 lua 腳本，首先看最開始的 while 循環(huán)

"while true do " +  // list為空，證明沒有人排隊，退出循環(huán)
    "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" +
    "if firstThreadId2 == false then " +
        "break;" +
    "end;" +
    // 能到這里，證明有人排隊，拿出在排隊的第一個人的超時時間，如果超時了，則移除相應數(shù)據(jù)
    "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" +
    "if timeout <= tonumber(ARGV[4]) then " +
        // 從時間隊列和線程隊列中移除
        "redis.call('zrem', KEYS[3], firstThreadId2);" +
        "redis.call('lpop', KEYS[2]);" +
    "else " +
        "break;" +
    "end;" +
"end;" +

具體的邏輯我在注釋中寫的很清楚了，看的時候記住 KEYS[2]、KEYS[3] 對應著線程隊列和時間隊列接口。主要注意的是，線程隊列只有當一個線程持有鎖，另一個線程獲取不到鎖時，才會有值（前面有人才排隊，沒人排什么隊）。接著看第二段

// 檢查是否可以獲取鎖。當鎖不存在，并且線程隊列不存在或者線程隊列第一位是當前線程，則可以獲取鎖
"if (redis.call('exists', KEYS[1]) == 0) " +
    "and ((redis.call('exists', KEYS[2]) == 0) or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then " +
 
    // remove this thread from the queue and timeout set
    // 都獲取鎖了，當然要從線程隊列和時間隊列中移除
    "redis.call('lpop', KEYS[2]);" +
    "redis.call('zrem', KEYS[3], ARGV[2]);" +
 
    // decrease timeouts for all waiting in the queue
    // 刷新時間隊列中的時間
    "local keys = redis.call('zrange', KEYS[3], 0, -1);" +
    "for i = 1, #keys, 1 do " +
        "redis.call('zincrby', KEYS[3], -tonumber(ARGV[3]), keys[i]);" +
    "end;" +
 
    // acquire the lock and set the TTL for the lease
    // 和公平鎖的設(shè)置一樣，值加1并且設(shè)置過期時間
    "redis.call('hset', KEYS[1], ARGV[2], 1);" +
    "redis.call('pexpire', KEYS[1], ARGV[1]);" +
    "return nil;" +
"end;" +

翻譯翻譯就是，鎖不存在（別人沒有持有鎖）并且線程隊列不存在或者線程隊列第一位是當前線程（不用排隊或者自己排第一）才能獲得鎖。因為時間隊列中存放的是各個線程等待鎖的超時時間戳，所以每次都需要刷新下。繼續(xù)下一段邏輯

// 能到這里，證明前面拿不到鎖，但是也要做可重入鎖的處理
"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;" +

這是可重入鎖的處理，繼續(xù)下一段

// 時間隊列中有值，證明線程已經(jīng)在隊列中，不需要往后執(zhí)行邏輯了
"local timeout = redis.call('zscore', KEYS[3], ARGV[2]);" +
"if timeout ~= false then " +
    // the real timeout is the timeout of the prior thread
    // in the queue, but this is approximately correct, and
    // avoids having to traverse the queue
    // 因為下面的timeout = ttl + tonumber(ARGV[3]) + tonumber(ARGV[4])
    // 所以這里的ttl = timeout - tonumber(ARGV[3]) - tonumber(ARGV[4])
    "return timeout - tonumber(ARGV[3]) - tonumber(ARGV[4]);" +
"end;" +

舉例子：線程1持有鎖，線程2嘗試第一次獲取鎖（不進入這段if），線程2第二次獲取鎖（進入了這段if）。繼續(xù)下一段

"local lastThreadId = redis.call('lindex', KEYS[2], -1);" +
"local ttl;" +
// 如果最后一個線程不是當前線程，則從時間集合取出（舉例：線程1/2/3按順序獲取鎖，此時pttl得到的是線程1的鎖過期時間，zscore拿到的是線程2的鎖的過期時間，此時線程3應該以線程2的為準）
"if lastThreadId ~= false and lastThreadId ~= ARGV[2] then " +
    "ttl = tonumber(redis.call('zscore', KEYS[3], lastThreadId)) - tonumber(ARGV[4]);" +
"else " +
    // 否則直接獲取鎖的存活時間
    "ttl = redis.call('pttl', KEYS[1]);" +
"end;" +
// 過期時間 = 鎖存活時間 + 等待時間 + 當前時間戳
"local timeout = ttl + tonumber(ARGV[3]) + tonumber(ARGV[4]);" +
// 如果添加到時間集合成功，則同時添加線程集合
"if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then " +
    "redis.call('rpush', KEYS[2], ARGV[2]);" +
"end;" +
"return ttl;",

ttl 這段的獲取邏輯，翻譯翻譯就是，如果前面有人排隊，就以前面的超時時間為準，如果沒人排隊，就拿鎖的超時時間。獲取到 ttl ，就對添加到線程集合和時間集合。

以上就是公平鎖的加鎖 lua 腳本的全部邏輯。講的有點亂，但是只要能搞清楚keys1、2、3對應著哪種數(shù)據(jù)類型，理解整個邏輯應該問題不大。

解鎖

解鎖的核心 lua 腳本是下面這段RedissonFairLock#unlockInnerAsync

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            // remove stale threads
            "while true do "  // 線程隊列為空，證明沒有人排隊，退出循環(huán)
            + "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);"
            + "if firstThreadId2 == false then "
                + "break;"
            + "end; "
            // 能到這里，證明有人排隊，拿出在排隊的第一個人的超時時間，如果超時了，則移除相應數(shù)據(jù)
            + "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));"
            + "if timeout <= tonumber(ARGV[4]) then "
                + "redis.call('zrem', KEYS[3], firstThreadId2); "
                + "redis.call('lpop', KEYS[2]); "
            + "else "
                + "break;"
            + "end; "
          + "end;"
            // 如果鎖不存在，則通過訂閱發(fā)布機制通知下一個等待中的線程
          + "if (redis.call('exists', KEYS[1]) == 0) then " + 
                "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + 
                "if nextThreadId ~= false then " +
                    "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " +
                "end; " +
                "return 1; " +
            "end;" +
            // 如果當前線程已經(jīng)不存在鎖里面，直接返回null
            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                "return nil;" +
            "end; " +
            // 可重入鎖處理邏輯，對當前線程的鎖次數(shù)減1
            "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
            "if (counter > 0) then " +
                // 鎖次數(shù)仍然大于0，則刷新鎖的存活時間
                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                "return 0; " +
            "end; " +
 
            // 刪除鎖
            "redis.call('del', KEYS[1]); " +
            // 訂閱發(fā)布機制通知下一個等待中的線程
            "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + 
            "if nextThreadId ~= false then " +
                "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " +
            "end; " +
            "return 1; ",
            Arrays.asList(getRawName(), threadsQueueName, timeoutSetName, getChannelName()),
            LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId), System.currentTimeMillis());
}

算了，不想寫了，看注釋吧。