基于Redis的分布式锁设计

指示材料库存的Key为

ProductStock_10000

• 自己实现的分布式锁中，指示锁的Key为

  DistributedLock_10000

• RedLock.net中，指示锁的Key为

  redlock:10000

假如没有锁

如果没有锁，我们可以通过Jmeter并发100个请求，看看最后库存是不是0

/// <summary>
/// 无锁扣减库存
/// </summary>
/// <returns></returns>
[HttpPost("DecreaseProductStockWithNoLock")]
public async Task<string> DecreaseProductStockWithNoLock()
{
var stockKey = GetProductStockKey(ProductId);
var currentQuantity = (long)(await _redisDatabase.Database.StringGetAsync(stockKey));
if (currentQuantity < 1)
throw new Exception("库存不足");

var leftQuantity = currentQuantity - 1;
await _redisDatabase.Database.StringSetAsync(stockKey, leftQuantity);

return $"剩余库存：{leftQuantity}";
}

完了，库存全乱了，收拾收拾，跑路吧o(╥﹏╥)o！

单应用中的锁

提到锁，大多数人首先想到的应该就是

Monitor

lock

/// <summary>
/// 在单应用中扣减库存
/// </summary>
/// <returns></returns>
[HttpPost("DecreaseProductStockInSingleApp")]
public string DecreaseProductStockInSingleApp()
{
long leftQuantity;
lock (_lockObj)
{
var stockKey = GetProductStockKey(ProductId);
var currentQuantity = (long)_redisDatabase.Database.StringGet(stockKey);
if (currentQuantity < 1)
throw new Exception("库存不足");

leftQuantity = currentQuantity - 1;
_redisDatabase.Database.StringSet(stockKey, leftQuantity);
}

return $"剩余库存：{leftQuantity}";
}

结果和我们所期望的一样，剩余库存为0

但是如果我们进行应用集群，部署多份一模一样的应用，那

lock

# 以开发环境运行，能看到更多信息
dotnet XXTk.Redis.DistributedLock.Api.dll --urls http://localhost:5000 --environment Development

dotnet XXTk.Redis.DistributedLock.Api.dll --urls http://localhost:5010 --environment Development

可见，一共发送了100个请求，本应该最后库存为0的，却还剩17个

应用集群中的锁

版本1

很明显，

lock

初步的思路是这样的：

1. 将材料Id作为Redis Key
2. 如果Redis中存在该Key，则认为锁已经被其他线程占用了
3. 如果Redis中不存在Key，则将该Key添加到Redis中，Value则随意赋值
4. 当获取到锁的业务执行完毕后，将Key从Redis中移除

有了思路，接下来就该想一下如何实现了。很幸运，Redis的命令

SETNX key value

/// <summary>
/// 在应用集群中扣减库存V1
/// </summary>
/// <returns></returns>
[HttpPost("v1/DecreaseProductStockInAppCluster")]
public async Task<string> DecreaseProductStockInAppClusterV1()
{
var lockKey = GetDistributedLockKey(ProductId.ToString());

// 使用 SETNX key value 命令加锁
if (await _redisDatabase.Database.StringSetAsync(lockKey, 1, null, When.NotExists, CommandFlags.DemandMaster))
{
try
{
var stockKey = GetProductStockKey(ProductId);
var currentQuantity = (long)await _redisDatabase.Database.StringGetAsync(stockKey);
if (currentQuantity < 1)
throw new Exception("库存不足");

var leftQuantity = currentQuantity - 1;
await _redisDatabase.Database.StringSetAsync(stockKey, leftQuantity);

return $"剩余库存：{leftQuantity}";
}
finally
{
// 释放锁
await _redisDatabase.Database.KeyDeleteAsync(lockKey, CommandFlags.DemandMaster);
}
}
else
throw new Exception("获取锁失败");
}

通过计算，成功50次，失败50次，而我们查到的库存也是还剩余50个，所以已经基本实现了我们的需求。

版本2

虽然版本1已经基本实现了我们的需求，但是试想一下：

• 代码执行在

  try

  块中时，应用崩溃了，导致锁未被释放
• 释放锁时，由于网络问题，连接Redis失败了，导致锁未被释放

如果发生了以上任何情况，都无法正确的释放锁，导致锁永远无法释放，导致死锁。

那我们应该怎么办呢？对，就是给锁加一个过期时间！不过

SETNX

EXPIRE

这样，可以吗？当然不可以，我们需要将

SET

EXPIRE

SET

SET key value EX seconds NX

NX

SETNX

/// <summary>
/// 在应用集群中扣减库存V2
/// </summary>
/// <returns></returns>
[HttpPost("v2/DecreaseProductStockInAppCluster")]
public async Task<string> DecreaseProductStockInAppClusterV2()
{
var lockKey = GetDistributedLockKey(ProductId.ToString());
var expiresIn = TimeSpan.FromSeconds(30);

// 使用 SET key value EX seconds NX 命令加锁，并设置过期时间
if (await _redisDatabase.AddAsync(lockKey, 1, expiresIn, When.NotExists, CommandFlags.DemandMaster))
{
try
{
var stockKey = GetProductStockKey(ProductId);
var currentQuantity = (long)await _redisDatabase.Database.StringGetAsync(stockKey);
if (currentQuantity < 1)
throw new Exception("库存不足");

var leftQuantity = currentQuantity - 1;
await _redisDatabase.Database.StringSetAsync(stockKey, leftQuantity);

return $"剩余库存：{leftQuantity}";
}
finally
{
// 释放锁
await _redisDatabase.Database.KeyDeleteAsync(lockKey, CommandFlags.DemandMaster);
}
}
else
throw new Exception("获取锁失败");
}

版本3

好，死锁的问题咱们已经解决了，那咱们的分布式锁是不是已经完美了呢？NO！NO！NO！还是有一些问题滴：

1. 如果线程A获取到了锁，并设置了锁的过期时间是30s，而业务的执行时长需要40s，这就出现了锁被提前释放的问题
2. 如果锁被提前释放了，然后被另一个线程B获取到了，此时线程A的业务执行完毕了，然后执行了

   finally

   代码块中的锁释放代码，这就把不属于线程A而属于线程B的锁释放掉了，这下可全乱套了。

是不是感觉越改问题越多？别灰心，咱们一个一个来解决，先来解决第二个“错误释放了不属于自己的锁”的问题。为了让线程知道哪个是自己的锁，我们需要给线程起个唯一不重复的名字，当需要释放锁的时候，先检查一下是不是自己的锁，如果是，才释放锁。那这个名字放在哪里呢？咱们之前LockKey对应的Value不是没有用嘛，那咱们就把名字存这里面，实现如下：

/// <summary>
/// 在应用集群中扣减库存V3
/// </summary>
/// <returns></returns>
[HttpPost("v3/DecreaseProductStockInAppCluster")]
public async Task<string> DecreaseProductStockInAppClusterV3()
{
var lockKey = GetDistributedLockKey(ProductId.ToString());
var resourceId = Guid.NewGuid().ToString();
var expiresIn = TimeSpan.FromSeconds(30);

// 使用 SET key value EX seconds NX 命令加锁，设置过期时间，并将值设置为业务Id
if (await _redisDatabase.AddAsync(lockKey, resourceId, expiresIn, When.NotExists, CommandFlags.DemandMaster))
{
try
{
var stockKey = GetProductStockKey(ProductId);
var currentQuantity = (long)await _redisDatabase.Database.StringGetAsync(stockKey);
if (currentQuantity < 1)
throw new Exception("库存不足");

var leftQuantity = currentQuantity - 1;
await _redisDatabase.Database.StringSetAsync(stockKey, leftQuantity);

return $"剩余库存：{leftQuantity}";
}
finally
{
// 释放锁
if (await _redisDatabase.GetAsync<string>(lockKey) == resourceId)
{
_redisDatabase.Database.KeyDelete(lockKey, CommandFlags.DemandMaster);
}
}
}
else
throw new Exception("获取锁失败");
}

版本4

上面的代码，你应该看出问题了吧？没错，最后的释放锁代码是分两步执行的，并不是原子操作，这肯定是不允许的啦！但是，Redis又没有提供相关命令，所以我们只能使用lua脚本了：

/// <summary>
/// 在应用集群中扣减库存V4
/// </summary>
/// <returns></returns>
[HttpPost("v4/DecreaseProductStockInAppCluster")]
public async Task<string> DecreaseProductStockInAppClusterV4()
{
var lockKey = GetDistributedLockKey(ProductId.ToString());
var resourceId = Guid.NewGuid().ToString();
var expiresIn = TimeSpan.FromSeconds(30);

// 使用 SET key value EX seconds NX 命令加锁，设置过期时间，并将值设置为业务Id
if (await _redisDatabase.AddAsync(lockKey, resourceId, expiresIn, When.NotExists, CommandFlags.DemandMaster))
{
try
{
var stockKey = GetProductStockKey(ProductId);
var currentQuantity = (long)await _redisDatabase.Database.StringGetAsync(stockKey);
if (currentQuantity < 1)
throw new Exception("库存不足");

var leftQuantity = currentQuantity - 1;
await _redisDatabase.Database.StringSetAsync(stockKey, leftQuantity);

return $"剩余库存：{leftQuantity}";
}
finally
{
// 释放锁，使用lua脚本实现操作的原子性
await _redisDatabase.Database.ScriptEvaluateAsync(@"
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end",
keys: new RedisKey[] { lockKey },
values: new RedisValue[] { resourceId },
CommandFlags.DemandMaster);
}
}
else
throw new Exception("获取锁失败");
}

RedisNewtonsoftSerializer

版本5

最后，我们来解决最后一个问题——业务执行时长超过了锁的过期时长，导致锁提前被释放。由于我们无法准确预测业务的执行时长，锁过期时间设置的太长也不合理，所以，若业务还未执行完，我们必须能够在锁快过期的时候，适当的延长锁过期时间。可以通过定时器来解决。

/// <summary>
/// 在应用集群中扣减库存V5
/// </summary>
/// <returns></returns>
[HttpPost("v5/DecreaseProductStockInAppCluster")]
public async Task<string> DecreaseProductStockInAppClusterV5()
{
var lockKey = GetDistributedLockKey(ProductId.ToString());
var resourceId = Guid.NewGuid().ToString();
var expiresIn = TimeSpan.FromSeconds(30);

// 使用 SET key value EX seconds NX 命令加锁，设置过期时间，并将值设置为业务Id
if (await _redisDatabase.AddAsync(lockKey, resourceId, expiresIn, When.NotExists, CommandFlags.DemandMaster))
{
try
{
// 启动定时器，定时延长key的过期时间
var interval = expiresIn.TotalMilliseconds / 2;
var timer = new System.Threading.Timer(
callback: state => ExtendLockLifetime(lockKey, resourceId, expiresIn),
state: null,
dueTime: (int)interval,
period: (int)interval);

var stockKey = GetProductStockKey(ProductId);
var currentQuantity = (long)await _redisDatabase.Database.StringGetAsync(stockKey);
if (currentQuantity < 1)
throw new Exception("库存不足");

var leftQuantity = currentQuantity - 1;
await _redisDatabase.Database.StringSetAsync(stockKey, leftQuantity);

timer.Change(Timeout.Infinite, Timeout.Infinite);
timer.Dispose();
timer = null;

return $"剩余库存：{leftQuantity}";
}
finally
{
// 释放锁，使用lua脚本实现操作的原子性
await _redisDatabase.Database.ScriptEvaluateAsync(@"
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end",
keys: new RedisKey[] { lockKey },
values: new RedisValue[] { resourceId },
CommandFlags.DemandMaster);
}
}
else
throw new Exception("获取锁失败");
}

private void ExtendLockLifetime(string lockKey, string resourceId, TimeSpan expiresIn)
{
_redisDatabase.Database.ScriptEvaluate(@"
local currentVal = redis.call('get', KEYS[1])
if (currentVal == false) then
return redis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2]) and 1 or 0
elseif (currentVal == ARGV[1]) then
return redis.call('pexpire', KEYS[1], ARGV[2])
else
return -1
end
",
keys: new RedisKey[] { lockKey },
values: new RedisValue[] { resourceId, (long)expiresIn.TotalMilliseconds },
CommandFlags.DemandMaster);
}

使用RedLock.net中的分布式锁

以上的版本5，已经包含了分布式锁的基本思想了，不过我写的肯定比较简陋，所以我给大家推荐一个比较不错的开源实现——RedLock.net

/// <summary>
/// 通过使用RedLock在应用集群中扣减库存
/// </summary>
/// <returns></returns>
[HttpPost("DecreaseProductStockInAppClusterWithRedLock")]
public async Task<string> DecreaseProductStockInAppClusterWithRedLock()
{
// 锁的过期时间为30s，等待获取锁的时间为20s，如果没有获取到锁，则等待1秒钟后再次尝试获取
using var redLock = await _distributedLockFactory.CreateLockAsync(
resource: ProductId.ToString(),
expiryTime: TimeSpan.FromSeconds(30),
waitTime: TimeSpan.FromSeconds(20),
retryTime: TimeSpan.FromSeconds(1)
);

// 确认是否已获取到锁
if (redLock.IsAcquired)
{
var stockKey = GetProductStockKey(ProductId);
var currentQuantity = (long)await _redisDatabase.Database.StringGetAsync(stockKey);
if (currentQuantity < 1)
throw new Exception("库存不足");

var leftQuantity = currentQuantity - 1;
await _redisDatabase.Database.StringSetAsync(stockKey, leftQuantity);

return $"剩余库存：{leftQuantity}";
}
else
throw new Exception("获取锁失败");
}

站在Redis角度上

我们上面站在程序的角度上已经实现了分布式锁，但是站在Redis角度上，还有几个问题需要思考一下：

Redis宕机导致无法加锁

如果Redis宕机了，就会导致Redis服务器不可用，从而导致无法进行加锁。

解决方法很简单，可以通过配置主从关系，提高Redis的高可用性，但这样又产生了下面的问题。

Redis主从切换导致锁失效

过程是这样的：

1. 客户端 A 从 Redis master 上获取到了锁
2. 在代表锁的 Key 同步到 Redis slave 之前，master 宕机了
3. 然后 Redis 进行主从切换， Redis slave 升级为 Redis master
4. 客户端 B 从新 Redis master 中获取到了上面客户端 A 持有的锁。

这显然出大问题了！因此，RedLock算法诞生了。

RedLock

基本原理

首先，我们需要至少5台（大于等于5的奇数个）Redis服务器，这5台Redis之间相互独立，没有任何主从、集群关系。

接着，我们按照从左到右的顺序，在Redis服务器上获取锁，我们假设

• 锁的过期时间为10s，
• 加锁的开始时间是00:00:00，
• 在第一台服务器上获取到锁的时间为00:00:01，
• 在第二台服务器上获取到锁的时间为00:00:02，
• 在第三台服务器上获取到锁的时间为00:00:03。

现在，已经有超过半数（3/5）的Redis服务器获取到了锁。

获取锁所用的时间 = 00:00:03 - 00:00:00 = 3s，TTL = 10s - (00:00:03 - 00:00:00) = 7s。所以，获取锁的时间并没有超过锁的有效期，我们认为获取锁成功。

重试

以上列举的示例是非常顺利获取到锁的情况，然而很多时候，分布式锁的获取没那么顺利，很可能出现以下情况：

• A已经获取到了两台Redis服务器的锁
• B已经获取到了两台Redis服务器的锁
• C已经获取到了一台Redis服务器的锁

如果三台客户端的请求一直处于阻塞状态（直到达到锁的有效期），会严重影响锁的获取效率，这时就需要重试机制。

重试机制：在一开始，同时向所有的（这里是5台）Redis服务器，发送

SET key value EX senconds NX

主从切换导致锁失效

实际上，在RedLock算法中，如果Redis服务配置了主从关系，仍然会出现我们之前提出的问题——主从切换导致锁失效。

为了解决这个问题，我们需要延迟Redis slave节点提升为Redis master节点的时间，延迟的时间就是锁的有效剩余时间（TTL），这样，就不会出现锁失效的问题了（这似乎只存在于理论层面，如果你知道如何延迟slave提升master的时间，请一定要分享给我）。

释放锁

释放锁就很简单了，给每台服务器都发送一个删除锁的命令就可以了，因为咱们的脚本已经保证了，只会删除与当前业务有关联的锁。

结语

梳理了那么多，终于来到了结尾，你也发现了，基于Redis实现一个分布式锁，并没有想象的那么简单，细节问题真的很多很多。另外，至少在我看来，RedLock算法实在是有些重量级了，如果不是那么在乎Redis主从切换导致的锁不一致的问题，单Redis其实就已经足够了。

另外，RedLock.net中使用了一个变量

extendUnlockSemaphore

lock

最后，还可以基于Zookeeper来实现分布式锁，有兴趣的可以去了解一下。