建站
咨詢
售后
建站咨詢
新聞中心
日常開發(fā)中，秒殺下單、搶紅包等等業(yè)務(wù)場景，都需要用到分布式鎖。而Redis非常適合作為分布式鎖使用。
分布式鎖是”線程同步”的延續(xù)
最近首度應(yīng)用”分布式鎖”，現(xiàn)在想想，分布式鎖不是孤立的技能點(diǎn)，這其實(shí)就是跨主機(jī)的線程同步。
單機(jī)服務(wù)器可以通過共享某堆內(nèi)存來標(biāo)記上鎖/解鎖，線程同步說到底是建立在單機(jī)操作系統(tǒng)的用戶態(tài)/內(nèi)核態(tài)對共享內(nèi)存的訪問控制。
而分布式服務(wù)器不是在同一臺機(jī)器上：跨主機(jī)，因此需要將鎖標(biāo)記存儲在所有機(jī)器進(jìn)程都能看到的地方。
在開發(fā)很多業(yè)務(wù)場景會使用到鎖，例如庫存控制，抽獎(jiǎng)等。
例如庫存只剩1個(gè)商品，有三個(gè)用戶同時(shí)打算購買，誰先購買庫存立即清零，不能讓其他二人也購買成功。
解讀分布式鎖
我們常說的線程安全、線程同步方案，包括此次的分布式鎖都是基于
“多線程/多進(jìn)程對特定資源同時(shí)有更新操作”。
基本考量
-
分布式系統(tǒng)，一個(gè)鎖在同一時(shí)間只能被一個(gè)服務(wù)器獲取 (這是分布式鎖的基礎(chǔ))
-
具備鎖失效機(jī)制，防止死鎖 (防止某些意外，鎖沒有得到釋放，別人也無法得到鎖)
Redis SET resource-name anystring NX EX max-lock-time
是一種最簡單的分布式鎖實(shí)現(xiàn)方案。
SET 命令支持多個(gè)參數(shù)：
-
EX seconds– 設(shè)置過期時(shí)間(s)
-
NX — 如果key不存在，則設(shè)置 ……
因?yàn)镾ET命令參數(shù)可以替代SETNX，SETEX，GETSET，這些命令在未來可能被廢棄。
上面的命令返回OK(或經(jīng)過重試)，客戶端就獲取到這個(gè)鎖;
使用DEL命令解鎖;到達(dá)超時(shí)時(shí)間會自動(dòng)釋放鎖。
在解鎖時(shí)，增加一些設(shè)計(jì)，讓系統(tǒng)更加健壯：
3.不要使用固定的String值作為鎖標(biāo)記值，而是使用一個(gè)不易被猜中的隨機(jī)值， 業(yè)內(nèi)稱為token
4.不使用DEL命令釋放鎖，而是發(fā)送script去移除key
第3、4點(diǎn)是為了解決 ：“鎖提前過期，客戶端A還沒有執(zhí)行完，然后客戶端B獲取了鎖，這時(shí)客戶端A執(zhí)行完了，會不會在刪鎖的時(shí)候把B的鎖給刪掉” — 4是3技術(shù)上的推薦實(shí)現(xiàn)。
腳本如下：
if redis.call("get",KEYS1] ==ARGV[1])
then
return redis.call("DEL",KEYS[1])
else
return 0
end
下面使用StackExchange.Redis 寫了基于以上考量的代碼示例：
///
/// Acquires the lock.
///
///
/// 隨機(jī)值
///
/// 非阻塞鎖
static bool Lock(string key, string token,int expireSecond=10, double waitLockSeconds = 0)
{
var waitIntervalMs = 50;
bool isLock;
DateTime begin = DateTime.Now;
do
{
isLock = Connection.GetDatabase().StringSet(key, token, TimeSpan.FromSeconds(expireSecond), When.NotExists);
if (isLock)
return true;
//不等待鎖則返回
if (waitLockSeconds == 0) break;
//超過等待時(shí)間，則不再等待
if ((DateTime.Now - begin).TotalSeconds >= waitLockSeconds) break;
Thread.Sleep(waitIntervalMs);
} while (!isLock);
return false;
}
///
/// Releases the lock.
///
/// true, if lock was released, false otherwise.
/// Key.
/// value
static bool UnLock(string key, string value)
{
string lua_script = @" if (redis.call('GET', KEYS[1]) == ARGV[1]) then redis.call('DEL', KEYS[1]) return true else return false end ";
try
{
var res = Connection.GetDatabase().ScriptEvaluate(lua_script,
new RedisKey[] { key },
new RedisValue[] { value });
return (bool)res;
}
catch (Exception ex)
{
Console.WriteLine($"ReleaseLock lock fail...{ex.Message}");
return false;
}
}
private static Lazy lazyConnection = new Lazy(() =>
{
ConfigurationOptions configuration = new ConfigurationOptions
{
AbortOnConnectFail = false,
ConnectTimeout = 5000,
};
configuration.EndPoints.Add("10.100.219.9", 6379);
return ConnectionMultiplexer.Connect(configuration.ToString());
});
public static ConnectionMultiplexer Connection => lazyConnection.Value;
以上代碼新增了第五點(diǎn)考量：
\5. 為避免無限制搶鎖，增加了非阻塞鎖：輪詢_s等待鎖，未等到則不再搶鎖
使用方式：
下面并行開啟三個(gè)任務(wù)，同時(shí)減少庫存：
static void Main(string[] args)
{
// 嘗試并行執(zhí)行3個(gè)任務(wù)
Parallel.For(0, 3, x =>
{
string token = $"loki:{x}";
bool isLocked = Lock("loki", token, 5, 10);
if (isLocked)
{
Console.WriteLine($"{token} begin reduce stocks (with lock) at {DateTime.Now}.");
Thread.Sleep(1000);
Console.WriteLine($"{token} release lock {UnLock("loki", token)} at {DateTime.Now}. ");
}
else
{
Console.WriteLine($"{token} begin reduce stocks at {DateTime.Now}.");
}
});
}
可以看到三個(gè)并行任務(wù)依次獲取/釋放鎖
輸出總結(jié)
本文從基礎(chǔ)的線程安全、線程同步，認(rèn)識到分布式鎖是跨主機(jī)的資源線程/進(jìn)程同步方案， 以步步為營的風(fēng)格 演示了RedisSET命令做分布式鎖的設(shè)計(jì)考量，好記性不如爛筆頭。
網(wǎng)站名稱：Redis分布式鎖講解
URL標(biāo)題：http://m.fisionsoft.com.cn/article/cdspoie.html
