高并发场景下的解决方案以及分布式锁的实现

作者: 知鱼君 | 来源:发表于2021-06-22 16:05 被阅读0次

分布式技术是从单体应用到微服务演进中必须要掌握的技能，希望借《分布式》这个专栏总结一些我对这方面的学习心得，同时分享一些我对分布式的思考。

背景

在项目中经常会遇到并发场景，比如最典型的秒杀场景，假设后端有一张库存表，当大批请求过来时，都会经历读库存+减库存两个步骤，比如库存为100，两个线程先分别读到了库存100，然后各自执行减1并写回数据库，结果数据库数据为99而不是预期的98，这种情况如果不加以技术手段进行处理，很容易导致库存超卖。

另一个场景的例子是本人之前做过的请假系统，后端有一张表存储着请员工的可支配假期天数，当有多条请假申请被同时审批时，它们都要去执行读假期+减假期的操作，如果两个线程都先读到了假期数据，然后分别执行减假期的操作，就很容易导致假期少扣的情况。

我将对这一开发中的典型场景总结一些我自己的思考。

思路一：SQL优化

为了具体模拟秒杀这一场景，假设库存表叫做stock，商品数量叫做num，有业务代码计算出的新数目为new_num=num-1，原来执行的SQL为：

update stock set num=new_num where id = id=#{id};

我们可以改进SQL，让数据库根据自己当前的值更新数据，比如写成下面这样：

update stock set num=num-1 where id = #{id};

这样虽然看似解决了问题，但是假设此时库存的数量为1，两个线程各自减1后，会发现库存被扣成了-1，同样也无法完全解决并发问题。

思路二：代码加锁

既然两个线程同时拿数据还是有问题，那么有没有办法让它们依次执行，于是很容易想到在查库存之前先加上锁，比如Java的synchronized关键字。

这能很好的保证串行，保证每次只有一个线程去执行查库存和扣库存的代码段，但同样缺点也非常多：

无法做到细粒度的控制，如果有的人秒杀商品A、有的秒杀商品B，都要走秒杀方法，相互之间不冲突但是也只能串行执行，访问会变得很慢。
因为spring里的@service是单例的，所以可以这么写，如果用python语言的django框架，则没法在代码层做类似的控制，不通用。
只支持单点（单机、服务器环境），无法做到水平扩展，现在基本都是集群部署的，仅能一台机器生效。

也许，我们需要一种第三方的机制，去实现这样的一把锁。

思路三：基于MySQL的悲观锁

因为所有的线程都在访问同一张表，可以在数据库上加锁，典型的做法有悲观锁和乐观锁两种。

悲观锁利用了select...for update 语法，例如：

select * from stock where id=#{id} for update;

注意：使用悲观锁要把业务代码放在事务里。

悲观锁获取数据时对数据行了锁定，其他事务要想获取锁，必须等原事务结束。

使用selec...for update会把数据给锁住，MySQL InnoDB默认Row-Level Lock，所以只有「明确」地指定主键，MySQL 才会执行Row lock (只锁住被选取的数据) ，否则MySQL 将会执行Table Lock (将整个数据表单给锁住)。

使用悲观锁的缺点是，因为要锁表，所以并发性不高，如果并发过高，数据库压力过大，会宕机。

思路四：基于MySQL的乐观锁

乐观锁相对悲观锁而言，乐观锁假设认为数据一般情况下不会造成冲突，所以在数据进行提交更新的时候，才会正式对数据的冲突与否进行检测，如果发现冲突了，则让返回用户错误的信息，让用户决定如何去做。

乐观锁的实现有两种方法。

使用版本号实现乐观锁

版本号的实现方式有两种，一个是数据版本机制，一个是时间戳机制。具体如下。

使用数据版本（Version）记录机制实现，这是乐观锁最常用的一种实现方式。何谓数据版本？即为数据增加一个版本标识，一般是通过为数据库表增加一个数字类型的 version 字段来实现。当读取数据时，将version字段的值一同读出，数据每更新一次，对此version值加一。当我们提交更新的时候，判断数据库表对应记录的当前版本信息与第一次取出来的version值进行比对，如果数据库表当前版本号与第一次取出来的version值相等，则予以更新，否则认为是过期数据。

如果使用乐观锁的话，扣库存的SQL变化为下：

1.查询出商品信息

select num,version from stock where id=#{id};

扣减数目。new_num = num -1
修改数据库。

update stock set num=new_num, version=version+1 where id=#{id} and version=#{version};

如果此时的version已经变化了，则会更新失败，可以根据具体业务采取重试或者报错。

使用条件限制实现乐观锁

这个适用于只更新是做数据安全校验，适合库存模型，扣份额和回滚份额，性能更高，十分方便实用。

SQL改为：

update stock SET num = num - #{buyNum} where id = #{id} and num - #{buyNum} >= 0;

注意：乐观锁的更新操作，最好用主键或者唯一索引来更新，这样是行锁，否则更新时会锁表。

乐观锁在扛并发能力上比悲观锁要高，应用会更多一些。但是当并发越来越高的话，对数据库压力也会比较大，因为如果更新失败去重试，可能会不断地去查询数据库。

思路五：基于Redis分布式锁

既然数据库的锁实现永远绕不开并发问题，我们就求助于第三方中间件，比如Redis。

这里提一下，如果在并发不高的情况下，基于MySQL的悲观锁或乐观锁能解决问题，而且维护起来比较简单，不需要引入额外的组件，系统可用性高。所谓做架构，其实就是在做trade-off。

简单讲下实现原理，线程A先在Redis里设置该商品的key（比如goods_1）为1，如果线程B也查到1，则不但轮询，直到线程A释放掉这把锁，线程B才能再拿到这把锁。

我这里用Python语言简单模拟了实现一下，用了redis-py的库，写了一个Lock类，里面包含两个方法，获取锁acquire和释放锁release：

import redis
class Lock:
    def __init__(self, name):
        self.redis_client = redis.Redis(host="127.0.0.1")
        self.name = name

    def acquire(self):
        # 如果为空或者为None那么代表获取到锁
        if not self.redis_client.get(self.name):
            self.redis_client.set(self.name, 1) 
            return True
        else:
            # 如果不想阻塞住轮询获取锁，可以直接返回False
            while True:
                import time
                time.sleep(1)
                if self.redis_client.get(self.name):
                    return True

    def release(self):
        self.redis_client.delete(self.name)

思路六：Redis分布式锁的问题和改进

确保原子性

有经验的小伙伴一定会发现我上面代码的问题，当并发非常高时，可能多个线程同时进入到if not self.redis_client.get(self.name)，于是同时获取到了锁，这就使得分布式锁失效了。

所以我们要保证在读取和设置值这两步的原子性。

如何解决呢？Redis天然提供了setnx命令，可以保证原子操作，命令在指定的key不存在时，为key设置指定的值。

于是我们可以把上面的Lock代码升级为如下：

import redis
class Lock:
    def __init__(self, name):
        self.redis_client = redis.Redis(host="127.0.0.1")
        self.name = name

    def acquire(self):
        if self.redis_client.setnx(self.name, 1): # 如果不存在设置并且返回1，否在返回0， 这是原子操作
            return True
        else:
            # 如果不想阻塞住轮询获取锁，可以直接返回False
            while True:
                import time
                time.sleep(1)
                if self.redis_client.setnx(self.name, 1): # 注意这个地方也要改
                    return True

    def release(self):
        self.redis_client.delete(self.name)

锁的过期时间

我们继续往下思考，假如一个线程拿到了锁，但是由于某些原因（比如网络连不上redis，代码异常，突然断电）并未释放锁，导致其他线程都拿不到锁，处于永远等待的状态，造成死锁。

比较好的解决办法就是设置给锁设置过期时间，我们利用redis-py库提供的set函数，并给加上原子操作和过期参数。

self.redis_client.set(self.name, self.id, nx=True, ex=15)

锁的续租

但是，过期设置会产生新的问题：

如果当前的线程如果在一段时间后没有执行完，然后key过期了，会导致另一个线程拿到锁也会继续执行，程序不安全。

另外，另一个线程有可能先执行完，会将当前锁给释放掉，释放掉了本属于原来的线程的锁。

所以，如果当前的线程没有执行完，那它还应该在适当的时候去续租，将过期时间重新设置。

比较经验性的做法是，比如过期时间是15s，那就在15s的2/3的时候去续租，也就是运行10s以后去将过期时间重新设置为15s。可以在程序里启动一个新的线程去完成去定时的完成这个续租的过程。

删除锁的安全性

前面还提到了一点，别的线程有把当前线程的锁给释放掉的风险。

我们可以给获取锁设置值的时候，不用1，而用uuid，并在释放的时候判断是否为当前线程的uuid，防止被其他线程给释放掉。

于是分布式锁的代码现在进化成了下面这样：

import uuid
import redis
class Lock:
    def __init__(self, name, id=None):
        self.id = uuid.uuid4()
        self.redis_client = redis.Redis(host="192.168.0.104")
        self.name = name

    def acquire(self):
        # 设置过期时间15s, 如果不存在设置并且返回1，否在返回0， 这是原子操作
        if self.redis_client.set(self.name, self.id, nx=True, ex=15): 
            # 启动一个线程然后去定时的刷新这个过期 这个操作最好也是使用lua脚本来完成
            return True
        else:
            while True:
                import time
                time.sleep(1)
                if self.redis_client.set(self.name, self.id, nx=True, ex=15):
                    return True

    def release(self):
        # 先做一个判断，先取出值来然后判断当前的值和你自己的lock中的id是否一致，如果一致删除，如果不一致报错
        # 这块代码不安全， 将get和delete操作原子化，但是Redis可以使用lua脚本去完成这个操作使得该操作原子化
        id = self.redis_client.get(self.name)
        if id == self.id:
            self.redis_client.delete(self.name)
        else:
            print("不能删除不属于自己的锁")

我上面的代码并不完整，因为想要get和delete操作原子化，Redis并没有相关操作指令，仍然会出现安全问题，同样在续租的过程里也会有这样的问题。

但是，Redis 为这样的用户场景提供了 lua 脚本支持，用户可以向服务器发送 lua 脚本来执行自定义动作，获取脚本的响应数据。Redis 服务器会单线程原子性执行 lua 脚本，保证 lua 脚本在处理的过程中不会被任意其它请求打断。

总结Redis分布式锁

至此，我们总结一下分布式锁需要解决的问题：

互斥性：任意时刻只能有一个客户端拥有锁，不能同时多个客户端获取
死锁：获取锁的客户端因为某些原因而宕机，而未能释放锁，其他客户端无法获取此锁，需要有机制来避免该类问题的发生
- 比如代码异常，导致无法运行到release
- 当前服务器网络出问题，比如该release时，突然访问不了Redis了
- 服务器断电
安全性：锁只能被持有该锁的用户删除，而不能被其他用户删除
容错：当部分节点宕机，客户端仍能获取锁或者释放锁

思路七：Python第三方的开源库

前面提到了需要用Lua脚本解决安全问题，但是lua脚本写起来比较麻烦，这里我推荐网络上一个第三方的开源库，有非常完整的Python实现Redis的分布式锁的方法。

github地址为：https://github.com/ionelmc/python-redis-lock

具体的实现可以参看 https://github.com/ionelmc/python-redis-lock/blob/master/src/redis_lock/init.py 这个文件

我在这里贴一下这个库的流程图：

这个库的总体实现还是跟我们前面讲的一致，但是有些细节上的差别。通过阅读源码，我总结如下：

为了减少无意义的轮询，会另外建一个Redis的列表，使用了Redis的blpop指令，移出并获取列表的第一个元素，如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。
续租那里比较难读懂，简单来讲，它会启动一个线程去做续租，假设锁的时间是15s，续租检查的间隔就是15s的2/3，即10s，续租做的事情，就是重新设置锁的超时为15s，并且在10s后继续进行检查。
如果业务代码产生阻塞，续租时可能造成代码无限等待，需要慎用，所以该库提供了可选参数。
使用了__enter__,__exit__魔术方法，客户端调用时支持with实现。

至此，Redis锁的分布式锁部分已经全部介绍完了，它的优点是使用简单，同时性能高，而且Redis也是非常常用的组件，不需要额外去维护。

此外，由于单机Redis也有挂掉的风险，我们在生产上会使用Redis cluster或Redis sentinel保证高可用。

思路八：基于zookeeper的分布式锁

除了使用Redis实现，zookeeper也可以实现分布式锁。但是如果项目中并未用到zookeeper，我觉得没必要引入额外的组件，对这种方式，我就不多做介绍了。

思路九：Java中的Redission

如果是Java代码，可以使用Redission包来实现分布式锁。这里不多做介绍了，推荐一篇文章：《Redis分布式锁》

示例代码如下：

// 1.构造redisson实现分布式锁必要的Config
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:5379").setPassword("123456").setDatabase(0);
// 2.构造RedissonClient
RedissonClient redissonClient = Redisson.create(config);
// 3.获取锁对象实例（无法保证是按线程的顺序获取到）
RLock rLock = redissonClient.getLock(lockKey);
try {
    /**
     * 4.尝试获取锁
     * waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败
     * leaseTime   锁的持有时间,超过这个时间锁会自动失效（值应设置为大于业务处理的时间，确保在锁有效期内业务能处理完）
     */
    boolean res = rLock.tryLock((long)waitTimeout, (long)leaseTime, TimeUnit.SECONDS);
    if (res) {
        //成功获得锁，在这里处理业务
    }
} catch (Exception e) {
    throw new RuntimeException("aquire lock fail");
}finally{
    //无论如何, 最后都要解锁
    rLock.unlock();
}

后续规划

因时间关系，这篇文章里只集中讲解了分布式锁理论，并未涉及完整项目代码，后续我会针对这一块做一个完整的demo，来模拟秒杀场景，欢迎持续关注，一键三连~