基于Redis的分布式锁设计

作者: xiaoxiaotank

前言

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

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

在开始之前,咱们先定些规则:

  • 关于示例代码:
  • 需要搭配准备的示例代码,该示例采用C编写
  • 示例中的材料Id固定为10000
  • 示例中的材料初始库存均为100
  • 关于Redis中的Key:
  • 指示材料库存的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 $"剩余库存:";
}

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

单应用中的锁

提到锁,大多数人首先想到的应该就是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 $"剩余库存:";
}

结果和们所期望的一样,剩余库存为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已经没用了,是时候进入咱们的主题了——基于Redis的分布式锁设计。

初步的思路是这样的:

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

有了思路,接下来就该想一下如何实现了。很幸运,Redis的命令SETNX key value完全满足们的需求,实现如下:

对Redis命令不熟悉的同学,可以参考这篇Redis命令文档

/// <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 $"剩余库存:";
        }
        finally
        {
            // 释放锁
            await _redisDatabase.Database.KeyDeleteAsync(lockKey, CommandFlags.DemandMaster);
        }
    }
    else
        throw new Exception("获取锁失败");
}

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

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

版本2

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

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

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

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

这样,可以吗?当然不可以,们需要将SETEXPIRE两个操作合并为一个原子性操作,那们应该怎么做呢?别担心,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 $"剩余库存:";
        }
        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 $"剩余库存:";
        }
        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 $"剩余库存:";
        }
        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[] ,
             values: new RedisValue[] , 
             CommandFlags.DemandMaster);
        }
    }
    else
        throw new Exception("获取锁失败");
}

如果你没有使用的示例代码,而是自己写的,可能会出现锁未被正确释放的问题:执行完lua脚本后,返回的是0。这可能是因为你使用了Json序列化工具来将对象序列化为字符串,以将其存放到Redis中。但是由于Json序列化字符串时,将引号(“)也序列化为了(“),这就会导致字符串"123"存入到Redis中为”\“123\““。具体解决办法可以参考实现的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 $"剩余库存:";
        }
        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[] ,
             values: new RedisValue[] ,
             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[] ,
    values: new RedisValue[] { resourceId, (long)expiresIn.TotalMilliseconds },
    CommandFlags.DemandMaster);
}

使用RedLock.net中的分布式锁

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

Redis官方文档整理了常用语言的分布式锁实现,也梳理了RedLock的实现原理。

/// <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 $"剩余库存:";
    }
    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服务器获取到了锁

  • 获取锁所用的时间 = 最后一台获取到锁的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的时间,请一定要分享给)。

释放锁

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

结语

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

另外,RedLock.net中使用了一个变量extendUnlockSemaphore,而不是使用的lock,具体原因可以参考:Reentrant Async Locks in C

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

更多推荐

更多
  • Java8反应式编程指南-六、使用调度器获得并发性和并行性 RxJava 的调度器,缓冲、节流和去抖动,调试可观察对象及其调度器,可观察的间隔及其默认调度程序,调度器的类型,将可观察对象和调度器相结合,平行性,节流,去抖动,缓冲器和窗口操作器,背压操作人员,调度器。即时调度器,调度员。蹦床调度员
  • Java8反应式编程指南-七、测试 RxJava 应用 使用简单订阅进行测试,阻塞可观测类,聚合运算符和 BlockingObservable 类,使用聚合运算符和 BlockingObservable 类进行测试,使用 TestSubscriber 类进行深入测试,借助 TestSched
  • Java8反应式编程指南-八、资源管理与 RxJava 扩展 资源管理,使用 Observable.cache 缓存数据,使用升降机创建自定义操作员,使用 Observable.compose 运算符组合多个运算符,介绍可观察的使用方法, 通过前面的章节,我们已经学习了如何使用 RxJava
  • Java8反应式编程指南-五、组合器、条件和错误处理 结合可观察实例,条件运算符,处理错误,HTTP 客户端示例,拉链操作员,组合测试操作符,合并运算符,concat 操作员,电磁轴承操作员,takeUntil、takeWhile、skipUntil和 skipWhile条件运算符,def
  • Java8反应式编程指南-四、转换、过滤和积累您的数据 可观测变换,过滤数据,积累数据,使用各种 flatMap 运算符的变换,分组项目,附加有用的变换运算符, 现在我们有了从各种源数据创建`Observable`实例的方法,是时候围绕这些实例构建编程逻辑了。我们将介绍用于实现逐步计算
  • Java8反应式编程指南-三、创建和连接可观察对象、观察者和主体 从方法中观察到的,可观察的、公正的方法,其他可观察的工厂方法,可观察的创建方法,订阅和取消订阅,冷热可观察实例,主体实例,可连通可观测类, RxJava 的`Observable`实例是反应式应用的构建块,RxJava 的这一优势
  • Java8反应式编程指南-一、反应式编程简介 什么是反应式编程?,我们为什么要被动?,介绍 RxJava,下载并设置 RxJava,比较迭代器模式和 RxJava 可观测值, 如今,术语反应式编程正在成为一种趋势。各种编程语言的库和框架正在涌现。关于反应式编程的博客文
  • Java8反应式编程指南-二、使用 Java 8 的函数结构 Java 8 中的 Lambdas,用 lambdas 实现无功和示例,纯函数与高阶函数,引入新的语法和语义,Java 8 和 RxJava 中的功能接口,纯函数,高阶函数,RxJava 与函数式编程, 函数式编程不是一个新概念;
  • Java8反应式编程指南-零、序言 这本书涵盖的内容,这本书你需要什么,这本书是给谁的,公约,读者反馈,客户支持,下载示例代码,勘误表,盗版,问题, 反应式编程已经存在了几十年。从 Smalltalk 还是一种年轻的语言时起,就有一些反应式编程的实现。然而,它只是最
  • Go编程秘籍-三、数据转换与组合 本章将展示一些在数据类型之间转换、使用非常大的数字、使用货币、使用不同类型的编码和解码(包括 Base64 和gob)以及使用闭包创建自定义集合的示例。转换数据类型和接口转换,使用 math 和 math/big ...
  • 近期文章

    更多
    文章目录

      推荐作者

      更多