Redisson公平鎖的源碼解讀分享
前言
我在上一篇文章聊瞭Redisson的分佈式鎖,這次繼續來聊聊Redisson的公平鎖。下面是官方原話:
它保證瞭當多個Redisson客戶端線程同時請求加鎖時,優先分配給先發出請求的線程。所有請求線程會在一個隊列中排隊,當某個線程出現宕機時,Redisson會等待5秒後繼續下一個線程,也就是說如果前面有5個線程都處於等待狀態,那麼後面的線程會等待至少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為空,證明沒有人排隊,退出循環 "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" + "if firstThreadId2 == false then " + "break;" + "end;" + // 能到這裡,證明有人排隊,拿出在排隊的第一個人的超時時間,如果超時瞭,則移除相應數據 "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 // 和公平鎖的設置一樣,值加1並且設置過期時間 "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 // 時間集合中有值,證明線程已經在隊列中,不需要往後執行邏輯瞭 "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的三種數據類型,對應著 lua 腳本裡面的keys1、2、3的參數:
KEYS[1]
鎖的名字,使用 Hash 數據類型,是可重入鎖的基礎,結構為 {”threadId1”: 1, “thread2”: 1},key為線程id,value是鎖的次數
KEYS[2]
線程隊列的名字,使用 List 數據類型,結構為 [ “threadId1”, “threadId2” ],按順序存放需要獲取鎖的線程的id
KEYS[3]
時間隊列的名字,使用 sorted set 數據類型,結構為 {”threadId2”:123, “threadId1”:190},key為線程id,value為獲取鎖的超時時間戳
我下面會用 鎖、線程隊列、時間隊列 來表示這3個數據結構,需要註意下我的表述。
同樣的,介紹下參數:
- ARGV[1]:leaseTime 鎖的持有時間
- ARGV[2]:線程id(描述不太準確,暫時按這樣理解)
- ARGV[3]:waitTime 嘗試獲取鎖的最大等待時間
- ARGV[4]:currentTime 當前時間戳
接下來,我們一段一段分析 lua 腳本,首先看最開始的 while 循環
"while true do " + // list為空,證明沒有人排隊,退出循環 "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" + "if firstThreadId2 == false then " + "break;" + "end;" + // 能到這裡,證明有人排隊,拿出在排隊的第一個人的超時時間,如果超時瞭,則移除相應數據 "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 // 和公平鎖的設置一樣,值加1並且設置過期時間 "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;" +
這是可重入鎖的處理,繼續下一段
// 時間隊列中有值,證明線程已經在隊列中,不需要往後執行邏輯瞭 "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)。繼續下一段
"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對應著哪種數據類型,理解整個邏輯應該問題不大。
解鎖
解鎖的核心 lua 腳本是下面這段RedissonFairLock#unlockInnerAsync
protected RFuture<Boolean> unlockInnerAsync(long threadId) { return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, // remove stale threads "while true do " // 線程隊列為空,證明沒有人排隊,退出循環 + "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" + "if firstThreadId2 == false then " + "break;" + "end; " // 能到這裡,證明有人排隊,拿出在排隊的第一個人的超時時間,如果超時瞭,則移除相應數據 + "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;" // 如果鎖不存在,則通過訂閱發佈機制通知下一個等待中的線程 + "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;" + // 如果當前線程已經不存在鎖裡面,直接返回null "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + "return nil;" + "end; " + // 可重入鎖處理邏輯,對當前線程的鎖次數減1 "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " + "if (counter > 0) then " + // 鎖次數仍然大於0,則刷新鎖的存活時間 "redis.call('pexpire', KEYS[1], ARGV[2]); " + "return 0; " + "end; " + // 刪除鎖 "redis.call('del', KEYS[1]); " + // 訂閱發佈機制通知下一個等待中的線程 "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()); }
算瞭,不想寫瞭,看註釋吧。
總結
本文介紹瞭Redisson的公平鎖,邏輯大體上和普通可重入鎖一致,核心在於 lua 腳本,運用瞭Redis的3種數據類型。
到此這篇關於Redisson公平鎖的源碼解讀分享的文章就介紹到這瞭,更多相關Redisson公平鎖內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- Redisson可重入鎖解鎖邏輯詳細講解
- Redis分佈式鎖Redlock的實現
- redis分佈式鎖的8大坑總結梳理
- redis lua腳本實戰秒殺和減庫存的實現
- Redisson分佈式信號量RSemaphore的使用超詳細講解