.net core 分布式锁之基于 Redis 的 RedLock-526互联

使用场景

分布式锁的业务场景涉及到并发控制、任务调度、缓存更新、分布式事务和防止重复操作等方面，能够保证分布式系统的数据一致性和正确性。

并发控制：当多个线程或进程同时访问共享资源时，使用分布式锁可以确保只有一个线程或进程能够访问该资源，避免数据竞争和并发冲突。
分布式任务调度：在分布式系统中，多个节点可能同时竞争执行某个任务。使用分布式锁可以确保只有一个节点能够获取到任务的执行权限，避免重复执行和资源浪费。
缓存更新：在使用缓存的场景下，当缓存失效时，多个请求可能会同时访问数据库或其他资源来重新生成缓存。使用分布式锁可以确保只有一个请求能够重新生成缓存，避免缓存雪崩和数据库压力过大。
分布式事务：在分布式系统中，涉及到多个服务之间的事务操作时，使用分布式锁可以协调各个服务的操作顺序和一致性，保证分布式事务的正确执行。
防止重复操作：在某些业务场景下，需要确保某个操作只能执行一次，例如订单支付、秒杀等。使用分布式锁可以防止重复操作，确保每个操作只会执行一次。

秒杀业务模拟

不使用锁的业务

备注：这里使用了parallel模拟多线程并发执行操作，也可以用Jemeter来模拟，后面的示例我就使用Jemeter来模拟并发

 public bool Buy3()
        {
            // 模拟执行的逻辑代码花费的时间
            Thread.Sleep(200);
            if (stock.stockCount > 0)
            {
                Thread.Sleep(10);  //这里为了更能够看出效果，休眠了10毫秒
                stock.stockCount--;
                return true;
            }
            return false;

        }

public static class stock
    {
        // 有5个商品库存
        public static int stockCount = 5;
    }

[HttpGet("TestLock")]
        public async Task<IActionResult> TestLock()
        {
            Test test=new Test();
            // 使用parallel模拟多线程并发执行操作
            Parallel.For(0, 15, (index) =>     
            {
                var stopwatch = new Stopwatch();
                stopwatch.Start();
                var data = test.Buy3();
                stopwatch.Stop();
                Console.WriteLine($"ThreadId:{Thread.CurrentThread.ManagedThreadId}, Result:{data}, Time:{stopwatch.ElapsedMilliseconds}");
            });
            return Ok();
        }

执行后，我们的5个库存变成了8个执行成功，出现了超卖的情况

单机锁

在没有集群的情况下，只有一个进程可以使用单机锁，会比分布式锁效率更高，因为是在进程内进行的，不会涉及到和其他应用进行通信

        public object obj=new object();

        public bool Buy()
        {
            Thread.Sleep(200);
            lock (obj)
            {
                if (stock.stockCount > 0)
                {
                    Thread.Sleep(10);
                    stock.stockCount--;
                    return true;
                }
                return false;
            }
           
        }

        [HttpGet("TestLock3")]
        public async Task<IActionResult> TestLock3()
        {
            Test test = new Test();
            var stopwatch = new Stopwatch();
            stopwatch.Start();
            var data = test.Buy();
            stopwatch.Stop();
            Console.WriteLine($"ThreadId:{Thread.CurrentThread.ManagedThreadId}, Result:{data}, Time:{stopwatch.ElapsedMilliseconds}");
            return Ok();
        }

执行后，结果正确，可以看到5个卖出，但是这个执行的时候每个进程都要等待前一个进程执行成功才能够开始执行，所以速度相对不加锁是比较慢的，但是没办法，为了保证数据的正确性，所以分清楚场景，并不是每一个场景都适合用单机锁和分布式锁

分布式锁RedLock

官方使用地址：samcook/RedLock.net：C语言中Redlock算法的实现# (github.com)

在集群模式下，系统部署于多台机器（一个系统运行在多个进程中），语言本身实现的锁只能确保当前进程内有效（基于内存），多进程就没办法共享锁状态，这时我们就得考虑采用分布式锁，分布式锁可以采用数据库、ZooKeeper、Redis等来实现，最终都是为了达到在不同的进程、线程内能共享锁状态的目的。

RedLock的原理：

　　RedLock 的思想是使用多台 Redis Master ，节点完全独立，节点间不需要进行数据同步，因为 Master-Slave 架构一旦 Master 发生故障时数据没有复制到 Slave，被选为 Master 的 Slave 就丢掉了锁，另一个客户端就可以再次拿到锁。锁通过 setNX（原子操作）命令设置，在有效时间内当获得锁的数量大于 (n/2+1) 代表成功，失败后需要向所有节点发送释放锁的消息。

官方的说明：

1. 引入nuget包

RedLock.net
StackExchange.Redis   //如果RedLock 2.2.0+ 需要 StackExchange.Redis 2.0+

2. 配置Program.cs

#region 分布式锁redlock
var existingConnectionMultiplexer1 = ConnectionMultiplexer.Connect("redis连接地址");
var multiplexers = new List<RedLockMultiplexer>
            {
                existingConnectionMultiplexer1
                //existingConnectionMultiplexer2,                            //redis节点2
                //existingConnectionMultiplexer3                            //redis节点3
            };
var redlockFactory = RedLockFactory.Create(multiplexers);
//注入锁和服务
builder.Services.AddSingleton(typeof(IDistributedLockFactory), redlockFactory);
#endregion

#region 应用生命周期释放分布式锁
app.Lifetime.ApplicationStopping.Register(() =>
{
    redlockFactory.Dispose();
});
#endregion

3. 业务模拟代码

        private readonly IDistributedLockFactory _distributedLockFactory;

        public ValuesController(IDistributedLockFactory distributedLockFactory)
        {
            _distributedLockFactory = distributedLockFactory;
        }
        [HttpGet("TestLock2")]
        public async Task<IActionResult> TestLock2()
        {
            var resource = "the-thing-we-are-locking-on";  // 锁定的对象
            var expiry = TimeSpan.FromSeconds(30);  //  锁定过期时间，锁区域内的逻辑执行如果超过过期时间，锁将被释放
            var wait = TimeSpan.FromSeconds(10);   //等待时间,相同的 resource 如果当前的锁被其他线程占用,最多等待时间,超过这个时间就不会等待锁了,即 redLock.IsAcquired 为false 
            var retry = TimeSpan.FromSeconds(1); //等待时间内，间隔多久尝试获取一次，如果超过上面设置的等待时间还没有获取到锁，也不会等待锁，即redLock.IsAcquired 为false 
            
            await using (var redLock = await _distributedLockFactory.CreateLockAsync(resource, expiry, wait, retry)) // 这里也可以不使用 wait 和 retry ，如果不使用当出现锁被占用就直接跳过，不会等待
            {
                // make sure we got the lock
                if (redLock.IsAcquired)
                {
                    var stopwatch = new Stopwatch();
                    stopwatch.Start();
                    var data = new Test().Buy2();
                    stopwatch.Stop();
                    Console.WriteLine($"ThreadId:{Thread.CurrentThread.ManagedThreadId}, Result:{data}, Time:{stopwatch.ElapsedMilliseconds}");
                    return Ok($"ThreadId:{Thread.CurrentThread.ManagedThreadId}, Result:{data}, Time:{stopwatch.ElapsedMilliseconds}");
                }
                else
                {
                    Console.WriteLine("等待超时");
                    return Ok("等待超时");
                }
            }
        }