前言

基于Redis的分布式锁实现，原理很简单嘛：检测一下Key是否存在，不存在则Set Key，加锁成功，存在则加锁失败。对吗？这么简单吗？

如果你真这么想，那么你真的需要好好听我讲一下了。接下来，咱们找个例子研究一下。

在开始之前，咱们先定些规则：

• 关于示例代码：
  • 需要搭配我准备的示例代码，该示例采用C#编写
  • 示例中的材料Id固定为10000
  • 示例中的材料初始库存均为100
• 关于Redis中的Key：
  • 指示材料库存的Key为ProductStock_10000
  • 自己实现的分布式锁中，指示锁的Key为DistributedLock_10000
  • RedLock.net中，指示锁的Key为redlock:10000

1、假如没有锁

如果没有锁，我们可以通过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！

2、单应用中的锁

提到锁，大多数人首先想到的应该就是Monitor的语法糖lock了，这是大多数人最先接触到的一种锁。在单应用中，因为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个

3、应用集群中的锁

3.1版本1

很明显，lock已经没用了，是时候进入咱们的主题了——基于Redis的分布式锁设计。

初步的思路是这样的：

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("获取锁失败");
}

我没找到Jmeter统计请求成功或失败次数的方法，所以使用了聚合报告，通过报告里的错误率手动计算。如果你知道，可以分享给我，谢谢！

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

3.2版本2

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

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

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

那我们应该怎么办呢？对，就是给锁加一个过期时间！不过SETNX命令并没有“过期时间”参数，那我们就需要在获取到锁后，通过EXPIRE命令设置锁的过期时间。

这样，可以吗？当然不可以，我们需要将SET和EXPIRE两个操作合并为一个原子性操作，那我们应该怎么做呢？别担心，Redis对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.3版本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("获取锁失败");
}

3.4版本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("获取锁失败");
}

如果你没有使用我的示例代码，而是自己写的，可能会出现锁未被正确释放的问题：执行完lua脚本后，返回的是0。这可能是因为你使用了Json序列化工具来将对象序列化为字符串，以将其存放到Redis中。但是由于Json序列化字符串时，将引号（"）也序列化为了（"），这就会导致字符串"123"存入到Redis中为"“123"”。

***3.5***版本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);
}

3.6使用RedLock.net中的分布式锁

以上的版本5，已经包含了分布式锁的基本思想了，不过我写的肯定比较简陋

/// <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("获取锁失败");
}

4.0站在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算法诞生了。

5.0RedLock

我们不讨论时钟漂移，所以我们假设，多台服务器之间的时钟漂移很小，以至于我们可以忽略它。

基本原理

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

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

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

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

• 获取锁所用的时间 = 最后一台获取到锁的Redis服务器获取到锁的时间 - 加锁的开始时间
• 锁的有效剩余时间（TTL） = 锁的过期时间 - 获取锁所用的时间

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

认为锁获取成功的条件有两个：

1. 超过半数的Redis服务器获取到了锁
2. 获取锁的时间没有超过锁的有效期

重试

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

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

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

重试机制：在一开始，同时向所有的（这里是5台）Redis服务器，发送SET key value EX senconds NX命令，当所有服务器都返回结果后，判断是否以达成“锁获取成功的两个条件”，如果达成了，则锁获取成功。如果没有，则立即将已获取的锁释放掉，并等待一小段时间，重复以上步骤（一般会尝试3次）。如果这期间仍未达成“锁获取成功的两个条件”，则认为锁获取失败。

主从切换导致锁失效

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

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

释放锁

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

6.0结语

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

另外，RedLock.net中使用了一个变量extendUnlockSemaphore，而不是使用的lock
最后，还可以基于Zookeeper来实现分布式锁，有兴趣的可以去了解一下。

原文链接:https://blog.csdn.net/weixin_48994163/article/details/118799565