admin管理员组

文章数量:817344

Redis分布式锁剖析和几种客户端的实现

1. 背景

 在传统的单体项目中,即部署到单个IIS上,针对并发问题,比如进销存中的出库和入库问题,多个人同时操作,属于一个IIS进程中多个线程并发操作的问题,这个时候可以引入线程锁lock/Monitor等,轻松解决这类问题。但是随着业务量的逐渐增大,比如"秒杀业务", 肯定是集群部署,这个时候线程锁已经没用了, 必须引入分布式锁。

 常见的分布式锁有:数据库、zookeeper、redis. 本节重点介绍redis的分布式锁.

如下图:

2. 分布式锁需要满足的条件

 (1).在分布式系统环境下,一个锁在同一时间只能被一个服务器获取;(这是所有分布式锁的基础)

 (2).高性能的获取锁和释放锁;(锁用完了,要及时释放,以供别人继续使用)

 (3).具备锁失效机制,防止死锁;(防止因为某些意外,锁没有得到释放,那别人将永远无法使用)

 (4).具备非阻塞锁特性,即没有获取到锁将直接返回获取锁失败。(满足等待锁的同时,也要满足非阻塞锁特性,便于多样性的业务场景使用)

3. 分布式锁种类/原理

(1).阻塞锁

  尝试在redis中创建一个字符串结构缓存,方法传入key和过期时间(AcquireLock), 其中key对应的value为锁的过期时间timeout的时间戳

  若redis中没有这个key,则创建成功(即抢到锁),然后立即返回;若已经有这个key,则先watch,然后校验value中的时间戳是否已经超过当前时间

  若已超过,则尝试使用提交事务的方式覆盖新的时间戳,事务提交成功(即抢到锁),然后立即返回;若未超过当前时间或事务提交失败(即被别人抢到锁),则进入一个内部优化过的微循环,不断重试。

 传入的timeout还有一个作用,就是控制重试时间,重试超时后则抛异常,using完成方法调用或者显式调用dispose,都会直接清除key。

总结:

  timeout有两个意思:一是如果成功加锁后锁的过期时间, 二是未成功加锁后阻塞等待的时间。数据锁服务通过检查value中时间戳来判断是否过期,并不是利用redis在key上设置expire time来通过key的过期实现的

(2).非阻塞锁

  尝试在redis中创建一个字符串结构缓存项,方法传入key、value、timeout(Add),其中value无实际意义,过期时间为传入的timeout。

  若redis中没有这个key,则创建成功(即抢到锁),然后立即返回true.若已经有这个key,则立即返回false。以上过程为全局单线程原子操作,整个过程为独占式操作。IsLock可以检测key是否存在。

注意:

  timeout即成功加锁后锁的过期时间,利用redis在key上设置expire time来通过key的过期实现。不要先用IsLock判断是否有锁再用Add加锁,因为这两个操作非原子性操作,期间会被其他操作干扰。

(3).底层实现主要用到以下几个指令

A.setnx

 setnx key val:当且仅当key不存在时,set一个key为val的字符串,返回1;若key存在,则什么都不做,返回0

B.expire

 expire key timeout:为key设置一个超时时间,单位为second,超过这个时间锁会自动释放,避免死锁

C.delete

 delete key:删除key

二. 案例模拟实现

1.场景模拟分析

 模拟多个用户进行秒杀业务,扣减库存→创建订单。 (PS:这里只是为了演示分布式锁而已,实际场景可以利用redis自减Api原子性实现扣减库存,从而干掉锁的问题)

总结:真正的秒杀是不会用分布式锁的, 因为用锁会存在等待的问题,会产生大量无响应的情况, 实际情况下可以利用Lua脚本结合redis原子性的特点,编写秒杀业务。详见:.html

下面分享3个不同的程序集实现分布式锁的业务.

C/C++Linux服务器开发高级架构师/C++后台开发架构师​免费学习地址

【文章福利】另外还整理一些C++后台开发架构师 相关学习资料,面试题,教学视频,以及学习路线图,免费分享有需要的可以点击领取

2. ServiceStack.Redis

(1).阻塞锁

代码分享

   /// <summary>/// 阻塞锁/// </summary>public class BlockingLock{public static void Show(int i, string key, TimeSpan timeout){using var client = new RedisClient("119.45.174.249", 6379, "123456");using (var myLock = client.AcquireLock(key, timeout))  //获取锁  (此处阻塞,其它线程等待){var goodNum = client.Get<int>("goodNum");if (goodNum > 0){client.Set<int>("goodNum", goodNum - 1);  //扣减库存var orderNum = client.Incr("orderNum");Console.WriteLine($"{i}抢购成功,此时的库存为{goodNum - 1},订单数量为:{orderNum}");}else{Console.WriteLine($"商品已经卖光了");}}}}

调用

 static void Main(string[] args){Console.WriteLine("请输入开始抢购的时间:");int minute = int.Parse(Console.ReadLine());using var client = new RedisClient("119.45.174.249", 6379, "123456");//商品数量(设置为10)client.Set<int>("goodNum", 10);//订单数(默认为0)client.Set<int>("orderNum", 0);//开启30个线程去抢购Console.WriteLine($"在{minute}分0秒正式开启秒杀!");var flag = true;while (flag){if (DateTime.Now.Minute == minute)//if (true){flag = false;Parallel.For(0, 30, (i) =>{int temp = i;Task.Run(() =>{BlockingLock.Show(i, "akey", TimeSpan.FromSeconds(100));   //阻塞锁//NoBlockingLock.Show(i, "akey", TimeSpan.FromSeconds(100));   //非阻塞锁});});}}Console.ReadKey();}

copy两套程序同时运行

(2).非阻塞锁

代码分享:

  /// <summary>/// 非阻塞锁/// </summary>public class NoBlockingLock{public static void Show(int i, string key, TimeSpan timeout){using var client = new RedisClient("119.45.174.249", 6379, "123456");bool isLocked = client.Add<string>(key, "xxxx", timeout);if (isLocked){try{var goodNum = client.Get<int>("goodNum");if (goodNum > 0){client.Set<int>("goodNum", goodNum - 1);  //扣减库存var orderNum = client.Incr("orderNum");   //订单数量自增1Console.WriteLine($"{i}抢购成功,此时的库存为{goodNum - 1},订单数量为:{orderNum}");}else{Console.WriteLine($"{i}商品已经卖光了");}}catch (Exception ex){Console.WriteLine($"{i}报错了{ex.Message}");}finally{client.Remove(key);}}else{Console.WriteLine($"{i}抢购失败:原因:没有拿到锁");}}}

调用

 static void Main(string[] args){Console.WriteLine("请输入开始抢购的时间:");int minute = int.Parse(Console.ReadLine());using var client = new RedisClient("119.45.174.249", 6379, "123456");//商品数量(设置为10)client.Set<int>("goodNum", 10);//订单数(默认为0)client.Set<int>("orderNum", 0);//开启30个线程去抢购Console.WriteLine($"在{minute}分0秒正式开启秒杀!");var flag = true;while (flag){if (DateTime.Now.Minute == minute)//if (true){flag = false;Parallel.For(0, 30, (i) =>{int temp = i;Task.Run(() =>{//BlockingLock.Show(i, "akey", TimeSpan.FromSeconds(100));   //阻塞锁NoBlockingLock.Show(i, "akey", TimeSpan.FromSeconds(100));   //非阻塞锁});});}}Console.ReadKey();}

copy两套程序同时运行

3. StackExchange.Redis

代码分享

 public class MyLock1{public static void Show(int i, string key, TimeSpan timeout){RedisHelp redis = new RedisHelp("119.45.174.249:6379,password=123456");var client = redis.GetDatabase();bool isLocked = client.LockTake(key, Environment.MachineName, timeout); //timeout秒后自动释放if (isLocked){try{var goodNum = int.Parse(client.StringGet("goodNum"));if (goodNum > 0){client.StringSet("goodNum", goodNum - 1);  //扣减库存var orderNum = client.StringIncrement("orderNum");   //订单数量自增1Console.WriteLine($"{i}抢购成功,此时的库存为{goodNum - 1},订单数量为:{orderNum}");}else{Console.WriteLine($"{i}商品已经卖光了");}}catch (Exception ex){Console.WriteLine($"{i}报错了{ex.Message}");}finally{client.LockRelease(key, Environment.MachineName);}}else{Console.WriteLine($"{i}抢购失败:原因:没有拿到锁");}}}

调用:

static void Main(string[] args)
{Console.WriteLine("请输入开始抢购的时间:");int minute = int.Parse(Console.ReadLine());RedisHelp redis = new RedisHelp("119.45.174.249:6379,password=123456");var client = redis.GetDatabase();//商品数量(设置为10)client.StringSet("goodNum", 10);//订单数(默认为0)client.StringSet("orderNum", 0);//开启30个线程去抢购Console.WriteLine($"在{minute}分0秒正式开启秒杀!");var flag = true;while (flag){if (DateTime.Now.Minute == minute)//if (true){flag = false;Parallel.For(0, 30, (i) =>{int temp = i;Task.Run(() =>{MyLock1.Show(i, "akey", TimeSpan.FromSeconds(2));});});}}Console.ReadKey();
}

4. CSRedisCore

代码分享

public class MyLock1{public static void Show(int i, string key, int timeout){RedisHelper.Initialization(new CSRedis.CSRedisClient("119.45.174.249:6379,password=123456,defaultDatabase=0"));var isLocked = RedisHelper.Lock(key, timeout, true); //timeout秒后自动释放if (isLocked != null)  //获取超时则返回null{try{var goodNum = int.Parse(RedisHelper.Get("goodNum"));if (goodNum > 0){RedisHelper.Set("goodNum", goodNum - 1);  //扣减库存var orderNum = RedisHelper.IncrBy("orderNum");   //订单数量自增1Console.WriteLine($"{i}抢购成功,此时的库存为{goodNum - 1},订单数量为:{orderNum}");}else{Console.WriteLine($"商品已经卖光了");}}catch (Exception ex){Console.WriteLine($"报错了{ex.Message}");}finally{RedisHelper.Del(key);  //上面可以自动删除,还需要手动删除吗?}}else{Console.WriteLine($"{i}抢购失败:原因:没有拿到锁");}}}

调用

  static void Main(string[] args){Console.WriteLine("请输入开始抢购的时间:");int minute = int.Parse(Console.ReadLine());var client = new CSRedis.CSRedisClient("119.45.174.249:6379,password=123456,defaultDatabase=0");//商品数量(设置为10)client.Set("goodNum", 10);//订单数(默认为0)client.Set("orderNum", 0);//开启30个线程去抢购Console.WriteLine($"在{minute}分0秒正式开启秒杀!");var flag = true;while (flag){if (DateTime.Now.Minute == minute)//if (true){flag = false;Parallel.For(0, 30, (i) =>{int temp = i;Task.Run(() =>{MyLock1.Show(i, "akey", 2);});});}}Console.ReadKey();}

原文链接:第十五节:Redis分布式锁剖析和几种客户端的实现 - Yaopengfei - 博客园

本文标签: Redis分布式锁剖析和几种客户端的实现