redis锁和分布式锁的实现

古城微笑少年丶 2022-12-12 15:25 204阅读 0赞

## [redis环境搭建][redis] ##

## [redis在java、spring、springboot中的实现][redis_java_spring_springboot] ##

## *redis锁* ##

### 1、添加依赖 ###

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>

### 2、application.yml配置 ###

spring:
      redis:
        host: 192.168.65.128
        password: root

### 3、注入redisTemplate对象 ###

@Autowired
    private RedisTemplate redisTemplate;
    /** * 单体架构版本 - 查询所有学生(双重锁机制) */
    @Override
    public List<Student> queryAll() { 
        //先查询redis是否有缓存该数据
        List<Student> students= (List<Student>) redisTemplate.opsForValue().get("students");
        //判断缓存中是否存在
        if(students== null){ 
            synchronized (this) { 
                students= (List<Student>) redisTemplate.opsForValue().get("students");
                if(students== null) { 
                    System.out.println("查询了数据库了！！！！！");
                    //查询数据库
                    students= stuDao.queryAll();
                    //重建缓存
                    redisTemplate.opsForValue().set("students", students);
                    redisTemplate.expire("students", 5, TimeUnit.MINUTES);
                }
            }
        }
        return students;
    }

## *redis分布式锁* ##

### 理解什么是分布式锁? ###

要介绍分布式锁，首先要提到与分布式锁相对应的是线程锁、进程锁。

线程锁：主要用来给方法、代码块加锁。当某个方法或代码使用锁，在同一时刻仅有一个线程执行该方法或该代码段。线程锁只在同一JVM中有效果，因为线程锁的实现在根本上是依靠线程之间共享内存实现的，比如synchronized是共享对象头，显示锁Lock是共享某个变量（state）。

进程锁：为了控制同一操作系统中多个进程访问某个共享资源，因为进程具有独立性，各个进程无法访问其他进程的资源，因此无法通过synchronized等线程锁实现进程锁。

分布式锁：当多个进程不在同一个系统中，用分布式锁控制多个进程对资源的访问。

### 分布式锁的原理 ###

基于redis的setnx命令，setnx的作用就是设置一个key值，如果在redis中该key值不存在就设置成功，如果存在就会设置失败。在分布式集群环境下，多个服务器的线程同时设置一个key，哪个服务器的线程设置成功，就表示该服务器的线程获得了锁对象，其他线程必须等待。获得锁的线程需要记得，在某个时刻进行锁的释放（删除那个key）。

### 分布式锁的使用场景 ###

线程间并发问题和进程间并发问题都是可以通过分布式锁解决的，但是强烈不建议这样做！因为采用分布式锁解决这些小问题是非常消耗资源的！分布式锁应该用来解决分布式情况下的多进程并发问题才是最合适的。

有这样一个情境，线程A和线程B都共享某个变量X。

如果是单机情况下（单JVM），线程之间共享内存，只要使用线程锁就可以解决并发问题。

如果是分布式情况下（多JVM），线程A和线程B很可能不是在同一JVM中，这样线程锁就无法起到作用了，这时候就要用到分布式锁来解决。

### 分布式锁的实现 ###

分布式锁实现的关键是在分布式的应用服务器外，搭建一个存储服务器，存储锁信息，这时候我们很容易就想到了Redis。首先我们要搭建一个Redis服务器，用Redis服务器来存储锁信息。

实现的时候要注意的几个关键点：

1、锁信息必须设置过期超时，不能让一个线程长期占有一个锁而导致死锁；

2、同一时刻只能有一个线程获取到锁。

几个要用到的redis命令：

setnx(key, value)：“set if not exits”，若该key-value不存在，则成功加入缓存并且返回1，否则返回0。

get(key)：获得key对应的value值，若不存在则返回nil。

getset(key, value)：先获取key对应的value值，若不存在则返回nil，然后将旧的value更新为新的value。

expire(key, seconds)：设置key-value的有效期为seconds秒。

### 代码实现 ###

**1、分布式锁工具类的封装**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNjkzNjAz_size_16_color_FFFFFF_t_70_pic_left]

/** * 分布式锁的工具类 */
    @Component
    public class LockUtil { 
        @Autowired
        private RedisTemplate redisTemplate;
    
        //redis原始连接对象
        private RedisConnection redisConnection;
    
        //lua脚本的缓存签名字符串
        private String lockSha;
        private String unlockSha;
    	//解决多线程访问同一个共享变量的时候容易出现的并发问题，使得同一线程共享，不同线程间访问隔离
        private ThreadLocal<String> threadLocal = new ThreadLocal<>();
    
        /** * 添加分布式锁的Lua脚本 * 1、为什么要给锁添加超时时间？ * 当获取锁后，程序突然宕机，导致锁无法进行释放。 * 2、为什么要使用lua脚本 * 当获取锁后，在设置过期时间之前程序突然宕机，导致锁无法进行释放。 * 加锁和设置过期时间必须是原子性，所以考虑使用lua脚本实现。 */
        private String lockLua = "local key = KEYS[1]\n" +
                "local value = ARGV[1]\n" +
                "local time = ARGV[2]\n" +
                "\n" +
                "local result = redis.call('setnx', key, value)\n" +
                "if result == 1 then\n" +
                " --当前获得了分布式锁\n" +
                " --设置锁的过期时间\n" +
                " redis.call('expire', key, time)\t\n" +
                " return true\t\n" +
                "end\n" +
                "\n" +
                "--没有获得分布式锁\n" +
                "return false";
    
        //解锁的lua脚本
        private String unlockLua = "--要删除的是什么锁\n" +
                "local key = KEYS[1]\n" +
                "local uuid = ARGV[1]\n" +
                "\n" +
                "--获取锁中的uuid\n" +
                "local lockUUID = redis.call('get', key)\n" +
                "\n" +
                "--判断是不是自己上的锁\n" +
                "if uuid == lockUUID then\n" +
                " --是自己上的锁，删除\n" +
                " redis.call('del', key)\n" +
                " return true\n" +
                "end\n" +
                "\n" +
                "--不是自己上的锁\n" +
                "return false";
    
        @PostConstruct
        public void init(){ 
            //获得原始连接，因为redis模板对象中没有封装对lua脚本的方法
            redisConnection = redisTemplate.getConnectionFactory().getConnection();
            //缓存lua脚本到redis端
            lockSha = redisConnection.scriptLoad(lockLua.getBytes());
            unlockSha = redisConnection.scriptLoad(unlockLua.getBytes());
        }
    
        /** * 加锁的方法 * 为什么要设置uuid。 * 当线程A获取到锁时，并设置了锁的超时时间，由于某种原因当线程A执行业务所需要的时间大于设置的锁超时时间时，锁失效后，线程B获取到了锁， * 在线程B执行过程中，线程A执行完毕后，释放锁的时候，就会误删线程B的锁。 * @return */
        public boolean lock(String key, int timeout){ 
    		//设置uuid，在解锁时，根据uuid进行删除对应的锁。遵循锁由谁创建则由谁释放锁
            String uuid = UUID.randomUUID().toString();
            threadLocal.set(uuid);
            //执行加锁的lua脚本
            boolean flag = redisConnection.evalSha(lockSha, ReturnType.BOOLEAN, 1,
                    key.getBytes(), uuid.getBytes(), (timeout + "").getBytes());
            return flag;
        }
    
        /** * 解锁的方法 * @return */
        public boolean unlock(String key){ 
            String uuid = threadLocal.get();
            //执行解锁的lua
            boolean flag = redisConnection.evalSha(unlockSha, ReturnType.BOOLEAN, 1,
                    key.getBytes(), uuid.getBytes());
            return flag;
        }
    }

@Autowired
    private RedisTemplate redisTemplate;
    
    @Autowired
    private LockUtil lockUtil;
    
    /** * 分布式集群版 - 分布式锁 * @return */
    @Override
    public List<Student> queryAll() { 
    
       //先查询redis是否有缓存该数据
       List<Student> students= (List<Student>) redisTemplate.opsForValue().get("students");
       //判断缓存中是否存在
       if(students== null) { 
           //通过lua脚本获得分布式锁，返回true说明获取到了锁
           boolean flag = lockUtil.lock("lock", 120);
           if(flag){ 
               //查询数据库
               students= stuDao.queryAll();//3M
    
               //重建缓存
               redisTemplate.opsForValue().set("students", students);
               redisTemplate.expire("students", 5, TimeUnit.MINUTES);
    
               //重建缓存后，释放锁
               lockUtil.unlock("lock");
           } else { 
               try { 
                   Thread.sleep(50);
               } catch (InterruptedException e) { 
                   e.printStackTrace();
               }
               return this.queryAll();
           }
       }
       return students;
    }

### 思考 ###

**Redis用setnx+expire实现分布式锁存在什么隐患，如何改进？**

用Redis实现分布式锁，2.6.12之前版本方案：setnx加锁，del释放锁，如果锁没释放，设置过期时间，到了时间，del释放锁。但是，这会存在一些问题。

1.  **setnx和expire不是原子操作：** 一旦redis宕机，expire没有设置成功，锁就无法释放。只有一个请求的setnx可以成功，任何一个请求的expire都可以成功。请求比较密集，过期时间一直刷新，导致锁一直有效。
2.  **超时后，删除其他线程的锁：** 在线程A执行过程中，锁已释放，A还在执行业务，但是还未删除锁。线程B获取锁执行业务，线程A执行完，A误删B的锁。
3.  **多个线程并发获取锁、释放锁：** 同一时间有线程A、B在访问同一代码块。

对于上面的隐患，Redis已改善。下面，我们针对隐患逐一改善。

1.  **Redis2.6.12以上版本，可以用set获取锁：** set可以实现setnx和expire，这个是原子操作。也可以通过lua脚本来实现。
2.  **Lua删除锁：** 通过uuid来控制锁由谁创建则由谁释放锁，Lua是原子操作。
3.  **让获取锁的线程开启一个守护线程，给线程还没执行完，又快要过期的锁续航：** 大概是这样的，线程A还没执行完，守护线程每当快过期时，延时expire时间。当线程A执行完，关闭守护线程。如果中间宕机，锁超过超时，守护线程也不在了，自动释放锁。

写作不易，既然来了，不妨点个关注，点个赞吧！！！

[redis]: https://blog.csdn.net/qq_40693603/article/details/108258712
[redis_java_spring_springboot]: https://blog.csdn.net/qq_40693603/article/details/108264563
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNjkzNjAz_size_16_color_FFFFFF_t_70_pic_left]: /images/20221123/def5cf2241ec4658910bd9e56ec8dcab.png