前言

今天给大家带来一篇关于分布式锁的好文，关于分布式系统下的内容，博主已经写了好几篇了，也收获了大家的不少好评。分布式系统在目前的开发中所占的比重还是比较大的，如果你还没接触过分布式系统，那么欢迎你去学习博主的微服务专栏，如果你对微服务略知一二，那么博主近期的挖祖坟系列就很适合你。让我们一起在Java这方土地上挖呀挖呀挖吧！

分布式锁

锁概念

代码中的锁其实和我们现实生活中很像，我们使用锁的目的是为了保证人身和财产的安全，这和在代码中的使用其实本质是一样的。

锁的使用是，我们希望的情况是：谁上的锁，由谁来打开，这样是最安全的情况，不管是外面上锁，还是进入室内上锁，都是如此。

在代码中，当多个线程同时才做一个变量的时候，就需要过程是同步的，操作是互斥的，而其本质就是锁的思想。

分布式锁概念

任何一个分布式系统都无法同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance），最多只能同时满足两项。这个我们在讲分布式事务的时候就已经给大家讲过了，关于CAP我不允许你还不知道，传送门：Java开发 - 不知道算不算详细的分布式事务详解

分布式锁要做的事情就是在多个线程同时去修改数据的时候保证数据的安全，而同一时间只能有一个线程操作此数据就是我们所说的安全。操作时上锁，用完解锁，把下一个获取锁的机会留给其他人。

这让博主想到了生活中一个比较有味道的事情：蹲坑！谁也不希望在自己蹲坑的时候其他人可以进来和你一起方便吧？这个案例和分布式锁的思想高度吻合，有木有？有木有？哈哈哈哈～～

分布式锁的特点

分布式锁的特点其实很简单，很好理解，我们来列举一下：

互斥性：同一时刻只能有一个服务在访问资源
原子性：一致性要求保证加锁和解锁的行为是原子的，原子性指最小颗粒，不能被中断的一个或一系列操作
安全性：谁上锁，谁释放
容错性：当持有锁的服务崩溃时，锁仍能被释放，这样才能避免死锁
高可用：高可用一般我们用来说明分布式系统，此处表示获取锁和释放锁是安全稳定的，也包括具有容错的能力
高性能：获取锁和释放锁的性能要好，因为一些操作属于低效锁，下面会讲
持久性：锁按业务需要自动续约/自动延期

分布式锁的应用场景

关于分布式锁的使用场景，简单列举几个比较常见的：

银行取款和手机转账同时进行，你看看能不能能不能获取双倍的钱，显然不能，就需要分布式锁
集群下不同的集群的用户去进行商品的秒杀，商品库存只有一个，那么显然只能有一个人秒杀成功，就需要用到分布式锁

我相信大家完全能想象到这个过程，所以关于其流程图就不再画出来了。

分布式锁的执行过程

虽然博主非常不想去画图，但是这里必须要有图了，文字再多，都不及一张逻辑清晰的流程图，下面，我们用一张图来说明分布式锁的执行流程：

看完有没有很简单，和你想的一样吗？不用把它想的那么难，其实就很简单啊！！！

数据库实现分布式锁

关于实现，其实还是要根据具体的业务来做的，所以博主不仅是这里，包括下面，如果涉及具体的业务，就不写具体代码了，如果是通用的，回忆代码的形式展现给大家。

表锁实现

根据表锁实现大致可分为以下几个步骤：

创建一张表，不需要有太多的参数，比如：

CREATE TABLE tb (
  check_no varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
  PRIMARY KEY (check_no) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

SET FOREIGN_KEY_CHECKS = 1;

SET FOREIGN_KEY_CHECKS=0; //关闭外键约束检查

SET FOREIGN_KEY_CHECKS=1; //开启外键约束检查
访问数据时，将一个编号存入到这张表中
只要insert成功，就代表获取锁成功，反之获取锁失败
由于主键冲突，一张表中只能存入一个相同的编号，锁就这样产生了
获取锁成功的程序在逻辑执行完以后，删除该编号的数据，就代表释放锁了，这是其他的线程就可以来竞争这个锁，谁insert成功，谁就获取了锁，可以执行自己的任务

这张表是博主看过别人创建的，个人感觉参数设置略复杂了，从各方面来说，一张简单的表就足以实现了，希望和大家一起讨论下原因，还请大家不吝赐教。

条件实现

根据条件查询是以行锁为前提的，这个条件和下面的版本号有异曲同工之妙，先来说说条件查询。我们先来看看下面这句SQL：

update tb_stock set stock=stock-#{saleNum} where id = #{id} and stock-#{saleNum}>=0"

大家应该知道，在innoDB中，update操作就是行锁的标志，所以当使用了行锁，我不管有多少个线程去操作这条数据，但同一时间只能有一个线程在操作，总库存减去购买数量后的剩余库存一定是要>=0的，这就保证了商品不会超卖。这是乐观锁的思想，但乐观锁并不是不加锁，他只是一种思想，仅此而已，你只需要这种方法可行。乐观锁有两种思想，就是版本号和CAS，显然这里是CAS的做法。

关于乐观锁和悲观锁，博主就不细说了，有不了解的点击此链接查看了解：Java开发 - 数据库中的基本数据结构

版本号实现

版本号的实现和条件很像，当没有判断依据的时候，就可以添加一个version字段，我们来看以下SQL：

update tb_user set name=#{name},version=version+1 where id=#{id} and version=#{version}

看完之后，其实感觉版本号和条件原理是一模一样的，都是基于行锁原理来完成的，但他们存在一定问题，我们在下面来说。

数据库实现分布式锁缺点

从以上我们不难看出，它的缺点很明显，首先就是频繁的操作数据库，这一点肯定不好，另外，分布式锁还需要具备可重入，失效时间等特性，不仅如此，数据库本身的业务也很繁忙，性能上多少还是存在一定的问题。

以上问题也并不是不能解决，但解决这些问题，实现方式将会越来越复杂，性能问题也不得不考虑。如果是单机系统还可以考虑，既然都上分布式了，那何必使用这种方式呢？

zookeeper实现分布式锁

实现原理

zookeeper能够实现分布式锁得益于其设计者的巧妙设计，主要是使用了zookeeper的znode节点特性和watch机制来完成。

znode节点

说起来也很简单，znode节点共分为两大类，分别是：临时节点和持久节点，他们各自又分为两类：分别是有序节点和无序节点，看下方详细说明：

持久节点：一旦创建，永久存在于zookeeper中，除非手动删除
持久有序节点：一旦创建，永久存在于zookeeper中，除非手动删除。不同点是每个节点都默认存在节点序号，且是有序递增的
临时节点：节点创建后，一旦服务器重启或宕机，则被自动删除
临时有序节点：节点创建后，一旦服务器重启或宕机，则被自动删除。不同的地方是每个节点默认存在节点序号，且有序递增

watch监听机制

watch监听机制主要用于监听节点的状态变更，并用于触发后续事件，当B节点监听A节点时，一旦A节点发生修改、删除、子节点列表发生变更等事件，B节点则会收到A节点改变的通知，接着完成其他任务，在这里指B监听到A释放了锁，就尝试去获取锁。

工作方式

虽然有序节点是有序的，但在创建第一个节点前，需要先创建一个父节点，然后在父节点下创建临时有序节点，之所以使用临时节点是为了保证服务器宕机或重启后，原先的锁能够被释放。随着节点的递增，那么最先获取到的锁的一定是序号最小的节点，任务完成后，删除该临时节点，接着去寻找下一个最小的节点，并且，监听机制是按照后一个监听前一个的规则来工作的，比如，B监听A，C监听B，D监听C，E监听D......以此类推，前一个完成后就会通知下一个节点去获取锁并执行任务，如下图：

锁思想实现（低效锁）

使用zookeeper时，有一种低效的实现方法也是很常见的，它的原理是只使用一个锁节点，当创建锁节点时，如果锁节点不存在，则创建成功，代表当前线程获取到锁，如果创建锁节点失败，代表已经有其他线程获取到锁，则该线程会监听锁节点的释放，释放就是删除此节点。当锁节点释放后，其他线程则继续尝试创建锁节点并加锁。

下面我们来做个模拟

创建一个新项目，pom文件如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.6.0</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.codingfire</groupId>
    <artifactId>zookeeper_lock</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>zookeeper_lock</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!--zookeeper-->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.5.5</version>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>com.101tec</groupId>
            <artifactId>zkclient</artifactId>
            <version>0.10</version>
            <exclusions>
                <exclusion>
                    <artifactId>slf4j-log4j12</artifactId>
                    <groupId>org.slf4j</groupId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.6.0</version>
            </plugin>
        </plugins>
    </build>

</project>

创建一个抽象类用于对外提供加解锁方法：

package com.codingfire.zookeeper_lock;

import org.I0Itec.zkclient.ZkClient;

public abstract class AbstractLock {

    //zookeeper服务器地址
    public static final String ZK_SERVER_ADDR="localhost:2181";

    //zookeeper超时时间
    public static final int CONNECTION_TIME_OUT=30000;
    public static final int SESSION_TIME_OUT=30000;

    //创建zk客户端
    protected ZkClient zkClient = new ZkClient(ZK_SERVER_ADDR,SESSION_TIME_OUT,CONNECTION_TIME_OUT);

    /**
     * 获取锁
     * @return
     */
    public abstract boolean tryLock();

    /**
     * 等待加锁
     */
    public abstract void waitLock();

    /**
     * 释放锁
     */
    public abstract void releaseLock();

    //加锁实现
    public void getLock() {

        String threadName = Thread.currentThread().getName();

        if (tryLock()){
            System.out.println(threadName+":   获取锁成功");
        }else {

            System.out.println(threadName+":   等待获取锁");
            waitLock();
            getLock();
        }
    }
}

创建实现抽象方法的子类：

package com.codingfire.zookeeper_lock;

import org.I0Itec.zkclient.IZkDataListener;

import java.util.concurrent.CountDownLatch;

public class LowLock extends AbstractLock {

    private static final String LOCK_NODE="/lock_node";

    private CountDownLatch countDownLatch;

    @Override
    public boolean tryLock() {
        if (zkClient == null){
            return false;
        }

        try {
            zkClient.createEphemeral(LOCK_NODE);
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    @Override
    public void waitLock() {

        //注册监听
        IZkDataListener listener = new IZkDataListener() {

            //节点数据改变触发
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {

            }

            //节点删除触发
            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                if (countDownLatch != null){
                    countDownLatch.countDown();
                }
            }
        };
        zkClient.subscribeDataChanges(LOCK_NODE,listener);

        //如果节点存在，则线程阻塞等待
        if (zkClient.exists(LOCK_NODE)){
            countDownLatch = new CountDownLatch(1);
            try {
                countDownLatch.await();
                System.out.println(Thread.currentThread().getName()+":  等待获取锁");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        //节点不存在，删除监听
        zkClient.unsubscribeDataChanges(LOCK_NODE,listener);

    }

    @Override
    public void releaseLock() {
        System.out.println(Thread.currentThread().getName()+":    释放锁");
        zkClient.delete(LOCK_NODE);
        zkClient.close();

    }
}

接着我们在启动类的main方法中来测试下以上代码：

package com.codingfire.zookeeper_lock;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.util.concurrent.TimeUnit;

@SpringBootApplication
public class ZookeeperLockApplication {

    public static void main(String[] args) {

        //模拟多个10个客户端
        for (int i = 0; i < 20; i++) {
            Thread thread = new Thread(new LockRunnable());
            thread.start();
        }
    }

    private static class LockRunnable implements Runnable {
        @Override
        public void run() {

            AbstractLock abstractLock = new LowLock();

            abstractLock.getLock();

            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            abstractLock.releaseLock();
        }
    }

}

然后就可以运行测试了，不出意外，任务是一个个执行的，由于输出的时候还是打印了zookeeper的debug日志，导致页面很乱，所以就不截图了，大家自己运行看看结果就行，这日志还去不掉了，明明已经去掉了打印日志的工具：

<!--zookeeper-->
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.5.5</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
        </exclusion>
        <exclusion>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
        </exclusion>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.10</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
        </exclusion>
    </exclusions>
</dependency>

有知道原因的童鞋可以给博主留言。刚刚我们的代码是基于单节点实现的锁，那么就会导致一个问题：一旦锁释放，其他等待线程都会收到通知，并去竞争唯一的一个锁，这种由于被watch的znode节点的变化，而造成大量的通知操作就叫做羊群效应，会严重降低zookeeper性能。所以这种低效锁做法一般来说我们在分布式系统中不会使用。

zookeeper高效锁实现

为了解决低效锁的羊群问题，我们会让这些任务排队，这就会用到有序节点。有序节点中，最小的节点获取锁并执行任务，执行完毕后释放锁（删除节点），下一个watch的最小节点就会收到通知并获取锁，避免竞争，这就是高效锁，高效锁的实现过程如下：

高效锁的工具类也需要继承自抽象类：

package com.codingfire.zookeeper_lock;

import org.I0Itec.zkclient.IZkDataListener;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class HighLock extends AbstractLock {

    private static  String PARENT_NODE_PATH="";

    public HighLock(String parentNodePath){
        PARENT_NODE_PATH = parentNodePath;
    }

    //当前节点路径
    private String currentNodePath;

    //前一个节点的路径
    private String preNodePath;

    private CountDownLatch countDownLatch;

    @Override
    public boolean tryLock() {

        //判断父节点是否存在
        if (!zkClient.exists(PARENT_NODE_PATH)){
            //父节点不存在，创建持久节点
            try {
                zkClient.createPersistent(PARENT_NODE_PATH);
            } catch (Exception e) {
            }
        }

        //创建第一个临时有序节点
        if (currentNodePath==null || "".equals(currentNodePath)){
            //在父节点下创建临时有序节点
            currentNodePath =  zkClient.createEphemeralSequential(PARENT_NODE_PATH+"/","lock");
        }

        //不是第一个临时有序节点
        //获取父节点下的所有子节点列表
        List<String> childrenNodeList = zkClient.getChildren(PARENT_NODE_PATH);

        //因为有序号，所以进行升序排序
        Collections.sort(childrenNodeList);

        //判断是否加锁成功，当前节点是否为父节点下序号最小的节点
        if (currentNodePath.equals(PARENT_NODE_PATH+"/"+childrenNodeList.get(0))){
            //当前节点是序号最小的节点
            return true;
        }else {
            //当前节点不是序号最小的节点，获取其前置节点，并赋值
            int length = PARENT_NODE_PATH.length();
            int currentNodeNumber = Collections.binarySearch(childrenNodeList, currentNodePath.substring(length + 1));
            preNodePath = PARENT_NODE_PATH+"/"+childrenNodeList.get(currentNodeNumber-1);
        }

        return false;
    }

    @Override
    public void waitLock() {

        //注册监听
        IZkDataListener listener = new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {

            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                if (countDownLatch != null){
                    countDownLatch.countDown();
                }
            }
        };
        zkClient.subscribeDataChanges(preNodePath,listener);

        //判断前置节点是否存在，存在则阻塞
        if (zkClient.exists(preNodePath)){

            countDownLatch = new CountDownLatch(1);
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        //删除监听
        zkClient.unsubscribeDataChanges(preNodePath,listener);

    }


    @Override
    public void releaseLock() {

        zkClient.delete(currentNodePath);
        zkClient.close();
    }
}

调用方式如下：

    //获取当前方法名
    String methodName = Thread.currentThread().getStackTrace()[1].getMethodName();
    AbstractLock lock = new HighLock("/"+methodName);

    try {
        //加锁
        lock.getLock();

        //执行操作
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        //释放锁
        lock.releaseLock();
    }

就是这么简单，以上也基本是业内普遍使用的封装类，差别不会很大，可直接使用。在测试前，请确保你的zookeeper是开启的。

Redis实现分布式锁

单节点Redis锁

如果是单节点的redis，那么分布式锁的实现就相当简单了，因为redis本身是单线程的，基于这个特性，在并发情况下，各个请求都需要排队才能进入，同一时间只有一个线程可以进入并获取到锁。

redis实现分布式锁基于三个核心API：

setNx()：向redis中存key-value，只有当key不存在时才会设置成功，否则返回0，体现了其互斥性
expire()：设置key的过期时间，用于避免死锁出现
delete()：删除key，用于释放锁

加锁

通过jedis.set进行加锁，如果返回值是OK，代表加锁成功，否则加锁失败，失败就自旋不断尝试获取锁，同时在一定时间内如果仍没有获取到锁，则退出自旋，不再尝试获取锁。加锁时需要使用 requestId来标识每个线程所持有的锁标记。

package com.codingfire.zookeeper_lock;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.params.SetParams;

public class SingleRedisLockUtil {

    JedisPool jedisPool = new JedisPool("localhost",6379);

    //锁过期时间
    protected long internalLockLeaseTime = 30000;

    //获取锁的超时时间
    private long timeout = 999999;

    /**
     * 加锁
     * @param lockKey 锁键
     * @param requestId 请求唯一标识
     * @return
     */
    SetParams setParams = SetParams.setParams().nx().px(internalLockLeaseTime);

    public boolean tryLock(String lockKey, String requestId){

        String threadName = Thread.currentThread().getName();

        Jedis jedis = this.jedisPool.getResource();

        Long start = System.currentTimeMillis();

        try{
            for (;;){
                String lockResult = jedis.set(lockKey, requestId, setParams);
                if ("OK".equals(lockResult)){
                    System.out.println(threadName+":   获取锁成功");
                    return true;
                }
                //否则循环等待，在timeout时间内仍未获取到锁，则获取失败
                System.out.println(threadName+":   获取锁失败，等待中");
                long l = System.currentTimeMillis() - start;
                if (l>=timeout) {
                    return false;
                }
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }finally {
            jedis.close();
        }

    }
}

在创建类之前，你还需要添加几个依赖，否则会报错：

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>3.0.1</version>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>
<!--Redis分布式锁-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.13.1</version>
</dependency>

解锁

解锁也是有些学问的，要避免当前线程将别人的锁释放掉，当线程A加锁成功，一段时间后线程A来解锁，线程A的锁已经过期了，此时线程B也来加锁，因为线程A的锁已经过期，所以线程B可加锁成功。此时，线程A就有可能将线程B的锁给释放掉。这也是为什么加锁时要引入requestId的原因。所以当解锁时要判断当前锁键的value与传入的value是否相同，相同的话，则代表是同一个人，可以解锁。否则不能解锁。

解锁时，如果先查询做对比，比对后相同再删除看着没错，但却有个坑，那就是原子性被忽略了，原子性要求我们这一系列的操作必须是一步完成的，否则在中间容易出现变故，所以查询和删除通常会用lua脚本完成。下面我们看看代码该怎么写：

在工具类中添加解锁方法：

/**
     * 解锁
     * @param lockKey 锁键
     * @param requestId 请求唯一标识
     * @return
     */
public boolean releaseLock(String lockKey,String requestId){

    String threadName = Thread.currentThread().getName();
    System.out.println(threadName+"：释放锁");
    Jedis jedis = this.jedisPool.getResource();

    String lua =
        "if redis.call('get',KEYS[1]) == ARGV[1] then" +
        "   return redis.call('del',KEYS[1]) " +
        "else" +
        "   return 0 " +
        "end";

    try {
        Object result = jedis.eval(lua, Collections.singletonList(lockKey),
                                   Collections.singletonList(requestId));
        if("1".equals(result.toString())){
            return true;
        }
        return false;
    }finally {
        jedis.close();
    }

}

测试

在测试前，请确保你的Redis服务是开启的，然后运行如下代码：

package com.codingfire.zookeeper_lock;

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

public class RedisTest {
    public static void main(String[] args) {

        //模拟多个5个客户端
        for (int i=0;i<5;i++) {
            Thread thread = new Thread(new LockRunnable());
            thread.start();
        }
    }

    private static class LockRunnable implements Runnable {
        @Override
        public void run() {

            SingleRedisLockUtil singleRedisLock = new SingleRedisLockUtil();

            String requestId = UUID.randomUUID().toString();
            boolean lockResult = singleRedisLock.tryLock("lock", requestId);
            if (lockResult){

                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            singleRedisLock.releaseLock("lock",requestId);
        }
    }
}

控制台输出一角：

没办法截图特别全，大家看下就明白了。

单节点问题

以上的单节点解决方案也只不过是一个基础的解决方案，还存在一定的问题，如果锁过期就会自动释放，此时如果任务还没执行完该怎么办？这时候同一个任务可能会重复执行，所以对于锁的超市时间的设置就比较讲究了，但是又没办法预测任务的时间长短，最好是能在锁快过期的时候动态延长时间，因为锁在还没过期时还没释放一定是任务没有执行完，这点毋庸置疑。

最好的办法就是创建一个守护线程，同时定义一个定时任务每隔一段时间去为未释放的锁增加过期时间。当业务执行完，释放锁后，再关闭守护线程。这就解决了锁续期的问题。

单节点虽好，但Redis也是有可能存在集群的吧？毕竟Redis也是数据库的一种，为了数据的安全，缓存提升查询的效率，我做个主从啥的不过分吧？那我们就接着往下看。

Redisson分布式锁

Redisson是redis官网推荐实现分布式锁的一个第三方类库，其功能非常强大，对各种锁都有实现，使用也非常简单，Redisson实现单节点的分布式锁相对而言更为容易，直接基于lock()&unlock()方法操作即可。

下面我们就先来看看Redisson的单节点分布式锁。

Redisson单节点分布式锁

依赖少不了：

<!--Redis分布式锁-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.13.1</version>
</dependency>

配置文件安排上：

server:
  redis:
    host: localhost
    port: 6379
    database: 0
    jedis:
      pool:
        max-active: 500
        max-idle: 1000
        min-idle: 4

启动类中添加如下代码，最终的启动类如下：

package com.codingfire.zookeeper_lock;

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

import java.util.concurrent.TimeUnit;

@SpringBootApplication
public class ZookeeperLockApplication {

    @Value("${spring.redis.host}")
    private String host;

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

    @Bean
    public RedissonClient redissonClient(){
        RedissonClient redissonClient;

        Config config = new Config();
        String url = "redis://" + host + ":" + port;
        config.useSingleServer().setAddress(url);

        try {
            redissonClient = Redisson.create(config);
            return redissonClient;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    public static void main(String[] args) {

        SpringApplication.run(ZookeeperLockApplication.class,args);
    }


}

编写加解锁工具类：

package com.codingfire.zookeeper_lock;

import org.redisson.api.RLock;

@Component
public class RedissonLock {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 加锁
     * @param lockKey
     * @return
     */
    public boolean addLock(String lockKey){

        try {
            if (redissonClient == null){
                System.out.println("redisson client is null");
                return false;
            }

            RLock lock = redissonClient.getLock(lockKey);

            //设置锁超时时间为5秒，到期自动释放
            lock.lock(10, TimeUnit.SECONDS);

            System.out.println(Thread.currentThread().getName()+":  获取到锁");

            //加锁成功
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean releaseLock(String lockKey){

        try{
            if (redissonClient == null){
                System.out.println("redisson client is null");
                return false;
            }

            RLock lock = redissonClient.getLock(lockKey);
            lock.unlock();
            System.out.println(Thread.currentThread().getName()+":  释放锁");
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }
}

测试环节：

本来想把测试写在main方法里，但是发现静态方法调用不了外部非静态方法，还是算了，创建了一个测试类RedissonLockTest：

package com.codingfire.zookeeper_lock;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

@SpringBootTest
@RunWith(SpringRunner.class)
public class RedissonLockTest {

    @Autowired
    private RedissonLock redissonLock;

    @Test
    public void easyLock(){
        //模拟多个5个客户端
        for (int i = 0; i < 5; i++) {
            Thread thread = new Thread(new LockRunnable());
            thread.start();
        }

        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private class LockRunnable implements Runnable {
        @Override
        public void run() {
            redissonLock.addLock("lock");
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            redissonLock.releaseLock("lock");
        }
    }
}

查看控制台输出：

五个线程按顺序执行完成，测试成功。

看门狗机制

虽然Redisson单节点实现分布式锁更方便，但是也存在一定的问题，我们在使用锁的时候应该看到了设置过期时间，大家有印象没？过期时间！！！和上面zookeeper中一样的问题，zookeeper使用了守护线程来给锁续期，这里我们使用的是看门狗机制。

看门狗机制用起来比守护线程更简便，分两步完成：

不设置过期时间，默认是30s
将加锁的方式由lock()改为tryLock()

另外，过期时间可修改：

config.setLockWatchdogTimeout(3000L);

分布式系统与红锁算法

单节点的Redis一旦出现宕机的情况锁就无法操作，所以像主从这样的配置是免不了的。但是主从时也存在一定的问题：master异步将数据复制到slave中，当某个线程持有了锁，在还没有将数据复制到slave时，master宕机，此时slave会被提升为master，新的master没有之前线程的锁信息，那么其他线程则又可以重新加锁，这该怎么办呢？好在有个红锁算法可以解决。

红锁是一种基于多节点redis实现分布式锁的算法，可以有效解决redis单点故障的问题。官方建议搭建五台redis服务器对红锁算法进行实现，就是比较烧钱～～～

红锁算法关于过期时间的计算非常严格，好在这些不需要我们亲自去计算，博主这里就偷个懒，不讲了，但是我们要确保在使用Redis的时候开启AOF，若Redis宕机，在ttl时间后再重启，缺点就是ttl时间内Redis无法对外提供服务，但在多节点情况下也是可接受的。

红锁算法实现

创建配置类：

package com.codingfire.zookeeper_lock;

import org.redisson.Redisson;
import org.redisson.RedissonRedLock;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RedissonRedLockConfig {

    public RedissonRedLock initRedissonClient(String lockKey){

        Config config1 = new Config();
        config1.useSingleServer().setAddress("redis://ip:8000").setDatabase(0);
        RedissonClient redissonClient1 = Redisson.create(config1);

        Config config2 = new Config();
        config2.useSingleServer().setAddress("redis://ip:8001").setDatabase(0);
        RedissonClient redissonClient2 = Redisson.create(config2);

        Config config3 = new Config();
        config3.useSingleServer().setAddress("redis://ip:8002").setDatabase(0);
        RedissonClient redissonClient3 = Redisson.create(config3);

        Config config4 = new Config();
        config4.useSingleServer().setAddress("redis://ip:8003").setDatabase(0);
        RedissonClient redissonClient4 = Redisson.create(config4);

        Config config5 = new Config();
        config5.useSingleServer().setAddress("redis://ip:8004").setDatabase(0);
        RedissonClient redissonClient5 = Redisson.create(config5);

        RLock rLock1 = redissonClient1.getLock(lockKey);
        RLock rLock2 = redissonClient2.getLock(lockKey);
        RLock rLock3 = redissonClient3.getLock(lockKey);
        RLock rLock4 = redissonClient4.getLock(lockKey);
        RLock rLock5 = redissonClient5.getLock(lockKey);

        RedissonRedLock redissonRedLock = new RedissonRedLock(rLock1,rLock2,rLock3,rLock4,rLock5);

        return redissonRedLock;
    }
}

由于需要多台Redis服务器，大家可以通过Docker本地模拟。

创建新的测试类进行测试：

package com.codingfire.zookeeper_lock;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.redisson.RedissonRedLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

@SpringBootTest
@RunWith(SpringRunner.class)
public class RedLockTest {

    @Autowired
    private RedissonRedLockConfig redissonRedLockConfig;

    @Test
    public void testRedLock(){
        //模拟多个5个客户端
        for (int i = 0; i < 5; i++) {
            Thread thread = new Thread(new RedLockTest.RedLockRunnable());
            thread.start();
        }

        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private class RedLockRunnable implements Runnable {
        @Override
        public void run() {
            RedissonRedLock redissonRedLock = redissonRedLockConfig.initRedissonClient("demo");

            try {
                boolean lockResult = redissonRedLock.tryLock(100, 10, TimeUnit.SECONDS);

                if (lockResult){
                    System.out.println("获取锁成功");
                    TimeUnit.SECONDS.sleep(3);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                redissonRedLock.unlock();
                System.out.println("释放锁");
            }
        }
    }
}

红锁原理

红锁算法允许加锁失败节点的个数限制为（N-(N/2+1)），加入当前假设五个节点，则允许失败节点数为2，不能超过一半，加锁时遍历所有节点执行lua脚本加锁，保证原子性。如果获取锁成功，就添加到已获取锁集合中，如果抛出异常，为了防止其他节点加锁成功，需要解锁所有节点，加锁失败。若是加锁失败，就没有必要继续从后面的节点申请锁，红锁算法要求至少N/2+1 个节点都加锁成功，才算最终的锁申请成功。此时，会计算从各个节点获取锁已经消耗的总时间，如果大于等于最大等待时间，申请锁失败，否则，申请锁成功，分布式锁加锁成功。

结语

看，分布式锁其实也没有那么难嘛，如果你能看到这里，那么相信你一定会有所收获，多看几遍，对理解分布式锁有好处，学完分布式锁，又可以开心的出去装13了，真开心！！！学习是一个循序渐进的过程，学完之后不代表就会了，过一段时间你多半是会忘记的，就好比博主，写完之后过一段很多东西都会忘记，也需要不断的回顾才能加深印象，学习是投资自己，只有金刚钻硬了，才能去揽那瓷器活儿。加油吧，少年们！