建站
咨詢
售后
建站咨詢
新聞中心
/**
* Creates an instance of {@code ReentrantLock} with the
* given fairness policy.
*
* @param fair {@code true} if this lock should use a fair ordering policy
*/
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
對(duì)于需要公平的場(chǎng)景，和我們真實(shí)生活一樣，這里的 FairSync 會(huì)通過(guò)一個(gè) CLH 隊(duì)列將請(qǐng)求線程排隊(duì)。
在單實(shí)例應(yīng)用中，本機(jī)的 CLH 排隊(duì)就足夠了，我們現(xiàn)在切換到分布式的場(chǎng)景。
在分布式場(chǎng)景下，為了實(shí)現(xiàn)鎖的功能，就出現(xiàn)了各種分布式鎖。相比單實(shí)例場(chǎng)景下的鎖只能鎖定自己的實(shí)例，分布式的鎖由于統(tǒng)一的外部中間件的介入，將鎖的信息提取到獨(dú)立的外部，所以可以將多個(gè)應(yīng)用實(shí)例做到互斥。
那分布式的場(chǎng)景下，怎么樣能保證公平呢?
和我們從單實(shí)例到分布式加鎖的思路一樣，要公平，就排隊(duì)，在統(tǒng)一的第三方處進(jìn)行排隊(duì)。
來(lái)第三方這里排隊(duì)，也有一些需要注意的點(diǎn)。比如你在判斷當(dāng)前隊(duì)列里有沒(méi)有等待，如果沒(méi)有就取鎖成功，執(zhí)行，有等待就入隊(duì)，而判斷的這個(gè)邏輯，仍然可能是并發(fā)操作，也需要做到加鎖處理。就好像你看了一眼某個(gè)飯店沒(méi)什么人，開心的去買了杯奶茶，回來(lái)一看滿了。
對(duì)于分布式鎖，基于 Redis 的 Redission 用得不少。如果換成Redis 分布式公平鎖，那基本就只有 Redission 了。
下面我通過(guò)兩段代碼，以及部分文字，描述下 Redission 的公平鎖的實(shí)現(xiàn)。
簡(jiǎn)要概括下：
Redission 的公平鎖，是通過(guò)「 Redis + Lua 腳本」來(lái)實(shí)現(xiàn)的。在拿到一個(gè) Redission 的 Redis 連接之后，通過(guò) 「eval()」可以執(zhí)行一段 Lua script，同時(shí)傳入一些 key 和 args。因?yàn)椴还苡卸嗌?Lua 的邏輯，都是在同一個(gè)連接內(nèi)，所以不會(huì)存在買完奶茶發(fā)現(xiàn)人滿了的情況。這里應(yīng)用到了 Redis 的 pub/sub 功能，等待的線程，會(huì)在輪到自己時(shí)收到 Redis 的提醒，前提是需要訂閱了相應(yīng)的通知。
來(lái)看加鎖的 Lua 邏輯，代碼寫的比較清楚，我也加了些對(duì)應(yīng)的Redis操作以及參數(shù)的注釋。通過(guò) list 來(lái)存儲(chǔ)排隊(duì)信息，同時(shí)每個(gè)等待線程都有一個(gè)超時(shí)時(shí)間，超時(shí)退出隊(duì)列。 所以eval 執(zhí)行這個(gè)的時(shí)候，返回的是一個(gè) ttl Long 類型。表示過(guò)期時(shí)間。
[[
用于 lock 操作。
KEYS[1] = lockName
KEYS[2] = waitQueueName
KEYS[3] = timeoutName
ARGV[1] = waitTime
ARGV[2] = lockName
ARGV[3] = leaseTime
ARGV[4] = currentTime
--]]
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) and
((redis.call("exists", KEYS[2]) == 0) or (redis.call("lindex", KEYS[2], 0) == ARGV[2]))
then
redis.call("lpop", KEYS[2])
redis.call("zrem", KEYS[3], ARGV[2]) -- 移除有序集合中的一個(gè)或多個(gè)成員
redis.call("hset", KEYS[1], ARGV[2], 1) -- 將哈希表 key 中的字段 field 的值設(shè)為 value 。
redis.call("pexpire", KEYS[1], ARGV[1])
return nil
end
if (redis.call("hexists", KEYS[1], ARGV[2]) == 1) then -- HEXISTS key field 查看哈希表 key 中，指定的字段是否存在。
redis.call("hincrby", KEYS[1], ARGV[2], 1) -- HINCRBY key field increment 為哈希表 key 中的指定字段的整數(shù)值加上增量 increment 。
redis.call("pexpire", KEYS[1], ARGV[1]) -- Redis PEXPIRE 命令和 EXPIRE 命令的作用類似，但是它以毫秒為單位設(shè)置 key 的生存時(shí)間，而不像 EXPIRE 命令那樣，以秒為單位。
return nil
end
local firstThreadId = redis.call("lindex", KEYS[2], 0)
local ttl
if firstThreadId ~= false and firstThreadId ~= ARGV[2] then
ttl = tonumber(redis.call("zscore", KEYS[3], firstThreadId)) - tonumber(ARGV[4])
else
ttl = redis.call("pttl", KEYS[1]) -- Redis Pttl 命令以毫秒為單位返回 key 的剩余過(guò)期時(shí)間。
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
再來(lái)看解鎖的邏輯。我前面加鎖的一些內(nèi)容對(duì)應(yīng)著看，重點(diǎn)在于 「publish」這里，在輪到某個(gè)線程時(shí)，nextThreadId 這個(gè) channel 會(huì)收到通知。
這里的 threadId 是我們?cè)诩渔i和解鎖的時(shí)候都需要傳入的。如果你留意過(guò) Java 的線程 Id 就會(huì)發(fā)現(xiàn)，不同實(shí)例之間有很大概率會(huì)重復(fù)的。為了避免，各個(gè) Client 在傳入 ThradId 的時(shí)候，除了真實(shí)的 id 外，還需要加入各個(gè) client 對(duì)應(yīng)的信息加以區(qū)分。
--[[
用于 unlock 操作。
KEYS[1] = lockName
KEYS[2] = waitQueueName
KEYS[3] = timeoutName
KEYS[4] = channelName
ARGV[1] = message
ARGV[2] = leaseTime
ARGV[3] = lockName
ARGV[3] = currentTime
--]]
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
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
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
看過(guò)了解鎖邏輯后，外面eval 執(zhí)行加鎖的時(shí)候，需要對(duì)應(yīng)有 sub ，才會(huì)收到這里解鎖的 pub 信息，否則就卡住了。
這里需要注意下，sub 這個(gè)功能是個(gè)阻塞操作，需要單獨(dú)的線程里執(zhí)行，通過(guò)一個(gè) Future 來(lái)實(shí)現(xiàn)一定等待時(shí)間的 sub 功能，超時(shí)再 unsub。這塊邏輯 Redission 封裝的比較多，感興趣的可以到源碼里點(diǎn)點(diǎn)。
網(wǎng)站名稱：怎樣實(shí)現(xiàn)一個(gè)分布式的公平鎖?
文章分享：http://www.5511xx.com/article/djsjphh.html
