前言

本文力争以最简单的语言，以博主自己对分布式锁的理解，按照自己的语言来描述分布式锁的概念、作用、原理、实现。如有错误，还请各位大佬海涵，恳请指正。分布式锁分两篇来讲解，本篇讲解客户端，下一篇讲解redis服务端。

概念

如果把分布式锁的概念搬到这里，博主也会觉得枯燥。博主这里以举例的形式来描绘它。

试想一种场景，在一个偏远小镇上的火车站，只有一个售票窗口。

火车站来了10名旅客，前往售票窗口购买火车票，旅客只能排队购票，排到第一的旅客，可以与售票员沟通，买票。

好啦，以上就是一个分布式锁的场景，我们来分析一下每一个细节。

每位旅客可以理解为一个系统或者线程。他们在竞争售票员的工作时间。

是不是觉得分布式锁也不是什么高大上的概念。有同学会问，锁到底在哪里呢？还是买票场景，我们看看锁长什么样子。

我们深入想一下，这10位旅客本来是并行的（没有买票前，他们有的在吃饭，有的在玩手机，等等等），而到了买票的时候，就必须排队（串行），而不是一起买票。

没错，就是在特定的场景下，将并行的场景，变成穿行，就是分布式锁的奥义所在。

作用

分布式锁的作用不但非常大，而且非常多。

在软件设计中，比如电商秒杀活动。商家预备了1000件货物，也就只有这1000件货，有1500人参与秒杀，可以理解为1500个线程来排队购买商品。那就必须将这1500个线程排个队（比如按照时间），设置一把锁，一个购买过程结束，再开始下一个。

为什么redis可以实现分布式锁呢？

我们以购票举例，购票窗口前的这个锁，是每位旅客都可以看到的。

这里我们可以得出一个结论，一把锁首先要具有的属性是：想要获得锁的人都可以看到。

这把锁既不能属于服务器A，也不能属于服务器B，因为他们都不知道另一方的存在，那就必须选择一个公信的第三方来作为锁。当_当当~ redis闪亮登场。当然zookeeper也可以实现，这里先挖一个坑，以后再填zookeeper吧。

原理

加锁的基本思路

redis中有一条指令非常有意思，它叫做setnx

当redis中不存在key值为“lock”的时候，可以设置成功；当存在key值时，设置失败。

这句指令，好比是，询问一下，到我买票了吗？返回结果是1的时候，到您买票了；返回结果是0的时候，还没到您，稍后再询问。

我们的锁过程可以这样来操作：

setnx lock 锁值
处理业务逻辑
释放锁 del lock

优化一

为什么要优化？

试想，如果setnx lock 1 加锁成功，这个时候系统因为其他原因，挂掉了，就永远无法执行del lock了。

要避免这种情况，怎么办呢？给锁一个过期时间。

这样无论系统是否宕机，都会在10秒后释放锁。看似很美好，虽然setnx lock 1 与 expire lock 10之间的时间间隙非常小，但仍然有风险，加入系统执行完 setnx lock 1 后，宕机了，并没有执行过期指令 expire lock 10，再次产生了一把无法解开的锁，“死锁”。

这时候引入了一个概念，叫做原子操作。即这两条指令需要在一个原子操作内执行完成。

set key value [expiration EX seconds|PX milliseconds] [NX|XX]

优化二

why?上一个优化已经把上锁过程做成了原子操作，还需要什么优化呢？

当然有，试想一下，之前代码set lock 1 ex 10 nx，设置过期时间是10秒，那么这个10秒是否可靠呢？显然不可靠。

我们加锁的过程是加锁---执行业务代码---释放锁

加入业务代码的执行时间超过10秒呢？是不是业务代码还没有执行完，锁就已经释放了。放在购票场景中，第一位旅客还没有完成购票，第二位旅客就开始购票。显然不合理。怎么办呢？

这里我们需要估计业务代码的执行时间，加入预估出来的时间是10秒，可以在业务代码中开辟一个“续命”的操作。

加锁 set lock 1 ex 10 nx
- 每过3秒，把该锁的时间重新设置为 10秒
执行业务代码
释放锁 del lock

这里的续命时间间隔 = 过期时间 10S / 3

这样设置比较合理，可以防止一次续命失败。

优化三

纳尼?还有问题吗？

有，而且可以算是一个bug，我们一直在用 set lock 1 ex 10 nx 来加锁，用del lock 来释放锁。

我们需要明确知道，释放的锁，是自己加上的。

可以set lock uuid ex 10 nx 来解决该问题。

拓展-可重入锁

一个线程获取到锁以后，再次获取锁，就是可重入锁。

但博主现在遇到的问题，一般不需要可重入锁即可解决。java中ReentrantLock就是可重入锁。

可重入锁，对代码的复杂度增加了很多，玩不好，容易扯裆。谨慎使用。

实现

已经讲了很多优化相关的内容，这里博主就直接写优化后的代码了。

博主使用java来实现。而redis官方（https://redis.io/clients#java）推荐的有三个框架。分别是Jedis、lettuce、Redisson。

由于博主在本篇中主要讨论单个redis的情况，而redisson主要用来处理分布式redis，下一篇博文使用redisson，敬请期待。

springboot2.x 默认采用了 lettuce，所以博主就使用lettuce来实现分布式锁。

引入依赖

<!-- data-redis中集成了lettuce -->

<dependency>

    <groupId>org.springframework.boot</groupId>

    <artifactId>spring-boot-starter-data-redis</artifactId>

</dependency>

<!-- redis链接池 -->

<dependency>

    <groupId>org.apache.commons</groupId>

    <artifactId>commons-pool2</artifactId>

</dependency>

<!-- alibaba json -->

<dependency>

    <groupId>com.alibaba</groupId>

    <artifactId>fastjson</artifactId>

    <version>1.2.72</version>

</dependency>

配置文件

既然要测试分布式锁，那么就至少应该跑两份代码，所以配置文件也应该是两份，这里博主偷个懒，提供一份配置文件，另一份配置文件修改下server的端口即可。

server:

  port: 80

spring:

  redis:

    # redis的ip地址

    host: redis的ip地址

    # redis的端口号

    port: 6379

    # redis的密码

    password: 你的密码

    lettuce:

      pool:

      	# 最大链接数

        max-active: 30

        # 链接池中最大空闲链接数

        max-idle: 15

        # 最大阻塞等待链接时长 默认不限制 -1

        max-wait: 2000

        # 最小空闲链接数

        min-idle: 10

      # 链接超时时长

      shutdown-timeout: 10000

lettuce配置类

这个类博主就不细讲了，springboot整合lettuce，序列化博主更偏爱FastJson

import com.alibaba.fastjson.support.spring.GenericFastJsonRedisSerializer;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.data.redis.connection.RedisConnectionFactory;

import org.springframework.data.redis.core.RedisTemplate;

import org.springframework.data.redis.serializer.StringRedisSerializer;

/**

 * @author xujp

 * redis 配置类 将RedisTemplate交给spring托管

 */

@Configuration

public class RedisConfig {

    @Bean

    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory){

        RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();

        redisTemplate.setConnectionFactory(redisConnectionFactory);

        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

        GenericFastJsonRedisSerializer genericFastJsonRedisSerializer = new GenericFastJsonRedisSerializer();

        redisTemplate.setKeySerializer(stringRedisSerializer);

        redisTemplate.setValueSerializer(genericFastJsonRedisSerializer);

        redisTemplate.setHashKeySerializer(stringRedisSerializer);

        redisTemplate.setHashValueSerializer(genericFastJsonRedisSerializer);

        redisTemplate.afterPropertiesSet();

        return redisTemplate;

    }

}

分布式锁

重头戏来了，手写分布式锁的核心代码示例。

import com.redis.demo1.thread.WatchDog;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.data.redis.core.RedisTemplate;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;

import java.util.concurrent.TimeUnit;

/**

 * @author xujp

 */

@RestController

@RequestMapping("/test")

public class TestController {

    @Autowired

    private RedisTemplate redisTemplate;

    @GetMapping

    public void lock(){

        String uuid = UUID.randomUUID().toString();

        //System.out.println(uuid);

        WatchDog watchDog;

        try {

            // 自旋

            while (true) {

                // 尝试获取锁

                Boolean hasLock = redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3l, TimeUnit.SECONDS);

                if(hasLock) {

                    // 看门狗“续命“

                    watchDog = new WatchDog(redisTemplate, uuid);

                    watchDog.start();

                    // 业务逻辑start

                    int num = (int) redisTemplate.opsForValue().get("num");

                    //Thread.sleep(4000); // 假设业务需要4s处理时间

                    redisTemplate.opsForValue().set("num", num - 1);

                    System.out.println(num);

                    // 业务逻辑处理 end

                    break;

                }else{

                    // 睡眠100ms再自旋

                    Thread.sleep(100);

                }

            }

        }catch (Exception e){

            System.out.println(e);

        }finally {

            // 关闭锁

            String l = (String) redisTemplate.opsForValue().get("lock");

            if (l.equalsIgnoreCase(uuid)) {

                redisTemplate.delete("lock");

            }

        }

    }

}

分布式锁“续命”代码示例

import org.springframework.data.redis.core.RedisTemplate;

import java.util.concurrent.TimeUnit;

/**

 * @author xujp

 */

public class WatchDog extends Thread {

    private RedisTemplate redisTemplate;

    private String uuid;

    public WatchDog(RedisTemplate redisTemplate, String uuid){

        this.redisTemplate = redisTemplate;

        this.uuid = uuid;

    }

    public void run(){

        // 续命逻辑

        while (true){

            try {

                // 获取锁的value

                Object redisUUID = redisTemplate.opsForValue().get("lock");

                // 判断当前父线程是否已经释放锁，如果父线程已释放，则跳出线程

                if(redisUUID==null || !redisUUID.toString().equals(uuid)){

                    break;

                }

                // 续命

                redisTemplate.expire("lock", 3l, TimeUnit.SECONDS);

                // 没隔1s续命一次

                Thread.sleep(1000);

            } catch (InterruptedException e) {

                e.printStackTrace();

                break;

            }

        }

    }

}

测试

首先我们将代码分别以80和81端口run起来。

有精力的同学，还可以再搭建一个nginx将请求分流到80和81。这里博主简单粗暴地使用jmeter请求。

博主使用jmeter来测试，博主默认大家都会使用（不会使用的童鞋需要学习喽）。

jmeter准备工作

在jmeter中设置50个线程

在该线程下设置两个接口，分别请求80和81

redis准备工作

在redis中设置一对键值 num

至此，就可以在jmeter中开启请求了

测试结果

我们先来看redis中num的值

我们再分别查看80和81的日志

总结

本文讲述了利用redis实现分布式锁的原理，分布式锁本质上是将并发请求按顺序处理，那么这把锁就成为了所有请求的瓶颈，如何打破锁的瓶颈呢？敬请关注博主，后续填坑（博主挖坑必填）。

本文留下的两个坑：

1，为了使redis高可用，redis集群后，如何解决redis端因为网络问题导致锁不同步问题？

2，分布式锁实现了并发排队，锁成为了性能瓶颈，如何提高性能？

巴特西

Redis 分布式锁（一）

前言

概念

作用

原理