redis分布式锁

ab测试：防止锁误删：

这个名字先用着

2665人浏览 · 2022-04-26 17:31:52

这个名字先用着 · 2022-04-26 17:31:52 发布

一、为什么需要分布式锁：

为了保证一个方法在高并发情况下的同一时间只能被同一个线程执行，在传统单体应用单机部署的情况下，可以使用Java并发处理相关的API(如ReentrantLcok或synchronized)进行互斥控制。

但是，随着业务发展的需要，原单体单机部署的系统被演化成分布式系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，即synchronized在分布式系统中失效了。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题。

卖票问题

二、什么是分布式锁：

分布式锁其实就是，控制分布式系统不同进程共同访问共享资源的一种锁的实现。如果不同的系统或同一个系统的不同主机之间共享了某个临界资源，往往需要互斥来防止彼此干扰，以保证一致性。

我们先来看下，一把靠谱的分布式锁应该有哪些特征：

「互斥性」: 任意时刻，只有一个客户端能持有锁。
「锁超时释放」：持有锁超时，可以释放，防止不必要的资源浪费，也可以防止死锁。
「可重入性」:一个线程如果获取了锁之后,可以再次对其请求加锁。
「高性能和高可用」：加锁和解锁需要开销尽可能低，同时也要保证高可用，避免分布式锁失效。
「安全性」：锁只能被持有的客户端删除，不能被其他客户端删除

三、Redis分布式锁的方案：

日常开发中，秒杀下单、抢红包等等业务场景，都需要用到分布式锁。而Redis非常适合作为分布式锁使用.

方案一：SETNX + EXPIRE
方案二：SETNX + value值是（系统时间+过期时间）
方案三：使用Lua脚本(包含SETNX + EXPIRE两条指令)
方案四：SET的扩展命令（SET EX PX NX）
方案五：SET EX PX NX + 校验唯一随机值,再释放锁
方案六: 开源框架~Redisson
方案七：多机实现的分布式锁Redlock

在这里插入图片描述

Redis分布式锁方案一：SETNX key value + EXPIRE key seconds

reids中的命令：

setnx key value #当key不存在的时候，给key加锁，并设置值value
expire key seconds #给指定的key设置过期时间，即给锁设置一个过期时间，从而避免死锁问题
del key #手动释放key的锁，可以一次性释放多个：del key1 key2 key3

在这里插入图片描述

但是这个方案中，setnx和expire两个命令分开了，「不是原子操作」。如果执行完setnx加锁，正要执行expire设置过期时间时，进程宕机或者要重启维护了，那么这个锁就“长生不老”了，「别的线程永远获取不到锁啦」。

提到Redis的分布式锁，很多小伙伴马上就会想到setnx+ expire命令。即先用setnx来抢锁，如果抢到之后，再用expire给锁设置一个过期时间，防止锁忘记了释放。

SETNX 是SET IF NOT EXISTS的简写.日常命令格式是SETNX key value，如果 key不存在，则SETNX成功返回1，如果这个key已经存在了，则返回0。

假设某电商网站的某商品做秒杀活动，key可以设置为key_resource_id,value设置任意值，java中的伪代码如下：

if（jedis.setnx(key_resource_id,lock_value) == 1）{ //1.加锁
    expire（key_resource_id，100）; //2.设置过期时间
    try {
        do something  //3.业务请求
    }catch(){
  }
  finally {
       jedis.del(key_resource_id); //4.释放锁
    }
}

在这里插入图片描述
但以上伪代码还存在以下几个问题：
问题1）如下图所示，如果在setnx获得锁后，发生了断电、服务器挂掉，expire就会得不到执行，也会造成死锁。如下图所示：

问题的根源在于setnx和expire是两条指令而不是原子指令。这里的问题不能用Redis事务来解决，因为expire和setnx之间存在依赖关系，如果setnx没有强盗锁，expire同样不能执行。在Redis2.8以后，加入了set指令的扩展参数，使setnx和expire可以一起执行。

问题2）del导致误删
如下图所示，线程A获取了锁，并设置了过期时间为30秒，30秒后锁自动释放，但A业务30秒内还未完成自己的事，此时线程B来了并获取了锁；然后A执行完成，执行del命令删除锁，但此时释放的是B的锁。如下图所示:

在这里插入图片描述

解决方法：
解决方法一：设置超时时间远大于业务执行时间，但是会带来性能问题，一般不用；

解决方法二：
删除锁的时候要判断，是不是自己的，如果是再删除，使用 UUID:
可以将set指令的value参数设置为一个随机数或设置为当前线程的ID或者一个UUDID，释放锁时先匹配随机数，即value是否一致，然后再删除key。

Redis分布式锁方案二：使用Lua脚本(脚本中包含SETNX + EXPIRE两条指令)

Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令的原子性.

在这里插入图片描述

实际上，我们还可以使用Lua脚本来保证原子性（包含setnx和expire两条指令），lua脚本如下：

if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then
   redis.call('expire',KEYS[1],ARGV[2])
else
   return 0
end;

加锁代码如下：

 String lua_scripts = "if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then" +
            " redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end";   
Object result = jedis.eval(lua_scripts, Collections.singletonList(key_resource_id), Collections.singletonList(values));
//判断是否成功
return result.equals(1L);

Redis分布式锁方案方案三：SET的扩展命令（SET EX PX NX）

除了使用Lua脚本，保证SETNX + EXPIRE两条指令的原子性，我们还可以巧用Redis的SET指令扩展参数！（SET key value[EX seconds][PX milliseconds][NX|XX]），它也是原子性的！

SET key value[EX seconds][PX milliseconds][NX|XX]
NX :表示key不存在的时候，才能set成功，也即保证只有第一个客户端请求才能获得锁，而其他客户端请求只能等其释放锁，才能获取。
EX seconds :设定key的过期时间，时间单位是秒。
PX milliseconds: 设定key的过期时间，单位为毫秒
XX: key存在时设置value，成功返回OK，失败返回(nil)

如：

127.0.0.1:6379> set test “111” EX 100 NX
这样就完美的解决了分布式锁的原子性。

在这里插入图片描述

在这里插入图片描述
伪代码demo如下：

if（jedis.set(key_resource_id, lock_value, "NX", "EX", 100s) == 1）{ //加锁
    try {
        do something  //业务处理
    }catch(){
  }
  finally {
       jedis.del(key_resource_id); //释放锁
    }
}

但是呢，这个方案还是可能存在问题：

问题一：「锁过期释放了，业务还没执行完」。假设线程a获取锁成功，一直在执行临界区的代码。但是100s过去后，它还没执行完。但是，这时候锁已经过期了，此时线程b又请求过来。显然线程b就可以获得锁成功，也开始执行临界区的代码。那么问题就来了，临界区的业务代码都不是严格串行执行的啦。

问题二：「锁被别的线程误删」。假设线程a执行完后，去释放锁。但是它不知道当前的锁可能是线程b持有的（线程a去释放锁时，有可能过期时间已经到了，此时线程b进来占有了锁）。那线程a就把线程b的锁释放掉了，但是线程b临界区业务代码可能都还没执行完呢。

一些代码测试：
在这里插入图片描述

在这里插入图片描述

ab测试：

在这里插入图片描述

方案五：SET EX PX NX + 校验唯一随机值,再删除

既然锁可能被别的线程误删，那我们给value值设置一个标记当前线程唯一的随机数，在删除的时候，校验一下，不就OK了嘛。伪代码如下：

if（jedis.set(key_resource_id, uni_request_id, "NX", "EX", 100s) == 1）{ //加锁
    try {
        do something  //业务处理
    }catch(){
  }
  finally {
       //判断是不是当前线程加的锁,是才释放
       if (uni_request_id.equals(jedis.get(key_resource_id))) {
        jedis.del(lockKey); //释放锁
        }
    }
}

防止锁误删：

在这里插入图片描述

在这里，「判断是不是当前线程加的锁」和「释放锁」不是一个原子操作。如果调用jedis.del()释放锁的时候，可能这把锁已经不属于当前客户端，会解除他人加的锁。

在这里插入图片描述

为了更严谨，一般也是用lua脚本代替。lua脚本如下：

if redis.call('get',KEYS[1]) == ARGV[1] then 
   return redis.call('del',KEYS[1]) 
else
   return 0
end;

在这里插入图片描述

Redis分布式锁方案六：Redisson框架

方案五还是可能存在「锁过期释放，业务没执行完」的问题。有些小伙伴认为，稍微把锁过期时间设置长一些就可以啦。其实我们设想一下，是否可以给获得锁的线程，开启一个定时守护线程，每隔一段时间检查锁是否还存在，存在则对锁的过期时间延长，防止锁过期提前释放。

当前开源框架Redisson解决了这个问题。我们一起来看下Redisson底层原理图吧：

在这里插入图片描述

只要线程一加锁成功，就会启动一个watch dog看门狗，它是一个后台线程，会每隔10秒检查一下，如果线程1还持有锁，那么就会不断的延长锁key的生存时间。因此，Redisson就是使用Redisson解决了「锁过期释放，业务没执行完」问题。

Redis分布式锁方案七：多机实现的分布式锁Redlock+Redisson

前面六种方案都只是基于单机版的讨论，还不是很完美。其实Redis一般都是集群部署的：

在这里插入图片描述

如果线程一在Redis的master节点上拿到了锁，但是加锁的key还没同步到slave节点。恰好这时，master节点发生故障，一个slave节点就会升级为master节点。线程二就可以获取同个key的锁啦，但线程一也已经拿到锁了，锁的安全性就没了。

为了解决这个问题，Redis作者 antirez提出一种高级的分布式锁算法：Redlock。Redlock核心思想是这样的：

搞多个Redis master部署，以保证它们不会同时宕掉。并且这些master节点是完全相互独立的，相互之间不存在数据同步。同时，需要确保在这多个master实例上，是与在Redis单实例，使用相同方法来获取和释放锁。

我们假设当前有5个Redis master节点，在5台服务器上面运行这些Redis实例。

在这里插入图片描述

RedLock的实现步骤:如下

❝
1.获取当前时间，以毫秒为单位。
2.按顺序向5个master节点请求加锁。客户端设置网络连接和响应超时时间，并且超时时间要小于锁的失效时间。（假设锁自动失效时间为10秒，则超时时间一般在5-50毫秒之间,我们就假设超时时间是50ms吧）。如果超时，跳过该master节点，尽快去尝试下一个master节点。
3.客户端使用当前时间减去开始获取锁时间（即步骤1记录的时间），得到获取锁使用的时间。当且仅当超过一半（N/2+1，这里是5/2+1=3个节点）的Redis master节点都获得锁，并且使用的时间小于锁失效时间时，锁才算获取成功。（如上图，10s> 30ms+40ms+50ms+4m0s+50ms）
如果取到了锁，key的真正有效时间就变啦，需要减去获取锁所使用的时间。
如果获取锁失败（没有在至少N/2+1个master实例取到锁，有或者获取锁时间已经超过了有效时间），客户端要在所有的master节点上解锁（即便有些master节点根本就没有加锁成功，也需要解锁，以防止有些漏网之鱼）。
❞
简化下步骤就是：

按顺序向5个master节点请求加锁
根据设置的超时时间来判断，是不是要跳过该master节点。
如果大于等于3个节点加锁成功，并且使用的时间小于锁的有效期，即可认定加锁成功啦。
如果获取锁失败，解锁！
Redisson实现了redLock版本的锁，有兴趣的小伙伴，可以去了解一下哈~

分布式锁的代码实现：

在这里插入图片描述

在服务器端给库存设置值：
set stock 50;
在这里插入图片描述

在这里插入图片描述

absent:不存在的

模拟异常退出：
在这里插入图片描述

在这里插入图片描述

改进方案一：try-catch:
在这里插入图片描述

在这里插入图片描述

改造方案二：

在这里插入图片描述

改进方案三：UUID标记
在这里插入图片描述
总的代码：

package com.fan.controller;

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@RestController
public class RedisController {
    //使用redis模板类
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @RequestMapping("/getLock")
    public String getLock(){
        //1.从redis数据库取库存值,先在服务器那边的redis存好值，在这里取
        String lockValue = UUID.randomUUID().toString();
        //2.加锁设置，Absent：不存在,类似于我们一个ReentrantLock对象的lock方法。
        //***设置过期时间，是为了防止死锁，防止服务器宕机后，不能手动释放锁的问题
        Boolean absent = stringRedisTemplate.opsForValue().
                setIfAbsent("stocklock", lockValue,20, TimeUnit.SECONDS);
        if(!absent){//如果存在
            return "error";//返回加锁失败
        }

        try {
            Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                int realStock = stock -1;//卖出一张票
                stringRedisTemplate.opsForValue().set("stock",""+realStock);//将redis数据库数据更新
                System.out.println("售卖成功，剩余"+realStock);
                //int n=1/0;//模拟程序报错
                return "success";
            }else{
                System.out.println("剩余库存不足");
                //int n=1/0;//模拟程序报错
                return "fail";
            }
        } finally {
            //uuid的value值如果和当前的锁得到数据库的value    相等的话，则释放锁
            if(lockValue.equals(stringRedisTemplate.opsForValue().get("stocklock"))){
                //3.释放锁，类似于我们一个ReentrantLock对象的unlock方法。
                stringRedisTemplate.delete("stocklock");
            }
        }

    }
}

在这里插入图片描述