《大厂高并发分布式锁从入门到实战》第2讲之redis分布式锁

         

目录

一、spring boot项目引入 redis依赖

二、redis的超卖问题

三、解决方案

四、分布式锁的三种实现方式

---

        上一讲中，我们了解了分布式锁的背景以及jvm锁失效的场景，mysql数据库锁来解决超卖问题，这一讲我们将会学习redis的超卖问题，以及解决方案。

一、spring boot项目引入 redis依赖

org.springframework.boot

<artifactId>spring-boot-srarter-data-redis

spring.redis.host = xxx.xxx.xx.xxx

注入StringRedisTemplate  

@Autowired

private StringRedisTemplate  stringRedisTemplate;

 二、redis的超卖问题

 执行同样的测试用例，发现也会产生超卖问题。那么我们有哪些方法来解决redis中的超卖问题呢？

三、解决方案

• jvm本地锁

我们最直接能够想到的方案就是使用jvm本地锁来处理

• redis乐观锁  watch multi exec

watch 可以监控一个或者多个key的值，如果在事务（exec）执行之前，key的值发生变化则取消事务执行

multi： 开启事务

exec ： 执行事务

执行测试用例后发现控制台输出如下报错：

 需要做如下改进：

 使用redisTemplate的execute方法，将watch，multi，exec 包裹在方法内。

• 分布式锁 ：跨进程 跨服务 跨服务器

场景：超卖现象  （NoSQL）缓存击穿

四、分布式锁的三种实现方式

通常情况下，我们使用的最多的就是基于以下三种方式实现的分布式锁

• 基于redis实现
• 基于zookeeper/etcd实现
• 基于mysql实现

（1）基于redis实现

加锁：setnx

解锁：del

重试：递归 循环

缺陷一：独占排他使用，防死锁发生（如果redis客户端程序从redis服务中获取到锁之后立马宕机，则锁不能得到释放，导致其他服务无法获取锁，解决：给锁添加过期时间，expire），

解决方法： 我们将代码做如下改进，把设值和设过期时间放到同一个redis指令中，保证原子性操作。

缺陷二：经过改造之后，虽然可以确保每个请求获取锁之后有3秒的过期时间，但是不能保证锁的误删，比如业务逻辑需要执行5秒，当多个请求同时过来的时候，第一个请求执行到3秒的时候，就已经把锁释放掉了，也就是说业务逻辑还没执行完，锁已经被释放， 这个时候第二个请求在第3秒就可以获取锁了，随着越来越多的请求进来，会出现更多的锁误删现象，就会导致一个锁失效问题的产生，那么我们如何来解决锁误删锁失效问题呢？解铃还须系铃人。

此时我们可以想到如果自己的请求处理结束之后，只释放自己的锁，并且在业务逻辑没处理完成的情况下把锁自动续期，这样不就能够解决我们的问题啦。

解决方法：

获取锁设置值的时候，我们可以set一个uuid。代表当前请求的唯一标识

 在finally中，增加当前请求的uuid和redis中设置的uuid的相等判断，如果相当，才去释放锁。

 缺陷三：由于最后的判断和释放锁还是分成两个指令来执行，如果判断指令刚刚执行完，3秒过期时间就到了，锁就被释放掉了。

解决方案：lua脚本，一次性发送多个指令给redis，redis单线程执行指令遵守one by one 规则

EVAL script numkeys key [key ...] arg [arg ...] ,  eval 输出的是返回值，而不是打印内容

script： lua脚本字符串

numkeys：key列表的元素数量

key 列表：以空格分割，key[index从1开始]

arg列表：以空格分割，arg[index从1开始]

if redis.call('get','lock')  == uuid 

then 

        return redis.call('del','lock')

else 

        return 0

end 

 String script = "

if redis.call('get',KEYS[1])  == ARGV[1] 

then 

        return redis.call('del',KEYS[1])

else 

        return 0

end ";

this.redisTemplate.execute(new DefaultRedisScript<>(script,Boolean.class),Arrays.asList("lock"),uuid);

缺陷四：不可重入，如果a方法里面调用了b方法，a和b方法都会进行获取redis锁，那么a获取锁之后，b就无法获取锁了，一直进行阻塞。 

解决方案：我们可以参照reentrantLock可重入锁的非公平锁实现方式去实现redis的可重入锁：

hash+lua脚本，首先判断锁是否存在，则直接获取锁 hset key field value

如果锁存在则判断是否自己的锁（hexists），如果是自己的锁则重入：hincrby key field increment ,根据以上分析编写加锁lua脚本以及传入参数。

if redis.call('exists',KEYS[1] ) == 0 or redis.call('hexists',KEYS[1] ,ARGV[1]) ==1

then 

        redis.call('hset',KEYS[1],ARGV[1],1)

        redis.call('expire', KEYS[1], ARGV[2])

        return 1

else

        return 0

end

KEY: lock

ARG: uuid , 30

有了加锁的lua脚本之后，我们肯定还是需要解锁的lua脚本，解锁先判断自己的锁是否存在（hexists）不存在则返回nil，如果自己的锁存在，则减1（hincrby -1）,判断减1后的值是否为0，为0则释放锁（del）并返回1，不为0，返回0.

if redis.call('hexists',KEYS[1],ARGV[1])  == 0

then

        return nil

elseif redis.call('hincrby',KEYS[1],ARGV[1],-1) == 0

then 

        return   redis.call('del',KEYS[1])

else 

        return 0

end 

key: lock

arg: uuid

代码改造：

加锁方法

 解锁方法

 经过hash+lua脚本的方式，实现之后，我们还需要考虑锁的自动续期，这个时候就需要一个定时任务了。

 判断自己的锁是否存在（hexists），如果存在则重置过期时间

if  redis.call('hexists',KEYS[1],ARGV[1]) == 1

then 

        return redis.call('expire',KEYS[1],ARGV[2])

else 

        return 0

end

key: lock

arg: uuid, 30

定时执行锁自动续期方法：