Redis 5.x——分布式锁

分布式锁1、为什么需要分布式锁2、非阻塞锁的实现和原理3、阻塞锁的实现和原理4、Redlock原理4.1、什么是Redlock4.2、单节点上实现Redlock4.3、Redlock算法1、为什么需要分布式锁首先我们通过下面的范例来说明分布式的来历。假如我们需要做一个秒杀抢购的业务操作流程主要是先获取库存如果库存大于0就可以下单然后库存数量减1订单数量加1。正常情况下肯定是没有任何问题的在并发情况下很可能出现超卖问题下面将使用C#代码编写的程序来模拟超卖现象。var minute 30;// 秒杀对应的分钟 using (var client new RedisClient(127.0.0.1, 6379)) { // 设置库存数量为10 client.Setint(inventoryNum, 10); // 设置订单数为0 client.Setint(orderNum, 0); } // 启动10个线程去抢购 Console.WriteLine$在{minute}分0秒正式开启秒杀; var flag true; while (flag) { // 判断分钟数是否等于秒杀时间 if (DateTime.Now.Minute minute) { flag false; // 模拟并发操作 Parallel.For(0, 30, (i) { int temp i; // 启动子线程 Task.Run(() { using (var client new RedisClient(127.0.0.1, 6379)) { var inventory client.Getint (inventoryNum); if (inventory 0) { // 库存减1 client.Setint(inventoryNum, inventory - 1); // 订单数量加1 var orderNum client.Incr(orderNum); Console.WriteLine($抢购成功*****线程id { Thread.CurrentThread.ManagedThreadId. ToString(00)},库存{inventory}, 订单数量{orderNum}); } else { Console.WriteLine抢购失败; } } }); }); } }上面代码的运行结果如图所示出现了大量的超卖数据。本来只有10个库存在并发操作下下的订单数量明显超过了10个库存。要解决这个问题我们想到的是在模拟多个线程的地方加锁。改进后的代码如下... while (flag) { if (DateTime.Now.Minute minute) { flag false; // 模拟并发操作 Parallel.For(0, 30, (i) { int temp i; // 启动子线程 Task.Run(() { lock 不同业务不同资源对象 { // 业务操作判断库存减库存加订单 ... } }); }); } }上面代码的运行结果如图所示使用程序锁后没有出现超卖问题。但是当我们对当前业务进行分布式部署的时候同一个业务在不同的进程或者不同的服务器中​同一时间多个客户端请求到不同的服务进程时多个客户端同时进行秒杀抢购就会出现超卖情况抢购结果如图所示。同时启动三个进程然后进行秒杀抢购时还是出现了超卖现象这是因为在C#代码中的锁是依赖于当前应用程序进程的因为锁中的对象在同一个进程中是同一个在不同的进程中是多个对象所以启动三个进程的时候实质上是三把锁所以还是会出现超卖。这时我们应该在多个进程中维护同一个对象对同一个资源进行控制。当然也可以选择关系型数据库MySQL和SQLServer​通过给表中的数据加标记的方式时间戳来判断是否可以抢到锁。针对高并发的应用场景关系型数据库存在IO瓶颈分布式内存数据库Redis却是一个不错的选择。2、非阻塞锁的实现和原理实现非阻塞锁的相关代码如下// 如果键不存在则设置成功如果键已经存在则返回失败 SETNX datalock VALUE // 设置键过期的时间 PEXPIRE datalock1500原理解析首先尝试在Redis中创建一个字符串结构的缓存该方法传入了键为业务标识和值此处可为任意值​非阻塞锁后为传入的过期时间(timeout)。若Redis中没有键则创建成功抢到锁​然后立即返回true。若已经有键则立即返回false没有抢到锁​。以上过程为全局单线程原子操作即整个过程为独占式操作。为了防止当前操作的Redis客户端宕机而出现锁一直不释放的可能可以设置过期时间。当超过这个过期时间之后键会被Redis服务自动清除。在正常情况下当拿到锁等待业务操作完成之后当前客户端会删除当前键释放锁以供给其他客户端使用。改进的代码如下var minute08;using(var clientnewRedisClient(127.0.0.1,6379)){// 设置库存数量为10client.Setint(inventoryNum,10);// 设置订单数为0client.Setint(orderNum,0);}// 启动10个线程去抢购Console.WriteLine$在{minute}分0秒正式开启秒杀;var flagtrue;while(flag){if(DateTime.Now.Minuteminute){flagfalse;// 模拟并发操作Parallel.For(0,30,(i){inttempi;// 启动子线程Task.Run((){using(var clientnewRedisClient(127.0.0.1,6379)){// 设置锁过期时间var timeoutTimeSpan.FromSeconds(10);// 获取锁如果key:DataLock不存在则返回true// 否则返回falsebool isLockedclient.Addstring(DataLock秒杀,,timeout);// 如果拿到锁则可以进行业务处理否则抢购失败if(isLocked){var inventoryclient.Getint(inventoryNum);if(inventory0){// 库存减1client.Setint(inventoryNum,inventory-1);// 订单数量加1var orderNumclient.Incr(orderNum);Console.WriteLine($抢购成功*****线程 id:{Thread.CurrentThread.ManagedThreadId.ToString(00)},库存{inventory},订单数量{orderNum});// 执行完成删除键释放锁给客户端使用client.Remove(DataLock);}else{Console.WriteLine抢购失败;}}else{Console.WriteLine没有抢到锁;}}});});}}上面代码的执行结果如图13-4所示使用非阻塞锁模拟多个客户端同时进行秒杀抢购也没有出现超卖情况。在图中库存没有因被同时抢购出现超卖其原因是同一时刻去抢锁的时候锁已经被一个线程抢到而且该线程的业务在短时间内没有执行完造成了其他线程没有拿到锁而直接返回false了这种没有抢到锁就立刻返回结果的锁称为非阻塞锁。3、阻塞锁的实现和原理在客户端拿锁却没有抢到锁的时候可以重试多等待一会等上一个客户端执行完业务并释放锁之后再一次去抢锁。如果在等待时间内抢到了就返回true如果没有抢到就再返回false这种锁就是阻塞锁。实现代码如下while (flag) { if (DateTime.Now.Minute minute) { flag false; // 模拟并发操作 Parallel.For(0, 30, (i) { int temp i; // 启动子线程 Task.Run(() { // 此处应该使用try catch防止重试抢不到锁而抛出异常 using (var client new RedisClient(127.0.0.1, 6379)) { // 这个时间代表两个意思一个是锁过期时间 // 另一个是抢不到锁需要等待的时间 var timeout TimeSpan.FromSeconds(10); // 阻塞方式抢锁 using (var datalock client.AcquireLock (BlockDataLock , timeout)) { // 此处省略拿到锁之后需要执行的业务代码 } } } ) ; // Thread.Sleep(100); }); } }上面代码的执行结果如图所示使用阻塞锁模拟多个客户端同时进行秒杀而没有出现超卖情况。需要注意的是非阻塞锁是否可以支持到把所有库存卖完这与业务执行时间和等待抢锁的时间有关但是最后的结果肯定是订单数量小于等于库存数量。接下来我们解析一下AcquireLock阻塞锁的源代码。AcquireLock源代码如下using System;using ServiceStack.Common;using ServiceStack.Text;namespace ServiceStack.Redis{public class RedisLock:IDisposable{private readonly IRedisClient redisClient;private readonly string key;publicRedisLock(IRedisClient redisClient,string key,TimeSpan?timeOut){this.redisClientredisClient;this.keykey;// 等待超时(timeOut)时间如果没有抢到就继续在此尝试抢锁ExecUtils.RetryUntilTrue((){// Calculate a unix time for when the lock should expire// 如果提供过期时间默认就创建一个过期时间比如一年var realSpantimeOut??newTimeSpan(365,0,0,0);// if nothing is passed in the timeout hold for a yearvar expireTimeDateTime.UtcNow.Add(realSpan);var lockString(expireTime.ToUnixTimeMs()1).ToString();// Try to set the lock, if it does not exist this will// succeed and the lock is obtained// 拿锁如果键不存在则直接写入拿到锁var nxredisClient.SetValueIfNotExists(key,lockString);if(nx)returntrue;// If weve gotten here then a key for the lock is// present. This could be because the lock is// correctly acquired or it could be because a client// that had acquired the lock crashed (or didnt release// it properly).// Therefore we need to get the value of the lock to// see when it should expire// 先监听当前这个字段的版本号为了下一次使用redisClient.Watch(key);var lockExpireStringredisClient.Getstring(key);// 时间格式--if(!long.TryParse(lockExpireString,out var lockExpireTime)){// 取消监听// since the client is scoped externallyredisClient.UnWatch();returnfalse;}// 锁没有失效if(lockExpireTimeDateTime.UtcNow.ToUnixTimeMs()){// 取消监听// since the client is scoped externallyredisClient.UnWatch();returnfalse;}// 锁过期// we started the Watch above; this tx will succeed// if the value has not movedusing(var transredisClient.CreateTransaction()){// 抢锁带着版本号去写如果版本号相同则执行成功// 如果版本号不相同则抢锁失败trans.QueueCommand(rr.Set(key,lockString));returntrans.Commit();// returns false ifTransaction failed}},timeOut);}publicvoidDispose(){redisClient.Remove(key);}}}通过上面的源代码可以看出客户端在Redis中创建一个字符串结构的缓存该方法传入了键业务标识和value为锁的过期时间timeout的时间戳​。如果Redis中没有键则创建成功抢到锁​然后立即返回。如果已经有键则先监视然后校验value中的时间戳是否已经超过当前时间。如果已超过则尝试使用提交事务的方式覆盖新的时间戳事务提交成功抢到锁​然后立即返回。若未超过当前时间或事务提交失败被别人抢到锁​则进入一个微循环不断重试抢锁。传入的timeout还有一个作用就是控制重试时间重试超时后则抛出异常。using方法体执行完成之后或者直接显式调用dispose都会直接清除键。需要注意的是传入timeout有两个意思一是成功加锁后锁的过期时间二是未成功加锁后阻塞等待的时间。数据锁服务通过检查value中的时间戳来判断是否过期并不是利用Redis在键上设置过期时间(expire time)来实现的。4、Redlock原理上面的两种锁存在两个明显的问题第一个是单机问题当单机宕机之后当前所有依赖Redis锁的业务都变得不可用了因为无法进行抢锁操作第二个问题就是出现下面的状况会出现锁失效。当客户端1抢到了锁然后锁过期时间是5秒当客户端1执行业务时延迟时间超过5秒当客户端2去抢锁的时候发现前次锁的过期时间已到或者失效了客户端2抢到锁。而后客户端1的业务执行完毕删除锁此时删除的锁其实是客户端2的锁不是上次客户端1自己抢到的锁。接下来客户端3、客户端4都是这样的问题进而造成多个客户端同时操作一份锁资源如图所示。4.1、什么是RedlockRedis官方网站的文章提出了一种基于Redis实现分布式锁的方式——Redlock算法此种方式比原先的单节点方法更安全可以保证以下特性安全特性互斥。在任何给定时刻只有一个客户端可以持有锁。高可用性无死锁。即使持有锁资源的客户端崩溃或分区了也始终可以获得锁。容错性容错能力。只要大多数Redis节点都处于运行状态客户端就可以获取和释放锁。4.2、单节点上实现Redlock单节点上实现Redlock的命令如下SET resource_name my_random_value NX PX30000该命令仅当键不存在时NX保证来设置值设置过期时间为3000毫秒PX保证​其中my_random_value的值必须是所有客户端和所有锁请求发生期间唯一的值释放锁的程序逻辑如下ifredis.call(get,KEYS[1])ARGV[1]thenreturnredis.call(del,KEYS[1])elsereturn0end这个Lua脚本可以避免错误地释放掉另一个客户端创建的锁。如果只有del命令那么当客户端1拿到锁lock1之后因为某些操作而阻塞了很长时间而Redis端的lock1已经过期了并且已经被重新分配给客户端2那么客户端1再去释放这把锁就会造成客户端2原本获得的锁被客户端1无故释放。现在为每个客户端都分配了一个唯一的string值就可以避免这个问题的发生。4.3、Redlock算法假设我们有N个Reids服务分布在不同的机房以尽量保证可用性。客户端会进行如下操作来获取锁它以毫秒为单位获取当前时间。它尝试在所有N个实例中依次使用各个实例中相同的键名和随机值来获取锁。在第2步中在每个实例中设置锁时客户端使用的超时时间(timeout)小于锁自动释放的时间。比如自动释放的时间为10秒超时时间可能在5到50毫秒之间这样可以防止客户端长时间与处于故障状态的Redis节点进行通信如果某个实例不可用那么我们应尝试与下一个实例尽快进行通信。客户端通过从当前时间中减去在步骤1中获得的时间戳来计算获取锁所花费的时间。当且仅当客户端能够在大多数实例至少3个中获取锁时获取锁所花费的总时间小于锁有效时间则认为已获取锁。如果获取了锁则将其有效时间视为初始有效时间减去经过的时间如步骤3中所计算的。如果客户端由于某种原因无法锁定N/21个实例或有效时间为负数而未能获得该锁就将尝试删除已经成功申请锁的节点的锁数据申请成功锁的主节点上执行释放锁的操作重置状态​。1、失败重试如果一个客户端申请锁失败了那么它需要稍等一会再重试以避免出现多个客户端同时申请锁的情况极端情况是一个客户端需要几乎同时向N个主节点发起锁申请。另外如果客户端申请锁失败了就需要尽快在它曾经申请到锁的主节点上执行unlock操作便于其他客户端获得这把锁避免这些锁要等到过期释放而造成时间的浪费。2、释放锁释放锁的操作很简单就是依次释放所有节点上的锁。3、可能出现的问题使用Redis作为锁服务器需要满足高性能因为许多用户在获取和释放锁的延迟以及每秒可能执行的获取或者释放锁操作等方面都需要高性能。所以与N个Redis服务器进行通信以减少延迟的策略肯定是多路复用。对于能从崩溃中恢复的系统还需要考虑持久化。假设我们没有持久性地配置Redis假如客户端已从5个实例或服务节点中的3个获取了锁而客户端要获取锁的一个实例被重新启动了之后我们要为这个实例再次锁定3个实例而另一个客户端也可以再次锁定它这就违反了锁的排他性的安全性。如果启用AOF持久化那么情况将会大大改善。当出现了停电等不可控的情况呢如果默认情况下将Redis配置为每秒在磁盘上执行fsync操作则重启后键可能丢失了。从理论上讲如果要保证在遇到任何形式的实例重新启动时保证锁的安全那么我们需要在持久化设置中始终启用fsyncalways。不过这样一来将完全牺牲了性能。解决这个问题的方法是当一个节点重启之后我们规定在max TTL实例崩溃时所有与锁有关的键存活的时间期间它是不可用的这样就不会干扰原本已经申请到的锁。以C#实现的源代码如下// 需要依赖于StackExchange.Redis // lock 结构类型 public class Lock { public Lock(RedisKey resource, RedisValue val, TimeSpan validity) { this.resource resource; this.val val ; this.validity_time validity; } private RedisKey resource; private RedisValue val; private TimeSpan validity_time; public RedisKey Resource { get { return resource; } } public RedisValue Value { get { return val; } } public TimeSpan Validity { get { return validity_time; } } } // Redlock的实现 public class Redlock { // 连接多个节点的Redis服务 public Redlock(params IConnectionMultiplexer[] list) { foreach(var item in list) this.redisMasterDictionary.Add(item.GetEndPoints().First(). ToString(),item); } // 默认重试次数 const int DefaultRetryCount 3; // 默认每次重试等待时间为200毫秒 readonly TimeSpan DefaultRetryDelay new TimeSpan(0, 0, 0, 0, 200); // 时钟驱动因子 const double ClockDriveFactor 0.01; // 节点的大多数 protected int Quorum { get { return (redisMasterDictionary.Count / 2) 1; } } // / summary // /释放锁的脚本 // / /summary const String UnlockScript if redis.call(get,KEYS[1]) ARGV[1] then return redis.call(del,KEYS[1]) else return 0 end; // 客户端抢锁的唯一标识全局唯一 protected static byte[] CreateUniqueLockId() { return Guid.NewGuid().ToByteArray(); } protected DictionaryString,IConnectionMultiplexer redisMasterDictionary new Dictionarystring, IConnectionMultiplexer(); // 构造连接Redis的实例 protected bool LockInstance(string redisServer, string resource, byte[] val, TimeSpan ttl) { bool succeeded; try { var redis this.redisMasterDictionary[redisServer]; // 写入键如果键不存在就返回true否则返回false // 键为资源值是客户端抢锁生成的唯一字符串(string) // 并且设置了过期时间 succeeded redis.GetDatabase().StringSet(resource, val, ttl, When.NotExists); } catch (Exception) { succeeded false; } return succeeded; } // 释放锁的方法执行释放锁的Lua脚本在每一个服务器上都执行 protected void UnlockInstance(string redisServer, string resource, byte[] val) { RedisKey[] key { resource }; RedisValue[] values { val }; var redis redisMasterDictionary[redisServer]; redis.GetDatabase().ScriptEvaluate( UnlockScript, key, values ); } // 抢锁如果成功就返回true否则返回失败 public bool Lock(RedisKey resource, TimeSpan ttl, out Lock lockObject) { // 生成抢锁前的唯一字符串 var val CreateUniqueLockId(); Lock innerLock null; // 重试多次抢锁 bool successfull retry(DefaultRetryCount, DefaultRetryDelay, () { try { int n 0; var startTime DateTime.Now; // 抢锁 for_each_redis_registered( redis { if (LockInstance(redis, resource, val, ttl)) n 1; } ); /* * Add 2 milliseconds to the drift to account for Redis expires * precision, which is 1 millisecond, plus 1 millisecond min drift * for small TTLs. */ var drift Convert.ToInt32((ttl.TotalMilliseconds * ClockDriveFactor) 2); // 计算抢到锁所花费的时间和设置的过期时间 var validity_time ttl - (DateTime.Now - startTime) - new TimeSpan(0, 0, 0, 0, drift); // 如果大部分节点抢到了锁而且抢锁所花费的时间小于设置的锁过期时间 if (n Quorumvalidity_time.TotalMilliseconds 0) { // 生成锁返回true innerLock new Lock(resource, val, validity_time); return true; } else { // 抢锁失败要么大部分节点没有抢到锁要么抢锁的时间超过了 // 设置的锁过期时间 for_each_redis_registered( redis { // 从所有节点执行删除锁的Lua脚本 UnlockInstance(redis, resource, val); } ); return false; } } catch (Exception) { return false; } }); lockObject innerLock; return successfull; } // 遍历所有节点返回节点连接 protected void for_each_redis_registered(Action IConnectionMultiplexer action) { foreach (var item in redisMasterDictionary) { action(item.Value); } } // 遍历所有节点返回节点地址信息 protected void for_each_redis_registered(ActionString action) { foreach (var item in redisMasterDictionary) { action(item.Key); } } // 重试方法 protected bool retry(int retryCount, TimeSpan retryDelay, Funcbool action) { int maxRetryDelay (int)retryDelay.TotalMilliseconds; Random rnd new Random(); int currentRetry 0; while (currentRetry retryCount) { if (action()) return true; Thread.Sleep(rnd.Next(maxRetryDelay)); } return false; } // 删除锁对象 public void Unlock(Lock lockObject) { for_each_redis_registered(redis { UnlockInstance(redis, lockObject.Resource, lockObject.Value); }); } }调用方式如下// 基于3台Redis服务器声明分布式锁 var dlm new Redlock( ConnectionMultiplexer.Connect(127.0.0.1:6379), ConnectionMultiplexer.Connect(127.0.0.1:6380), ConnectionMultiplexer.Connect(127.0.0.1:6381) ); // 声明锁对象 Lock lockObject; // 尝试获取锁使用resourceName作为锁标识符有效期为10秒 var locked dlm.Lock( resourceName, new TimeSpan(0, 0, 10), out lockObject ); // 如果locked为true则会填充lockObject并获取锁 // 否则无法获取锁 If(locked ) { // 此处进行业务处理 // 释放lockObject中包含的锁 dlm.Unlock(lockObject); }