• .NET开源分布式锁DistributedLock


    一、线程锁和分布式锁

    线程锁通常在单个进程中使用,以防止多个线程同时访问共享资源。

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

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

    更多的线程同步锁,可以看这篇文章:https://www.cnblogs.com/Z7TS/p/16463494.html

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

    常见的分布式锁实现:

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

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

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

    基础实现

    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次),如果线程还持有锁,则刷新过期时间。

    2、实现快速失败机制:当我们解锁时发现锁已经被其他线程获取了,说明此时我们执行的操作已经是“不安全”的了,此时需要进行回滚,并返回失败。

    以下是使用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 集群模式下的分布式锁并发控制问题。

    它是基于多个 Redis 节点的分布式锁,即使有节点发生了故障,锁变量仍然是存在的,客户端还是可以完成锁操作

    Redlock 算法加锁三个过程:

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

    加锁成功后,客户端需要重新计算这把锁的有效时间,计算的结果是「锁最初设置的过期时间」减去「客户端从大多数节点获取锁的总耗时(t2-t1)」。如果计算的结果已经来不及完成共享数据的操作了,我们可以释放锁,以免出现还没完成数据操作,锁就过期了的情况。

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

    三、DistributedLock开源项目简介

    项目介绍

    DistributedLock 是一个 .NET 库,它基于各种底层技术提供强大且易于使用的分布式互斥体、读写器锁和信号量。

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

    基础使用

    以下两种方法,都是基于RedLock来实现的,在单机上,使用了续租机制,更多细节可以自己观看源码,下文中会简单介绍源码。

    • 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(_ => new PostgresDistributedSynchronizationProvider(myConnectionString));
    services.AddTransient();
    
    // 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 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 ValueTaskbool>>?> 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 Dictionarybool>>? 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(IReadOnlyDictionarybool>> tryAcquireTasks)
    {
        // 超时或取消时自动停止等待
    
        using var timeout = new TimeoutTask(this._primitive.AcquireTimeout, this._cancellationToken);
        var incompleteTasks = new HashSet(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 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脚本

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


    __EOF__

  • 本文作者: Broder
  • 本文链接: https://www.cnblogs.com/Z7TS/p/17359113.html
  • 关于博主: 评论和私信会在第一时间回复。或者直接私信我。
  • 版权声明: 本博客所有文章除特别声明外,均采用 BY-NC-SA 许可协议。转载请注明出处!
  • 声援博主: 如果您觉得文章对您有帮助,可以点击文章右下角推荐一下。
  • 相关阅读:
    MFC的CPen与CBush画图对象使用步骤
    ARM接口编程—UART(exynos 4412平台)
    Coremail城市沙龙活动|齐聚古城西安,护航信息安全
    Java_类和对象
    城市内涝一维二维耦合技术及在城市排水防涝领域中的实践应用
    从零实现深度学习框架——重构计算图的实现
    计算机毕业设计ssm+vue基本微信小程序的购物商城小程序
    独孤思维:副业再苦再累,请学会善待自己
    使用canvas(2d)+js实现一个简单的傅里叶级数绘制方波图
    Android 动画
  • 原文地址:https://www.cnblogs.com/Z7TS/p/17359113.html