怎样用Redis Nx方式实现分布式锁？

拼搏现实的明天。 2023-10-15 10:07 4阅读 0赞

本地锁只能控制所在虚拟机中的线程同步执行，现在要实现分布式环境下所有虚拟机中的线程去同步执行就需要让多个虚拟机去共用一个锁，虚拟机可以分布式部署，锁也可以分布式部署，如下图：

![cb82ff465c0f37c3bcc00baa9ba1aaff.png][]

虚拟机都去抢占同一个锁，锁是一个单独的程序提供加锁、解锁服务，谁抢到锁谁去查询数据库。

该锁已不属于某个虚拟机，而是分布式部署，由多个虚拟机所共享，这种锁叫分布式锁。

### **分布式锁实现方案** ###

实现分布式锁的方案有很多，常用的如下：

1、基于数据库实现分布锁

利用数据库主键唯一性的特点，或利用数据库唯一索引的特点，多个线程同时去插入相同的记录，谁插入成功谁就抢到锁。

2、基于redis实现锁

redis提供了分布式锁的实现方案，比如：SETNX、set nx、redisson等。

拿SETNX举例说明，SETNX命令的工作过程是去set一个不存在的key，多个线程去设置同一个key只会有一个线程设置成功，设置成功的的线程拿到锁。

3、使用zookeeper实现

zookeeper是一个分布式协调服务，主要解决分布式程序之间的同步的问题。zookeeper的结构类似的文件目录，多线程向zookeeper创建一个子目录(节点)只会有一个创建成功，利用此特点可以实现分布式锁，谁创建该结点成功谁就获得锁。

### **Redis NX实现分布式锁** ###

redis实现分布式锁的方案可以在[http://redis.cn][http_redis.cn]网站查阅，地址[http://www.redis.cn/commands/set.html][http_www.redis.cn_commands_set.html]

使用命令： SET resource-name anystring NX EX max-lock-time 即可实现。

NX：表示key不存在才设置成功。

EX：设置过期时间

这里启动三个ssh客户端，连接redis: docker exec -it redis redis-cli

先认证: auth redis

同时向三个客户端发送测试命令如下：

表示设置lock001锁，value为001，过期时间为30秒

Plain Text
    SET lock001 001 NX EX 30

命令发送成功，观察三个ssh客户端发现只有一个设置成功，其它两个设置失败，设置成功的请求表示抢到了lock001锁。

如何在代码中使用Set nx去实现分布锁呢?

使用spring-boot-starter-data-redis 提供的api即可实现set nx。添加依赖：

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-pool2</artifactId>
        <version>2.6.2</version>
    </dependency>

添加依赖后，在bean中注入restTemplate。我们先分析一段伪代码如下：

if(缓存中有){
    
      返回缓存中的数据
    }else{
    
      获取分布式锁
      if(获取锁成功）{
           try{
             查询数据库
          }finally{
             释放锁
          }
      }
     
    }

#### **1、获取分布式锁** ####

使用redisTemplate.opsForValue().setIfAbsent(key,vaue)获取锁。

这里考虑一个问题，当set nx一个key/value成功1后，这个key(就是锁)需要设置过期时间吗?

如果不设置过期时间当获取到了锁却没有执行finally这个锁将会一直存在，其它线程无法获取这个锁。所以执行set nx时要指定过期时间，即使用如下的命令。

SET resource-name anystring NX EX max-lock-time

具体调用的方法是：redisTemplate.opsForValue().setIfAbsent(K var1, V var2, long var3, TimeUnit var5)

#### **2、如何释放锁** ####

释放锁分为两种情况：key到期自动释放，手动删除。

1)key到期自动释放的方法

因为锁设置了过期时间，key到期会自动释放，但是会存在一个问题就是 查询数据库等操作还没有执行完时key到期了，此时其它线程就抢到锁了，最终重复查询数据库执行了重复的业务操作。

怎么解决这个问题?

可以将key的到期时间设置的长一些，足以执行完成查询数据库并设置缓存等相关操作。

如果这样效率会低一些，另外这个时间值也不好把控。

2)手动删除锁

如果是采用手动删除锁可能和key到期自动删除有所冲突，造成删除了别人的锁。

比如：当查询数据库等业务还没有执行完时key过期了，此时其它线程占用了锁，当上一个线程执行查询数据库等业务操作完成后手动删除锁就把其它线程的锁给删除了。

要解决这个问题可以采用删除锁之前判断是不是自己设置的锁，伪代码如下：

if(缓存中有){
    
      返回缓存中的数据
    }else{
    
      获取分布式锁: set lock 01 NX
      if(获取锁成功）{
           try{
             查询数据库
          }finally{
             if(redis.call("get","lock")=="01"){
                释放锁: redis.call("del","lock")
             }
             
          }
      }
     
    }

以上代码第11行到13行非原子性，也会导致删除其它线程的锁。查看文档上的说明：[http://www.redis.cn/commands/set.html][http_www.redis.cn_commands_set.html]

上述优化方法会避免下述场景：a客户端获得的锁(键key)已经由于过期时间到了被redis服务器删除，但是这个时候a客户端还去执行DEL命令。而b客户端已经在a设置过期时间之后重新获取了这个同样key的锁，那么a执行DEL就会释放了b客户端加好的锁。

解锁脚本的一个例子将类似于以下：

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

在调用setnx命令设置key/value时，每个线程设置不一样的value值，这样当线程去删除锁时可以先根据key查询出来判断是不是自己当时设置的vlaue，如果是则删除。

这整个操作是原子的，实现方法就是去执行上边的lua脚本。

Lua 是一个小巧的脚本语言，redis在2.6版本就支持通过执行Lua脚本保证多个命令的原子性。

什么是原子性?

这些指令要么全成功要么全失败。

以上就是使用Redis Nx方式实现分布式锁，为了避免删除别的线程设置的锁需要使用redis去执行Lua脚本的方式去实现，这样就具有原子性，但是过期时间的值设置不存在不精确的问题。

[cb82ff465c0f37c3bcc00baa9ba1aaff.png]: https://img-blog.csdnimg.cn/img_convert/cb82ff465c0f37c3bcc00baa9ba1aaff.png
[http_redis.cn]: https://link.zhihu.com/?target=http%3A//redis.cn
[http_www.redis.cn_commands_set.html]: https://link.zhihu.com/?target=http%3A//www.redis.cn/commands/set.html