分布式锁的实现——Redis/Redisson-吉桂权的个人主

一、前言——在没有分布式锁之前，碰到的问题

1、我简单写了一个程序，以借助Redis实现减库存为例；

起始库存stock在redis中为1000个；

写一个简单的接口来减库存，首先申明两点；

Redis为单现场应用，不存在线程不安全性；
我在程序中加了synchronized同步代码块，是为了直接排除进程内的多线程问题；不考虑性能；

显然在单体应用中，即使碰到多线程并发，也不会存在问题；

@RestController
public class LockController {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Value("${server.port}")
    private String port;

    @GetMapping("/lock/deduct")
    public String deductStock(){
        synchronized (this){
            int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
            if (stock > 0){
                int realStock = stock - 1;
                redisTemplate.opsForValue().set("stock", String.valueOf(realStock));
                System.out.println(port + "扣减成功，剩余库存：" + realStock);
            }else {
                System.out.println(port + "扣减失败， 库存不足");
            }
        }

        return port + "--end";
    }
}

2、我使用Jmeter进行了并发测试：

瞬间压了100次请求：重复2次；

3、查看结果：

显然，结果正确，没什么好讲的；

4、但是在集群部署高可用的情况下呢，因为实际业务中不可能只部署一台应用服务，肯定要是多台的集群，我就简单地模拟两台：

http://127.0.0.1:9000
http://127.0.0.1:9001

并使用nginx作为代理负载均衡：

那么我们只要访问 http://localhost/lock/deduct 即可实现对上述两个服务的负载均衡请求：

5、执行测试：如果没有问题的话，最终的结果应该是库存剩余为：

显然，即使加了synchronized同步代码块的程序，在集群部署的高并发情况下，还是出现了问题；

原因就是因为：

synchronized为进程内（单JVM多线程间）的锁，对于单进程来说，是没有问题的；但是对于多进程的集群情况（多JVM），synchronized也已经无能为力了，这个时候就需要引出我们的“分布式锁”了！！！

注意：

上面的问题表面上看是最后的库存不对，实际是“有中间商品出现了超卖问题！”

所谓我们分布式锁，其实解决的是：同一个资源，在多个进程之间被同时操作，且相互不可见，同一瞬间只能有一个服务在对此资源进行操作；

二、“分布式锁”的概念

为了防止分布式系统中的多个进程之间相互干扰，我们需要一种分布式协调技术来对这些进程进行调度。而这个分布式协调技术的核心就是来实现这个分布式锁。

1、为什么要使用分布式锁？

成员变量 A 存在 JVM1、JVM2、JVM3 三个 JVM 内存中
成员变量 A 同时都会在 JVM 分配一块内存，三个请求发过来同时对这个变量操作，显然结果是不对的
不是同时发过来，三个请求分别操作三个不同 JVM 内存区域的数据，变量 A 之间不存在共享，也不具有可见性，处理的结果也是不对的注：该成员变量 A 是一个有状态的对象

2、分布式锁应该具备哪些条件？

在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行
高可用的获取锁与释放锁
高性能的获取锁与释放锁
具备可重入特性（可理解为重新进入，由多于一个任务并发使用，而不必担心数据错误）
具备锁失效机制，防止死锁
具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败

3、分布式锁的实现有哪些？

Memcached：利用 Memcached 的 add 命令。此命令是原子性操作，只有在 key 不存在的情况下，才能 add 成功，也就意味着线程得到了锁。
Redis：和 Memcached 的方式类似，利用 Redis 的 setnx 命令。此命令同样是原子性操作，只有在 key 不存在的情况下，才能 set 成功。
Zookeeper：利用 Zookeeper 的顺序临时节点，来实现分布式锁和等待队列。Zookeeper 设计的初衷，就是为了实现分布式锁服务的。
Chubby：Google 公司实现的粗粒度分布式锁服务，底层利用了 Paxos 一致性算法。

三、通过Redis实现分布式锁：

Redis在实现分布式锁的过程中：真正的代码并不复杂，但是我们需要好好理解一下！

分布式锁实现的三个核心要素：

1、加锁

最简单的方法是使用 setnx 命令。key 是锁的唯一标识，按业务来决定命名。比如想要给一种商品的秒杀活动加锁，可以给 key 命名为 “lock_sale_商品ID” 。而 value 设置成什么呢？我们可以姑且设置成 1。加锁的伪代码如下：

setnx（lock_sale_商品ID，1）

当一个线程执行 setnx 返回 1，说明 key 原本不存在，该线程成功得到了锁；当一个线程执行 setnx 返回 0，说明 key 已经存在，该线程抢锁失败。

加锁过程如下：

2、解锁

有加锁就得有解锁。当得到锁的线程执行完任务，需要释放锁，以便其他线程可以进入。释放锁的最简单方式是执行 del 指令，伪代码如下：

del（lock_sale_商品ID）

释放锁之后，其他线程就可以继续执行 setnx 命令来获得锁。

3、锁超时

锁超时是什么意思呢？如果一个得到锁的线程在执行任务的过程中挂掉，来不及显式地释放锁，这块资源将会永远被锁住（死锁），别的线程再也别想进来。所以，setnx 的 key 必须设置一个超时时间，以保证即使没有被显式释放，这把锁也要在一定时间后自动释放。setnx 不支持超时参数，所以需要额外的指令，伪代码如下：

expire（lock_sale_商品ID， 30）

综合伪代码如下：

if（setnx（lock_sale_商品ID，1） == 1）{
    expire（lock_sale_商品ID，30）
    try {
        do something ......
    } finally {
        del（lock_sale_商品ID）
    }
}

貌似是没问题了？

实际上，上面的伪代码，还存在很多问题：

四、以上伪代码的问题汇总

1、setnx 和 expire 的非原子性

设想一个极端场景，当某线程执行 setnx，成功得到了锁：

setnx 刚执行成功，还未来得及执行 expire 指令，节点 1 挂掉了。

这样一来，这把锁就没有设置过期时间，变成死锁，别的线程再也无法获得锁了。

怎么解决呢？setnx 指令本身是不支持传入超时时间的，set 指令增加了可选参数，伪代码如下：

set（lock_sale_商品ID，1，30，NX）  //这里的NX相当于执行的是setnx

这样就可以取代 setnx 指令。

2、del 导致误删

又是一个极端场景，假如某线程成功得到了锁，并且设置的超时时间是 30 秒。

如果某些原因导致线程 A 执行的很慢很慢，过了 30 秒都没执行完，这时候锁过期自动释放，线程 B 得到了锁。

随后，线程 A 执行完了任务，线程 A 接着执行 del 指令来释放锁。但这时候线程 B 还没执行完，线程A实际上删除的是线程 B 加的锁。

这里可不是简单的一次误删！！！

想象一下，在高并发情况下，C又会得到锁，待会B执行完成了，删除的课就是C的锁了；

之后的D、E、F…

这样导致的最终结局现象就是锁失效…

那么，该怎么避免这种情况呢？

可以在 del 释放锁之前做一个判断，验证当前的锁是不是自己加的锁。至于具体的实现，可以在加锁的时候把当前的线程 ID 当做 value，并在删除之前验证 key 对应的 value 是不是自己的线程 ID。

加锁时：

String threadId = Thread.currentThread().getId()
set（key，threadId ，30，NX）          //当然不一定非要用threadId，使用其他的能够唯一确定锁拥有者的标识都可以！

解锁时：

if（threadId .equals(redisClient.get(key))）{
    del(key)
}

但是，这样做又隐含了一个新的问题，判断和释放锁是两个独立操作，不是原子性。

3、操作仍未完成，锁却被自动释放掉了

基于误删锁的前提下，由于我们无法确定程序成功处理完成数据的具体时间，这就为超时时间的设置提出了难题。设置时间过长、过短都将影响程序并发的效率。

如果A的过期时间设置的是30秒，但是A的操作超过了30s，还没有完成，A的锁岂不是要被自动释放掉，怎么办呢？

我们可以让获得锁的线程开启一个守护线程，用来给快要过期的锁“续命”。

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

当线程 A 执行完任务，会显式关掉守护线程。

另一种情况，即使节点 1 在获取到锁之后，忽然断电，由于线程 A 和守护线程在同一个进程，守护线程也会停下。这把锁到了超时的时候，没人给它续命，也就自动释放了。

4、小结：

解决了上面考虑到的这一系列问题，我们设计出来的分布式锁，才会是实际可用的！！！

五、Redis实现分布式锁代码实战：

package com.jiguiquan.lock.redislock.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @author jiguiquan
 * @create 2020-07-05 17:06
 */
@RestController
public class LockController {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Value("${server.port}")
    private String port;

    @GetMapping("/lock/deduct")
    public String deductStock(){
        String lockKey = "product_001";   //一般来说，我们得锁肯定是为了某一个具体商品设置的，防止超卖
        String clientId = UUID.randomUUID().toString();   //使用UUID生成当前线程调用此方法时的唯一标识

        try{
            while (!(redisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS))){
                //做了个while自旋，但是要根据业务，也可以获取不到锁时，执行其他操作
//                try {
//                    TimeUnit.SECONDS.sleep(1);   //为了防止while自旋太过频繁，我设置了1s自旋一次，但是后来发现效率太慢了
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
            }
            //获取到锁之后，才进行后面的操作
            int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
            if (stock > 0){
                int realStock = stock - 1;
                redisTemplate.opsForValue().set("stock", String.valueOf(realStock));
                System.out.println(port + "扣减成功，剩余库存：" + realStock);
            }else {
                System.out.println(port + "扣减失败， 库存不足");
            }
        }finally {
            //使用try_finally是为了防止异常，最后没有删除锁
            if (clientId.equals(redisTemplate.opsForValue().get(lockKey))){
                redisTemplate.delete(lockKey);
            }
        }

        return port + "--end";
    }
}

关于4.3小结，提到的使用守护线程续命，我在本代码中没有实现，但是，其实这个已经是很不错了；

而且，在没有获取到锁时候的后续处理，应该由业务决定，不一定非要自旋去获取这把锁；（比如，若抢不到，可能就直接返回给前端：业务繁忙，请稍后尝试）；

上面为什么不自己再去实现守护线程，而是实际使用中，可以使用Redisson高效地实现一套完善的分布式锁，

且Redisson已经实现了守护线程“自动续命”的逻辑；

六、通过Redisson分布式锁

1、Redisson和Jedis差不多，都是Redis在java中的客户端实现，

但是不同的是，Redisson更多地解决了很多分布式场景下的Redis使用；

https://redisson.org/

Redisson实现分布式锁的原理图如下：

2、使用Redisson快速实现分布式锁实战

2.1、pom.xml：

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.6.5</version>
</dependency>

2.2、启动类中注入Redisson的依赖：

package com.jiguiquan.lock.redislock;

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class RedisLockApplication {

    public static void main(String[] args) {
        SpringApplication.run(RedisLockApplication.class, args);
    }

    /**
     * 注意，此处的Redisson并不是为了替代RedisTemplate，是独立的，不冲突，我们只用Redisson处理分布式锁问题
     * 其它的，redis使用，我们仍是使用RedisTemplate,底层仍是使用Jedis；
     * 因为Redisson有自己的缺点，就是在处理Redis的api上面不够丰富（术业有专攻）
     * @return
     */
    @Bean
    public Redisson redisson(){
        //此为单机模式
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.174.141:6379").setDatabase(0);
        return (Redisson)Redisson.create(config);
    }
}

2.3、RedissonLockController.java：

package com.jiguiquan.lock.redislock.controller;

import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

@RestController
public class RedissonLockController {
    @Autowired
    private Redisson redisson;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Value("${server.port}")
    private String port;

    @GetMapping("/lock/deduct")
    public String deductStock(){
        String lockKey = "product_001";   //一般来说，我们得锁肯定是为了某一个具体商品设置的，防止超卖

        RLock redissonLock = redisson.getLock(lockKey);  //这只是拿到锁对象，并没有真正开始加锁
        try{
            boolean isLock = redissonLock.tryLock(30000, 15000, TimeUnit.MILLISECONDS);
            if (isLock){
                //获取到锁之后，才进行后面的操作
                int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));
                if (stock > 0){
                    int realStock = stock - 1;
                    redisTemplate.opsForValue().set("stock", String.valueOf(realStock));
                    System.out.println(port + "扣减成功，剩余库存：" + realStock);
                }else {
                    System.out.println(port + "扣减失败， 库存不足");
                }
            }
        }catch (Exception e){

        }finally {
            //使用try_finally是为了防止异常，最后没有释放锁
            redissonLock.unlock();
        }

        return port + "--end";
    }
}

2.4、测试结果：

2.5、偶尔有时候，发现最后的总库存不正确，很郁闷，一直以为是Redisson实现分布式锁有漏洞，导致超卖了；

其实最后才发现，是并发量太大了，Jmeter发出的请求，并没有全部成功，存在一定的Error率：

这样正好就对应上了！1000*6.9% = 69，就是最后的剩余库存！

七、补充

其实Redisson在处理redis集群主从复制时候，开始会出现锁丢失的情况，从而导致锁失效；

Redis原生是解决不了这个问题的；

此时，还有一个解决方案，就是RedLock（红锁）；

自行了解：后面探索使用Zookeeper解决分布式锁问题！