2025年5月20日 星期二 乙巳(蛇)年 二月廿二 设为首页 加入收藏
rss
您当前的位置:首页 > 计算机 > 软件应用 > 数据库 > Redis

Redis分布式锁的原理和实现

时间:03-16来源:作者:点击数:50

前言

我们之前聊过redis的,对基础不了解的可以移步查看一下:

几分钟搞定redis存储session共享——设计实现:https://www.cdsy.xyz/computer/soft/database/redis/230316/cd41544.html

【原创】详细案例解剖——浅谈Redis缓存的常用5种方式(String,Hash,List,set,SetSorted ):https://www.cdsy.xyz/computer/soft/database/redis/230316/cd41543.html

对同一个资源进行操作,单一的缓存读取没问题了,但是存在并发的时候怎么办呢,为了避免数据不一致,我们需要在操作共享资源之前进行加锁操作。

我们在开发很多业务场景会使用到锁,例如库存控制,抽奖,秒杀等。一般我们会使用内存锁的方式来保证线性的执行。

但现在大多站点都会使用分布式部署,那多台服务器间的就必须使用同一个目标来判断锁。分布式与单机情况下最大的不同在于其不是多线程而是多进程。


图1:分布式站点使用内存锁

图2:分布式站点使用分布式锁

当然我们暂时用不了这么复杂的场景,我们就简单访问redis就行。


设计(悲观锁/乐观锁)

悲观锁方式(认为操作的时候,会出现问题,所以都加锁)

悲观锁(Pessimistic Lock), 顾名思义,就是很悲观,每次去拿数据的时候都认为别人会修改,

所以每次在拿数据的时候都会上锁,这样别人想拿这个数据就会block直到它拿到锁。

传统的关系型数据库里边就用到了很多这种锁机制,比如行锁,表锁等,读锁,写锁等,都是在做操作之前先上锁。

乐观锁方式(认为什么时候不会出问题,所以不上锁,更新的时候去查询判断一下,再此期间是否有人修改过这个数据。)

乐观锁(Optimistic Lock), 顾名思义,就是很乐观,每次去拿数据的时候都认为别人不会修改,

所以不会上锁,但是在更新的时候会判断一下在此期间别人有没有去更新这个数据,可以使用版本号等机制。

乐观锁适用于多读的应用类型,这样可以提高吞吐量,像数据库如果提供类似于write_condition机制的其实都是提供的乐观锁。

  两种锁各有优缺点,不可认为一种好于另一种,像乐观锁适用于写比较少的情况下,即冲突真的很少发生的时候,这样可以省去了锁的开销,

加大了系统的整个吞吐量。但如果经常产生冲突,上层应用会不断的进行retry,这样反倒是降低了性能,所以这种情况下用悲观锁就比较合适。


Redis三个命令

1、SETNX

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

2、expire

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

3、delete

delete key:删除key

在使用Redis实现分布式锁的时候,主要就会使用到这三个命令。

命题:某商品进行库存秒杀。

假设要给某个商品举行秒杀活动,我们事先把库存数据100已经存入到了redis中,我们现在需要来进行库存扣减。

            图3:加锁请求示意图


代码实现

我们基于ServiceStack.Redis操作

我们创建一个控制台应用(.NET Framework),命名为RedisLock,注意,如果创建的是net core的应用,引入的ServiceStack.Redis就要选择core的。

然后在NuGet里面安装ServiceStack.Redis。

Redis连接池

  • //Redis连接池(配置连接地址,读写连接地址等)
  • public static PooledRedisClientManager RedisClientPool = CreateManager();
  • private static PooledRedisClientManager CreateManager()
  • {
  • //写节点(主节点)
  • List<string> writes = new List<string>();
  • writes.Add("10.17.3.97:6379");
  • //读节点
  • List<string> reads = new List<string>();
  • reads.Add("10.17.3.97:6379");
  • //配置连接池和读写分类
  • return new PooledRedisClientManager(writes, reads, new RedisClientManagerConfig()
  • {
  • MaxReadPoolSize = 50, //读节点个数
  • MaxWritePoolSize = 50,//写节点个数
  • AutoStart = true,
  • DefaultDb = 0
  • });
  • }

使用Redis的SetNX命令实现加锁

  • /// <summary>
  • /// 加锁(使用Redis的SetNX命令实现加锁)
  • /// </summary>
  • /// <param name="key">锁key</param>
  • /// <param name="selfMark">自己标记</param>
  • /// <param name="lockExpirySeconds">锁自动过期时间[默认10](s)</param>
  • /// <param name="waitLockMilliseconds">等待锁时间(ms)</param>
  • /// <returns></returns>
  • public static bool Lock(string key, out string selfMark, int lockExpirySeconds = 10, long waitLockMilliseconds = long.MaxValue)
  • {
  • DateTime begin = DateTime.Now;
  • selfMark = Guid.NewGuid().ToString("N");//自己标记,释放锁时会用到,自己加的锁除非过期否则只能自己打开
  • using (RedisClient redisClient = (RedisClient)RedisClientPool.GetClient())
  • {
  • string lockKey = "Lock:" + key;
  • while (true)
  • {
  • string script = string.Format("if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then redis.call('PEXPIRE',KEYS[1],{0}) return 1 else return 0 end", lockExpirySeconds * 1000);
  • //循环获取取锁
  • if (redisClient.ExecLuaAsInt(script, new[] { lockKey }, new[] { selfMark }) == 1)
  • {
  • return true;
  • }
  • //不等待锁则返回
  • if (waitLockMilliseconds == 0)
  • {
  • break;
  • }
  • //超过等待时间,则不再等待
  • if ((DateTime.Now - begin).TotalMilliseconds >= waitLockMilliseconds)
  • {
  • break;
  • }
  • Thread.Sleep(100);
  • }
  • return false;
  • }
  • }

因为ServiceStack.Redis提供的SetNX方法,并没有提供设置过期时间的方法,对于加锁业务又不能分开执行(如果加锁成功设置过期时间失败导致的永久死锁问题),所以就使用脚本实现,解决了异常情况死锁问题.

如果设置为0,为乐观锁机制,获取不到锁,直接返回未获取到锁.

默认值为long最大值,为悲观锁机制,约等于很多很多天,可以理解为一直等待.

释放锁

  •      /// <summary>
  • /// 释放锁
  • /// </summary>
  • /// <param name="key">锁key</param>
  • /// <param name="selfMark">自己标记</param>
  • public static void UnLock(string key, string selfMark)
  • {
  • using (RedisClient redisClient = (RedisClient)RedisClientPool.GetClient())
  • {
  • string lockKey = "Lock:" + key;
  • var script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
  • redisClient.ExecLuaAsString(script, new[] { lockKey }, new[] { selfMark });
  • }
  • }

业务调用(我们使用多线程模拟多用户秒杀的场景)

  • //业务:悲观锁方式
  • public static void PessimisticLock()
  • {
  • int num = 10; //总数量
  • string lockkey = "xianseng";
  • Console.WriteLine("我是【悲观锁】");
  • //悲观锁开启20个人同时拿宝贝
  • for (int i = 0; i < 20; i++)
  • {
  • Task.Run(() =>
  • {
  • string selfmark = "";
  • try
  • {
  • if (Lock(lockkey, out selfmark))
  • {
  • if (num > 0)
  • {
  • num--;
  • Console.WriteLine($"我拿到了宝贝:宝贝剩余{num}个\t\t{selfmark}");
  • }
  • else
  • {
  • Console.WriteLine("宝贝已经没有了");
  • }
  • Thread.Sleep(100);
  • }
  • }
  • finally
  • {
  • UnLock(lockkey, selfmark);
  • }
  • });
  • }
  • Console.ReadLine();
  • }
  • //业务:乐观锁方式
  • public static void OptimisticLock()
  • {
  • int num = 10; //总数量
  • string lockkey = "xianseng";
  • Console.WriteLine("我是【乐观锁】");
  • //乐观锁开启10个线程,每个线程拿5次
  • for (int i = 0; i < 10; i++)
  • {
  • var lineOn = "线程" + (i + 1);
  • Task.Run(() =>
  • {
  • for (int j = 0; j < 5; j++)
  • {
  • string selfmark = "";
  • try
  • {
  • if (Lock(lockkey, out selfmark, 10, 0))
  • {
  • if (num > 0)
  • {
  • num--;
  • Console.WriteLine($"{lineOn}{(j+1)}次 我拿到了宝贝:宝贝剩余{num}个\t\t{selfmark}");
  • }
  • else
  • {
  • Console.WriteLine($"{lineOn}{(j + 1)}次 宝贝已经没有了");
  • }
  • Thread.Sleep(1000);
  • }
  • else
  • {
  • Console.WriteLine($"{lineOn}{(j+1)}次 没有拿到,不想等了");
  • }
  • }
  • finally
  • {
  • UnLock(lockkey, selfmark);
  • }
  • }
  • });
  • }
  • Console.ReadLine();
  • }

然后在main函数里面调用查看展示效果

  • static void Main(string[] args)
  • {
  • ////调用:悲观锁方式(认为我操作的时候,会出现问题,所以都加锁)
  • PessimisticLock();
  • ///调用:乐观锁方式(认为什么时候不会出问题,所以不上锁,更新的时候去查询判断一下,再此期间是否有人修改过这个数据。)
  • //OptimisticLock();
  • }

这就简单实现了Redis分布锁的功能,快去试试吧。

方便获取更多学习、工作、生活信息请关注本站微信公众号城东书院 微信服务号城东书院 微信订阅号
推荐内容
相关内容
栏目更新
栏目热门