Redis高阶知识：实现分布式锁

谁践踏了优雅 2023-05-21 12:51 19阅读 0赞

### 文章目录 ###

*   *  Redis高阶知识：实现分布式锁
     *   *  分布式锁概念
         *  Redis的分布式锁实现
         *   *  加锁过程
             *  释放锁过程
             *  直接del(key)，释放锁存在的问题
             *  并发泄露问题
         *  参考文章

## Redis高阶知识：实现分布式锁 ##

> redis的一些基础知识，可以看这篇博客
> 
> [Redis，看这篇博客，吊打所有面试官][Redis]
> 
> 但是实际找工作的时候，单单基础知识是不够的，面试官大概率问完基础知识后，会问Redis的一些高阶知识，比如**Redis实现分布式锁**、**Redis实现消息队列**等。
> 
> 本文从代码角度，详细阐述如何使用Redis实现分布式锁

### 分布式锁概念 ###

本地锁：单个节点(机器)中实现线程同步的方式，Synchronized、ReentrantLock均为本地锁。

**分布式锁：分布式环境下(多个节点)，实现不同节点的线程同步的方法**

如下图一个分布式锁实例，不同节点抢占**操作数据库的分布式锁**，只有获取到分布式锁，才能对数据库进行操作。

![format_png][]

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

#### 加锁过程 ####

**set重载函数**

/* nxxx:可以取"NX"、"XX" NX:只有键key不存在时才会设置key的值 XX:只有键key存在时才会设置key的值 expx:可以取"EX"、"PX" EX、PX均为设置过期时间，前者单位是秒，后者单位是毫秒 time：就是过期时间值 */
    public String set(final String key, final String value, final String nxxx, final String expx,
          final long time) { 
        checkIsInMultiOrPipeline();
        client.set(key, value, nxxx, expx, time);
        return client.getStatusCodeReply();
      }

了解了上面的set重载函数,可以很轻松得写出正确得加锁代码

public boolean lock(String key, String value) { 
        //原子操作
        String statusCode = jedis.set(key, value, "NX", "EX", 1);
        //获取锁成功，返回true,否则返回false
        return "OK".equals(statusCode);
    }

考虑一个场景，分布式环境下，秒杀一件商品，使用**商品id作为key**,当一个节点的线程，发现key(商品id)不存在，则直接设置<key,value>,返回`true`,分布式锁获取成功。如果另一个节点尝试获取锁，即设置商品id，发现已经存在，返回false，分布式锁获取失败。

#### 释放锁过程 ####

在分布式环境下，直接del解锁存在问题，下文会详细讲解

//直接删除key，就是释放锁过程
    jedis.del(key);

**为什么key要设置过期时间**

如果一个节点的线程获取锁后，直接宕机了，没有释放锁，则会导致死锁存在，任何节点的线程都无法获取该分布式锁。设置key有过期时间后，如果获取到锁的节点宕机后，经过一段时间会直接释放锁。

#### 直接del(key)，释放锁存在的问题 ####

场景实例：节点A获取到分布式锁后(成功设置key)，过期时间为30s，30s后，节点A没有执行完任务，但是过期时间到了，锁被直接释放，即key被del。与此同时另一个节点B尝试获取锁，由于锁已经释放，所以节点B获取锁成功。经过一段时间后，节点A终于执行完毕，会进行释放锁操作，但是联系上下文，可以知道，**节点A此时释放的锁，是节点B的锁**

为了解决以上情况，可以采取如下方法：获取锁的时候，可以设置key的value为ThreadID(获取分布式锁的节点线程id).锁释放的时候,判断当前尝试释放锁的线程，是否与加锁线程一致，一致则允许释放锁。

//加锁修改代码
    String threadId = Thread.currentThread().getId();
    jedis.set（Key, threadId, "NX", "EX", 1）;
    
    //解锁修改代码，lua代码保证操作的原子性
    String script = "if Redis.call('get', KEYS[1]) == ARGV[1] then return Redis.call('del', KEYS[1]) else return 0 end";
    RedisClient.eval(script, Collections.singletonList(Key), Collections.singletonList(threadId));

#### 并发泄露问题 ####

虽然上面的方法，解决了**节点A误删key的情况**，但是存在节点A和节点B,都进入同步块的情况

解决方法：获得锁的线程，开启一个守护线程，用来给快要过期的锁续航。

![format_png 1][]

*  当过去了29秒，线程A还没执行完，这时候守护线程会执行expire指令，为这把锁“续命”20秒。守护线程从第29秒开始执行，每20秒执行一次。

![format_png 2][]

*  线程执行完后，会显式得关闭守护线程
    
    ![format_png 3][]
 *  获取锁的线程忽然宕机，守护线程与该线程在同一进程，也会结束，所以key到了超时时间，也会自动释放
    
    ![format_png 4][]

### 参考文章 ###

[基于Redis的分布式锁实现][Redis 1]

[Redis]: https://blog.csdn.net/zycxnanwang/article/details/105477972
[format_png]: /images/20230521/8469351171214120973984d5d8cabab9.png
[format_png 1]: /images/20230521/5365b073050247a18134b694bea95491.png
[format_png 2]: /images/20230521/e0a8a8d244f340b29fef82195d262cec.png
[format_png 3]: /images/20230521/9d1f1777a719419084b3a83f5d60a7e7.png
[format_png 4]: /images/20230521/890dfb85eee8461f85b26b4be23f4b68.png
[Redis 1]: https://juejin.im/post/5cc165816fb9a03202221dd5#heading-0