> 科技资讯 > .NET开源分布式锁DistributedLock

.NET开源分布式锁DistributedLock

科技资讯 投稿 7100℃ 0 评论

.NET开源分布式锁DistributedLock

一、线程锁和分布式锁

在我们.NET中常见的线程锁有：

    自旋锁：当线程尝试获取锁时，它会重复执行一些简单的指令，直到锁可用
  • 互斥锁: Mutex，可以跨进程使用。Mutex 类定义了一个互斥体对象，可以使用 WaitOne( 方法等待对象上的锁
  • 混合锁：Monitor，可以通过 lock 关键字来使用
  • 读写锁：允许多个线程同时读取共享资源，但只允许单个线程写入共享资源
  • 信号量：Semaphore，它允许多个线程同时访问同一个资源

分布式锁是一种用于协调多个进程/节点之间的并发访问的机制，某个资源在同一时刻只能被一个应用所使用,可以通过一些共享的外部存储系统来实现跨进程的同步和互斥

    Redis 分布式锁
  • ZooKeeper 分布式锁
  • Mysql 分布式锁
  • SqlServer 分布式锁
  • 文件分布式锁

DistributedLock开源项目中有多种实现方式，我们今天主要讨论Redis中的分布式锁实现。

二、Redis分布式锁的实现原理

基础实现

Redis 的 SET 命令有个 NX 参数可以实现「key不存在才插入」，所以可以用它来实现分布式锁：

    如果 key 不存在，则显示插入成功，可以用来表示加锁成功；
  • 如果 key 存在，则会显示插入失败，可以用来表示加锁失败。
SET lock_keyunique_value NX PX 10000
    lock_key 就是 key 键；
  • unique_value 是客户端生成的唯一的标识，区分来自不同客户端的锁操作；
  • NX 代表只在 lock_key 不存在时，才对 lock_key 进行设置操作；
  • PX 10000 表示设置 lock_key 的过期时间为 10s，这是为了避免客户端发生异常而无法释放锁。

释放锁的时候需要删除key，或者使用lua脚本来保证原子性。

// 释放锁时，先比较 unique_value 是否相等，避免锁的误释放
if redis.call("get",KEYS[1] == ARGV[1] then
    return redis.call("del",KEYS[1]
else
    return 0
end

续租机制

基于上文中的实现方式，我们在设置key过期时间时，不能准确的描述业务处理时间。为了防止因为业务处理时间较长导致锁过期而提前释放锁，通过不断更新锁的过期时间来保持锁的有效性，避免了因锁过期而导致的并发问题。

1、实现自动续租机制：额外起一个线程，定期检查线程是否还持有锁，如果有则延长过期时间。DistributedLock里面就实现了这个方案，使用“看门狗”定期检查（每1/3的锁时间检查1次），如果线程还持有锁，则刷新过期时间。

以下是使用StackExchange.Redis 库实现分布式锁和续租机制的示例代码：

public class RedisLock
{
    private readonly IDatabase _database;
    private readonly string _lockKey;
    private string _lockValue;
    private readonly TimeSpan _lockTimeout;

    private readonly TimeSpan _renewInterval;
    private bool _isLocked;

    public RedisLock(IDatabase database, string lockKey, TimeSpan lockTimeout, TimeSpan renewInterval
    {
        _database = database;
        _lockKey = lockKey;
        _lockTimeout = lockTimeout;
        _renewInterval = renewInterval;
    }

    //尝试获取锁，如果成功，则启动一个续租线程
    public async Task<bool> AcquireAsync(
    {
        _lockValue = Guid.NewGuid(.ToString(;
        var acquired = await _database.StringSetAsync(_lockKey, _lockValue, _lockTimeout, When.NotExists;
        if (acquired
        {
            _isLocked = true;
            StartRenewal(;
        }
        return acquired;
    }

    //定期使用 KeyExpireAsync 命令重置键的过期时间，从而实现续租机制
    private async void StartRenewal(
    {
        while (_isLocked
        {
            await Task.Delay(_renewInterval;
            await _database.KeyExpireAsync(_lockKey, _lockTimeout;
        }
    }
}

RedLock

Redlock 是一种分布式锁实现方案，它的设计目标是解决 Redis 集群模式下的分布式锁并发控制问题。

Redlock 算法加锁三个过程：

    客户端获取当前时间（t1）。
  1. 客户端按顺序依次向 N 个 Redis 节点(官方推荐是至少部署 5 个 Redis 节点执行加锁操作：
    加锁操作使用 SET 命令，带上 NX，EX/PX 选项，以及带上客户端的唯一标识。
  • 如果某个 Redis 节点发生故障了，为了保证在这种情况下，Redlock 算法能够继续运行，我们需要给「加锁操作」设置一个超时时间（不是对「锁」设置超时时间，而是对「加锁操作」设置超时时间），加锁操作的超时时间需要远远地小于锁的过期时间，一般也就是设置为几十毫秒。
  1. 一旦客户端从超过半数（大于等于 N/2+1）的 Redis 节点上成功获取到了锁，就再次获取当前时间（t2），然后计算计算整个加锁过程的总耗时（t2-t1）。如果 t2-t1 < 锁的过期时间，此时，认为客户端加锁成功，否则认为加锁失败。

加锁失败后，客户端向所有 Redis 节点发起释放锁的操作，释放锁的操作和在单节点上释放锁的操作一样，只要执行释放锁的 Lua 脚本就可以了。

三、DistributedLock开源项目简介

项目介绍

DistributedLock 包含基于各种技术的实现；可以单独安装实现包，也可以只安装 DistributedLock NuGet 包，这是一个“元”包，其中包含所有实现作为依赖项。请注意，每个包都根据 SemVer 独立进行版本控制。

基础使用

    Acquire 方法

Acquire 方法返回一个代表持有锁的“句柄”对象。当句柄被处理时，锁被释放：

  var redisDistributedLock = new RedisDistributedLock(name, connectionString; 
  using (redisDistributedLock.Acquire(
  {
      //持有锁
  } //释放锁及相关资源
    TryAcquire 方法

虽然 Acquire 将阻塞直到锁可用，但还有一个 TryAcquire 变体，如果无法获取锁（由于在别处持有），则返回 null ：

using (var handle = redisDistributedLock.TryAcquire(
{
    if (handle != null
    {
        // 我们获得锁
    }
    else
    {
        // 别人获得锁
    }
}

支持异步和依赖注入，依赖注入：

// Startup.cs:
services.AddSingleton<IDistributedLockProvider>(_ => new PostgresDistributedSynchronizationProvider(myConnectionString;
services.AddTransient<SomeService>(;

// SomeService.cs
public class SomeService
{
    private readonly IDistributedLockProvider _synchronizationProvider;

    public SomeService(IDistributedLockProvider synchronizationProvider
    {
        this._synchronizationProvider = synchronizationProvider;
    }

    public void InitializeUserAccount(int id
    {
        // 通过provider构造lock
        var @lock = this._synchronizationProvider.CreateLock($"UserAccount{id}";
        using (@lock.Acquire(
        {
            // 
        }
      
        using (this._synchronizationProvider.AcquireLock($"UserAccount{id}"
        {
            // 
        }
    }
}

四、浅析DistributedLock的Redis实现

源码地址

https://github.com/madelson/DistributedLock

目录解析

    DistributedLock.Core 是项目的抽象类库，基础分布式锁、读写锁、信号量的Provider和接口。
  • 其它几个类库是用不同存储系统的具体实现

Redis的实现过程

定义一个工厂接口，返回IDistributedLock，在依赖注入场景中，使用这个工厂接口可能会更加方便

public interface IDistributedLockProvider
{
    IDistributedLock CreateLock(string name;
}

IDistributedLock：定义了控制并发访问的基本操作。该接口支持同步和异步方式获取锁，并提供超时和取消功能，以适应各种情况

public interface IDistributedLock
{
    // 唯一Name
    string Name { get; }
    // 获取锁的方法
    IDistributedSynchronizationHandle Acquire(TimeSpan? timeout = null, CancellationToken cancellationToken = default;

    //......
}

DistributedLock.Redis类库，对Acquire的具体实现，该方法是尝试获取Redis分布式锁实例。

  private async ValueTask<RedisDistributedLockHandle?> TryAcquireAsync(CancellationToken cancellationToken
  {
      // 初始化Redis连接和相关参数
      //CreateLockId = $"{Environment.MachineName}_{currentProcess.Id}_" + Guid.NewGuid(.ToString("n"
      var primitive = new RedisMutexPrimitive(this.Key, RedLockHelper.CreateLockId(, this._options.RedLockTimeouts;

      // 获取和设置锁
      var tryAcquireTasks = await new RedLockAcquire(primitive, this._databases, cancellationToken.TryAcquireAsync(.ConfigureAwait(false;

      // 成功后，RedLockHandle这个里边实现了续租机制
      return tryAcquireTasks != null 
          ? new RedisDistributedLockHandle(new RedLockHandle(primitive, tryAcquireTasks, extensionCadence: this._options.ExtensionCadence, expiry: this._options.RedLockTimeouts.Expiry 
          : null;
  }

根据当前线程是否在同步上下文，对单库和多库实现进行区分和实现

// 该方法用于尝试获取分布式锁，并返回一个表示各个数据库节点获取锁状态的任务字典
public async ValueTask<Dictionary<IDatabase, Task<bool>>?> TryAcquireAsync(
{
    // 检查当前线程是否在同步上下文中执行，以便根据不同情况采取不同的获取锁策略
    if (SyncViaAsync.IsSynchronous&& this._databases.Count == 1
        return this.TrySingleFullySynchronousAcquire(;

    // 创建一个任务字典，将每个数据库连接和其对应的获取锁任务关联起来
    var tryAcquireTasks = this._databases.ToDictionary(
        db => db,
        db => Helpers.SafeCreateTask(state => state.primitive.TryAcquireAsync(state.db, (primitive, db
    ;

    // 等待所有获取锁任务完成，并返回一个表示整体状态的任务
    var waitForAcquireTask = this.WaitForAcquireAsync(tryAcquireTasks.AwaitSyncOverAsync(.ConfigureAwait(false;

    // 执行清理操作 
 
    // 返回结果
    return succeeded ? tryAcquireTasks : null;
}

单库获取Redis分布式锁，就是通过set nx 设置值，返回bool,失败就释放资源，成功检查是否超时。不超时就返回任务字典

private Dictionary<IDatabase, Task<bool>>? TrySingleFullySynchronousAcquire(
{
    var database = this._databases.Single(;

    bool success;
    var stopwatch = Stopwatch.StartNew(;

    // 通过StackExchange.Redis的StringSet进行无值设置key（set nx）
    try { success = this._primitive.TryAcquire(database; }
    catch
    {
        // 确保释放锁，以便防止出现死锁等问题。然后重新抛出异常
    }

    if (success
    {
        // 检查是否在超时时间内，并返回一个包含成功状态的任务字典；否则继续释放锁并返回null
    }

    return null;
}

多库中是否获取到分布式锁

private async Task<bool> WaitForAcquireAsync(IReadOnlyDictionary<IDatabase, Task<bool>> tryAcquireTasks
{
    // 超时或取消时自动停止等待

    using var timeout = new TimeoutTask(this._primitive.AcquireTimeout, this._cancellationToken;
    var incompleteTasks = new HashSet<Task>(tryAcquireTasks.Values { timeout.Task };

    // 计数器
    var successCount = 0;
    var failCount = 0;
    var faultCount = 0;

    while (true
    {
        // 不断等待任务完成，如果任务为timeout，则表示超时；否则需要根据任务的状态和信号来判断是否成功获取锁
        var completed = await Task.WhenAny(incompleteTasks.ConfigureAwait(false;

        if (completed == timeout.Task
            return false; // 超时

        // 判断是否超过成功或者失败的阀值,是否超过1/2
        if (completed.Status == TaskStatus.RanToCompletion
        {
            var result = await ((Task<bool>completed.ConfigureAwait(false;

            if (result
            {
                ++successCount;
                // 是否超过1/2的库
                if (RedLockHelper.HasSufficientSuccesses(successCount, this._databases.Count { return true; }
            }
            else
            {
                ++failCount;
                if (RedLockHelper.HasTooManyFailuresOrFaults(failCount, this._databases.Count { return false; }
            }
        }
        else 
        {      
            ++faultCount;
            // ......          
        }
        // ......
    }
}

截止到目前，我们就知道如何获取和设置分布式锁了。接下来我们就看下是如何实现续租机制的。就是LeaseMonitor这个对象。

private static Task CreateMonitoringLoopTask(WeakReference<LeaseMonitor> weakMonitor, TimeoutValue monitoringCadence, CancellationToken disposalToken
{
    // 创建监视任务
    return Task.Run(( => MonitoringLoop(;

    async Task MonitoringLoop(
    {
        var leaseLifetime = Stopwatch.StartNew(;
        do
        {
            await Task.Delay(monitoringCadence.InMilliseconds, disposalToken.TryAwait(;
        }
        // 检查RedLock租约的状态和可用性
        while (!disposalToken.IsCancellationRequested && await RunMonitoringLoopIterationAsync(weakMonitor, leaseLifetime.ConfigureAwait(false;
    }
}

RunMonitoringLoopIterationAsync 里边最终调用了续时的lua脚本

你们在公司中，都是如何实现分布式锁的呢？可以在评论区留下您宝贵的建议。

编程笔记 » .NET开源分布式锁DistributedLock

赞同 (30) or 分享 (0)
游客 发表我的评论   换个身份
取消评论

表情
(0)个小伙伴在吐槽