前言

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

几分钟搞定redis存储session共享——设计实现：https://www.geek-share.com/detail/2759732180.html

【原创】详细案例解剖——浅谈Redis缓存的常用5种方式（String，Hash，List，set，SetSorted ）：https://www.geek-share.com/detail/2758456280.html

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

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

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

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

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

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

设计（悲观锁/乐观锁）

悲观锁方式（认为操作的时候，会出现问题，所以都加锁）

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

　　两种锁各有优缺点，不可认为一种好于另一种，像乐观锁适用于写比较少的情况下，即冲突真的很少发生的时候，这样可以省去了锁的开销，
加大了系统的整个吞吐量。但如果经常产生冲突，上层应用会不断的进行retry，这样反倒是降低了性能，所以这种情况下用悲观锁就比较合适。

Redis三个命令

1、SETNX

2、expire

3、delete

在使用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方法，并没有提供设置过期时间的方法，对于加锁业务又不能分开执行(如果加锁成功设置过期时间失败导致的永久死锁问题)，所以就使用脚本实现，解决了异常情况死锁问题.

释放锁

　　　　 /// <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分布锁的功能，快去试试吧。

参考文献

• Redis如何实现分布式锁：https://www.geek-share.com/detail/2840417000.html
• 浅谈Redis缓存的常用5种方式（String，Hash，List，set，SetSorted ）：https://www.geek-share.com/detail/2758456280.html
• 几分钟搞定redis存储session共享——设计实现：https://www.geek-share.com/detail/2759732180.html