當(dāng)前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

RedLock 实现分布式锁

發(fā)布時(shí)間：2023/12/4 编程问答 26 豆豆

生活随笔收集整理的這篇文章主要介紹了 RedLock 实现分布式锁小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

并發(fā)是程序開發(fā)中不可避免的問題，根據(jù)系統(tǒng)面向用戶、功能場(chǎng)景的不同，并發(fā)的重視程度會(huì)有不同。從程序的角度來說，并發(fā)意味著相同的時(shí)間點(diǎn)執(zhí)行了相同的代碼，而有些情況是不被允許的，比如：轉(zhuǎn)賬、搶購占庫存等，如果沒有做好臨界條件的驗(yàn)證，會(huì)帶來非常嚴(yán)重的后果。追根結(jié)底是因?yàn)椴l(fā)引起的數(shù)據(jù)不一致問題，面對(duì)并發(fā)，我們通常會(huì)采用鎖來優(yōu)化。

場(chǎng)景模擬

如下模擬搶購的示例代碼（C#）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

// 有10個(gè)商品庫存
private static int stockCount = 10;

public bool Buy()
{
// 模擬執(zhí)行的邏輯代碼花費(fèi)的時(shí)間
Thread.Sleep(new Random().Next(100,500));
if (stockCount > 0)
{
stockCount--;
return true;
}
return false;
}

1
2
3
4
5
6
7
8
9
10
11

模擬并行調(diào)用 Buy 方法 15 次（內(nèi)部使用的是線程池，所以 ThreadId 會(huì)有重復(fù)），實(shí)際上只有 10 個(gè)庫存，返回結(jié)果卻顯示 11 個(gè)請(qǐng)求都購買成功了。

單機(jī)部署模式解決方案

在單機(jī)部署模式下，我們只需要加 lock(){} 就可以解決問題：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

// 有10個(gè)商品庫存
private static int stockCount = 10;

private static object obj = new object();

public bool Buy()
{
lock (obj)
{
// 模擬執(zhí)行的邏輯代碼花費(fèi)的時(shí)間
Thread.Sleep(new Random().Next(100, 500));
if (stockCount > 0)
{
stockCount--;
return true;
}
return false;
}
}

從輸出結(jié)果中可以看出，確實(shí)只有10個(gè)請(qǐng)求是顯示購買成功，但同時(shí)發(fā)現(xiàn)部分請(qǐng)求的執(zhí)行時(shí)間明顯變長(zhǎng)，這就是加鎖帶來的最直觀影響，當(dāng)某個(gè)線程獲得鎖之后，在沒有釋放之前，其他線程只能繼續(xù)等待，并發(fā)越高，更多的線程需要等待輪流被處理。

各種語言一般都提供了鎖的實(shí)現(xiàn)，用法大同小異，語言本身實(shí)現(xiàn)的鎖只能作用于當(dāng)前進(jìn)程內(nèi)，所以在單機(jī)模式部署的系統(tǒng)中使用基本沒什么問題。

集群部署模式解決方案（分布式鎖）

在集群模式下，系統(tǒng)部署于多臺(tái)機(jī)器（一個(gè)系統(tǒng)運(yùn)行在多個(gè)進(jìn)程中），語言本身實(shí)現(xiàn)的鎖只能確保當(dāng)前進(jìn)程內(nèi)有效（基于內(nèi)存），多進(jìn)程就沒辦法共享鎖狀態(tài)，這時(shí)我們就得考慮采用分布式鎖，分布式鎖可以采用?數(shù)據(jù)庫、ZooKeeper、Redis?等來實(shí)現(xiàn)，最終都是為了達(dá)到在不同的進(jìn)程、線程內(nèi)能共享鎖狀態(tài)的目的。

這里將介紹基于 Redis 的?RedLock.net?來解決分布式下的并發(fā)問題，RedLock.net 是 RedLock 分布式鎖算法的 .NET 版實(shí)現(xiàn) (大部分語言都有對(duì)應(yīng)的實(shí)現(xiàn)，查看) ，RedLock 分布式鎖算法是由 Redis 的作者提出。

RedLock 簡(jiǎn)介

RedLock 的思想是使用多臺(tái) Redis Master ，節(jié)點(diǎn)完全獨(dú)立，節(jié)點(diǎn)間不需要進(jìn)行數(shù)據(jù)同步，因?yàn)?Master-Slave 架構(gòu)一旦 Master 發(fā)生故障時(shí)數(shù)據(jù)沒有復(fù)制到 Slave，被選為 Master 的 Slave 就丟掉了鎖，另一個(gè)客戶端就可以再次拿到鎖。鎖通過 setNX（原子操作）命令設(shè)置，在有效時(shí)間內(nèi)當(dāng)獲得鎖的數(shù)量大于 (n/2+1) 代表成功，失敗后需要向所有節(jié)點(diǎn)發(fā)送釋放鎖的消息。

獲取鎖：

1	SET resource_name my_random_value NX PX 30000

釋放鎖：

1
2
3
4
5

if redis.call("get",KEYS[1]) == ARGV[1] then
return redis.call("del",KEYS[1])
else
return 0
end

RedLock.net 集成

創(chuàng)建 .NETCore API 項(xiàng)目

Nuget 安裝 RedLock.net

1	Install-Package RedLock.net

appsettings.json 添加 redis 配置

1
2
3
4

{
"RedisUrl": "127.0.0.1:6379", // 多個(gè)用,分割
...
}

添加 ProductService.cs，模擬商品購買

1
2
3
4
5
6
7
8
9
10
11
12
13

// 有10個(gè)商品庫存，如果同時(shí)啟動(dòng)多個(gè)API服務(wù)進(jìn)行測(cè)試，這里改成存數(shù)據(jù)庫或其他方式
private static int stockCount = 10;
public async Task<bool> BuyAsync()
{
// 模擬執(zhí)行的邏輯代碼花費(fèi)的時(shí)間
await Task.Delay(new Random().Next(100, 500));
if (stockCount > 0)
{
stockCount--;
return true;
}
return false;
}

修改 Startup.cs ，創(chuàng)建 RedLockFactory

定義 RedLockFactory 變量：

1	private RedLockFactory lockFactory;

添加方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

private RedLockFactory GetRedLockFactory()
{
var redisUrl = Configuration["RedisUrl"];
if (string.IsNullOrEmpty(redisUrl))
{
throw new ArgumentException("RedisUrl 不能為空");
}
var urls = redisUrl.Split(",").ToList();
var endPoints = new List<RedLockEndPoint>();
foreach (var item in urls)
{
var arr = item.Split(":");
endPoints.Add(new DnsEndPoint(arr[0], Convert.ToInt32(arr[1])));
}
return RedLockFactory.Create(endPoints);
}

在 ConfigureServices 注入 IDistributedLockFactory：

1
2
3

lockFactory = GetRedLockFactory();
services.AddSingleton(typeof(IDistributedLockFactory), lockFactory);
services.AddScoped(typeof(ProductService));

修改 Configure，應(yīng)用程序結(jié)束時(shí)釋放 lockFactory ：

1
2
3
4
5
6
7
8
9

public void Configure(IApplicationBuilder app, IHostingEnvironment env, IApplicationLifetime lifetime)
{
...

lifetime.ApplicationStopping.Register(() =>
{
lockFactory.Dispose();
});
}

在 Controller 添加方法 DistributedLockTest

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

private readonly IDistributedLockFactory _distributedLockFactory;
private readonly ProductService _productService;

public HomeController(IDistributedLockFactory distributedLockFactory,
ProductService productService)
{
_distributedLockFactory = distributedLockFactory;
_productService = productService;
}

[HttpGet]
public async Task<bool> DistributedLockTest()
{
var productId = "id";
// resource 鎖定的對(duì)象
// expiryTime 鎖定過期時(shí)間，鎖區(qū)域內(nèi)的邏輯執(zhí)行如果超過過期時(shí)間，鎖將被釋放
// waitTime 等待時(shí)間,相同的 resource 如果當(dāng)前的鎖被其他線程占用,最多等待時(shí)間
// retryTime 等待時(shí)間內(nèi)，多久嘗試獲取一次
using (var redLock = await _distributedLockFactory.CreateLockAsync(productId, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(1), TimeSpan.FromMilliseconds(20)))
{
if (redLock.IsAcquired)
{
var result = await _productService.BuyAsync();
return result;
}
else
{
Console.WriteLine($"獲取鎖失敗：{DateTime.Now}");
}
}
return false;
}

調(diào)用接口測(cè)試

1
2
3
4
5
6
7
8

Parallel.For(1, 16, (i) =>
{
var stopwatch = new Stopwatch();
stopwatch.Start();
var data = GetAsync($"http://localhost:5000/home/distributedLockTest").Result;
stopwatch.Stop();
Console.WriteLine($"ThreadId:{Thread.CurrentThread.ManagedThreadId}, Result:{data}, Time:{stopwatch.ElapsedMilliseconds}");
});

關(guān)于 RedLock 分布式鎖算法的爭(zhēng)議大家可以參考：

How to do distributed locking
Is Redlock safe?

總結(jié)

如果使用鎖，必然對(duì)性能上會(huì)有一定影響，我們需要根據(jù)實(shí)際場(chǎng)景來判斷是真正需要。在指定鎖過期時(shí)間時(shí)要相對(duì)合理，避免出現(xiàn)鎖已過期，但邏輯還沒執(zhí)行完成，這樣就失去了鎖的意義，當(dāng)然這種情況下我們還可以考慮重入鎖。

最后推薦一下微軟開源的一個(gè)基于 Actor 模型的分布式框架?Orleans，也可以達(dá)到分布式鎖的效果。

參考鏈接

Distributed locks with Redis
RedLock.net

原文地址：http://beckjin.com/2019/01/06/redLock-net/

.NET社區(qū)新聞，深度好文，歡迎訪問公眾號(hào)文章匯總?http://www.csharpkit.com?

總結(jié)

以上是生活随笔為你收集整理的RedLock 实现分布式锁的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇：基于Kebernetes 构建.NET
下一篇： [翻译] NumSharp的数组切片功能