基于Redis+ZooKeeper实现分布式锁

以你之姓@ 2022-04-04 11:07 339阅读 0赞

1>  分布式锁一般有三种实现方式：

1. 数据库乐观锁,悲观锁；（很少用）

要实现分布式锁，最简单的方式可能就是直接创建一张锁表，然后通过操作该表中的数据来实现了。当我们要锁住某个方法或资源的时候，我们就在该表中增加一条记录，想要释放锁的时候就删除这条记录。

2. 基于Redis的分布式锁；（用的最多，推荐）

3. 基于ZooKeeper的分布式锁（其次）

2> 为什么要使用分布式锁

我们在开发应用的时候，如果需要对某一个共享变量进行多线程同步访问的时候，可以使用我们学到的Java多线程的18般武艺进行处理，并且可以完美的运行，毫无Bug！

注意这是单机应用，也就是所有的请求都会分配到当前服务器的JVM内部，然后映射为操作系统的线程进行处理！而这个共享变量只是在这个JVM内部的一块内存空间！

后来业务发展，需要做集群，一个应用需要部署到几台机器上然后做负载均衡，大致如下图  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70][]

上图可以看到，变量A存在JVM1、JVM2、JVM3三个JVM内存中（这个变量A主要体现是在一个类中的一个成员变量，是一个有状态的对象，例如：UserController控制器中的一个整形类型的成员变量），如果不加任何控制的话，变量A同时都会在JVM分配一块内存，三个请求发过来同时对这个变量操作，显然结果是不对的！即使不是同时发过来，三个请求分别操作三个不同JVM内存区域的数据，变量A之间不存在共享，也不具有可见性，处理的结果也是不对的！

如果我们业务中确实存在这个场景的话，我们就需要一种方法解决这个问题！

为了保证一个方法或属性在高并发情况下的同一时间只能被同一个线程执行，在传统单体应用单机部署的情况下，可以使用Java并发处理相关的API(如ReentrantLock或Synchronized)进行互斥控制。在单机环境中，Java中提供了很多并发处理相关的API。但是，随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题！

3> 分布式锁应该具备哪些条件

1、在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行；   
     2、高可用的获取锁与释放锁；   
     3、高性能的获取锁与释放锁；   
     4、具备可重入特性；   
     5、具备锁失效机制，防止死锁；   
     6、具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败。

4> 核心思路：在多台服务器集群中，只能够保证一个jvm进行做操作。

### **一. 基于Redis实现分布式锁** ###

RedisTool，分布式锁工具类，有上锁和解锁方法

@Component
    public class RedisLockTool {
        private String redisLockKey = "redis_lock";
        @Autowired
        private StringRedisTemplate stringRedisTemplate;
        /**
         * 获取redis锁
         * @param acquireTimeout 在获取锁之前的超时时间，在尝试获取锁的时候，如果在规定的时间内还没有获取锁，直接放弃
         * @param timeOut 在获取锁之后的超时时间，当获取锁成功之后，对应的key有对应的有效期，超时则失效
         * @return
         */
        public String getRedisLock(Long acquireTimeout, Long timeOut) {
            // 1.定义 redis 对应key 的value值( uuid)  作用:释放锁 随机生成value
            String identifierValue = UUID.randomUUID().toString();
            // 2.定义在获取锁之后的超时时间(以秒为单位,所以除以1000)
            int expireLock = (int) (timeOut / 1000);
            // 3.定义在获取锁之前的超时时间
            Long endTime = System.currentTimeMillis() + acquireTimeout;//也就是当前时间后的acquireTimeout毫秒
            /* 使用循环机制，如果没有获取到锁，要在规定acquireTimeout时间 保证重复进行尝试获取锁（相当于乐观锁）*/
            while (System.currentTimeMillis() < endTime) {//如果当前系统时间<获取锁之前的超时时间
                // 4.使用setNx命令插入对应的redisLockKey，如果返回为true，则代表成功获取锁
                /* 如果不等于1，则继续循环并判断超时结束时间endTime，一旦当前时间大于endTime，则循环退出，返回null
                   如果等于1，插入成功，返回对应锁的value */
                if (stringRedisTemplate.opsForValue().setIfAbsent(redisLockKey, identifierValue)) {
                    // 设置对应key的有效期
                    stringRedisTemplate.expire(redisLockKey, expireLock, TimeUnit.SECONDS);
                    return identifierValue;
                }
            }
            return null;
        }
        // 释放redis锁
        public void unRedisLock(String identifierValue) {
            String redisValue = stringRedisTemplate.opsForValue().get(redisLockKey);
            // 如果redis中的value等于传入的identifierValue，则可以删除
            if (StringUtils.equals(redisValue, identifierValue)) {
                stringRedisTemplate.delete(redisLockKey);
            }
        }
        /*
          获取锁：
            为什么获取锁之后，还要设置锁的超时时间 目的是为了防止死锁
          释放锁：
            如果直接使用 stringRedisTemplate.delete(key)，则会出现一个问题，a获取到锁，b给删除了（这里是课程讲的，但原则上a在释放锁之前，b是获取不到的）
            保证对应是自己的创建redisLockKey,删除对应自己的
         */
    }

DAO层，主要用于更新库存和查新商品信息

@Mapper
    public interface RedisLockDAO {
    	
    	@Update("update product_lock set stock=stock-1 where stock>0 and id=#{id}")
    	int updateStock(@Param("id") Integer id);
    	
    	@Select("select * from product_lock where id=#{id}")
    	Map<String,Object> getById(@Param("id") Integer id);
    
    }

写个controller测试一下

@RestController
    public class RedisLockController {
    
        @Autowired
        private StringRedisTemplate stringRedisTemplate;
    
        @Autowired
        private RedisLockDAO redisLockDAO;
    
        @Autowired
        private RedisLockTool redisLockTool;
    
        @PostMapping("/redisLock")
        public ResultBO<?> redisLock() {
            int row = 0;
            String identifierValue = redisLockTool.getRedisLock(3000L, 3000L);// 获取锁
    
            if (identifierValue == null) {
                System.out.println("获取锁失败，原因:获取锁时间超时...");
            }
    
            Map<String, Object> map = redisLockDAO.getById(1);
            if (Integer.valueOf(map.get("stock").toString()) > 0) {
                row = redisLockDAO.updateStock(1);
            }
            redisLockTool.unRedisLock(identifierValue);
            return ResultTool.success(row);
        }
    }

下面是数据库表product\_lock

![20181216221225596.png][]

感兴趣的同学可以用Jmeter测试一下并发，会发现无论多少线程都不会出现库存为-1的情况

6> 注意事项

删除(解锁)的时候，要注意一点，一定要定义一个value(可以当成锁的id)，在删除的时候确保删除自己的，不能把别人的删掉

### 二. 基于ZooKeeper实现分布式锁 ###

这里我们基于模板方法设计模式去设计我们的分布式锁，可以提供一套分布式锁的全方位模板（数据库，redis，zk方式）

1. 引入zk依赖

<dependency>
        <groupId>com.101tec</groupId
        <artifactId>zkclient</artifactId>
        <version>0.8</version>
    </dependency>

2. 代码实现

定义统一接口，有获取锁和释放锁两个方法

public interface Lock {
        /**
         * 获取锁
         */
        public void getLock();
        /**
         * 释放锁
         */
        public void unLock();
    }

定义一个抽象类，相同的方法自己写（不是抽象方法），不同的写个抽象方法留给子类实现(抽象方法)

abstract class AbstractTemplzateLock implements Lock {
        @Override
        public void getLock() {
            // 模版方法 定义共同抽象的骨架
            if (tryLock()) {
                System.out.println(">>>" + Thread.currentThread().getName() + ",获取锁成功");
            } else {
                // 开始实现等待
                waitLock(); // 事件监听
                // 重新获取（递归）
                getLock();
            }
        }
    
        @Override
        public void unLock() {
            unImplLock();
        }
    
        /**
         * 获取锁
         * @return
         */
        protected abstract boolean tryLock();
        /**
         * 等待锁
         * @return
         */
        protected abstract void waitLock();
        /**
         * 释放锁
         * @return
         */
        protected abstract void unImplLock();
    }

编写具体实现：

public class ZkTemplateImplLock extends AbstractTemplzateLock {
    
        // 参数1 连接地址
        private static final String ADDRES = "192.168.0.8:2181";
        // 参数2 zk超时时间
        private static final int TIMEOUT = 5000;
    
        // 创建我们的zk连接
        private ZkClient zkClient = new ZkClient(ADDRES, TIMEOUT);
    
        // 临时节点路径
        private String lockPath = "/lockPath";
        // 计数器
        //private CountDownLatch countDownLatch = null;
        private CountDownLatch countDownLatch;
    
    
        @Override
        protected boolean tryLock() {
            try {
                // createEphemeral方法返回结果为void，那么怎么知道已经重复了？ 答案：try catch异常 （如果创建重复，那么会走异常）
                zkClient.createEphemeral(lockPath); //【临时节点】
                return true;
            } catch (Exception e) {
                // 如果创建的节点已经存在则返回false，走抽象类getLock方法的else，重新等待，重新获取
                return false;
            }
        }
    
        @Override
        protected void waitLock() {
            // 1.使用【事件监听】监听lockPath节点是否已经被删除，如果被删除，则调用countDownLatch.countDown();
            IZkDataListener iZkDataListener = new IZkDataListener() {
                @Override
                public void handleDataChange(String s, Object o) {
    
                }
                @Override
                // 监听该节点是否被删除（删除了才会走） 【一旦删除，则countDown，唤醒，直接走到await后】
                public void handleDataDeleted(String s) {
                    if (countDownLatch != null) {
                        countDownLatch.countDown(); // -1 （计数器变为0，一旦变为0，则countDownLatch.await()就解放了）
                    }
                }
            };
            // 事件通知（发布订阅）
            zkClient.subscribeDataChanges(lockPath, iZkDataListener);
            // 2.使用countDownLatch等待
            //if (countDownLatch == null) {
            if (zkClient.exists(lockPath)) {
                countDownLatch = new CountDownLatch(1); //计数器设为1
            }
            try {
                countDownLatch.await(); // 如果当前计数器不是为0，就一直等待（主程序执行到await()函数会阻塞等待线程的执行，直到计数为0）
            } catch (Exception e) {
    
            }
            // 3.删除事件通知（取消订阅），防止继续监听
            zkClient.unsubscribeDataChanges(lockPath, iZkDataListener);
        }
    
        /**
         * 释放锁
         */
        @Override
        protected void unImplLock() {
            if (zkClient != null) {
                zkClient.close();
                System.out.println(Thread.currentThread().getName() + ",释放了锁");
            }
        }
    }

此时，基于模板方法方法的zookeeper实现分布式锁编写完毕~

可以把Redis实现分布式锁用模板方法优化下，定义成一个模板子类实现，感兴趣的童鞋可以自己去优化下。

同样的，用到zookeeper分布式锁的场景直接new出一个ZkTemplateImplLock对象就可

private Lock lock = new ZkTemplateImplLock();

拿上面的controller举例，只需在需要上锁，解锁的时候，分别调用** lock.getLock()** 和** lock.unLock()**即可。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 1][]

如需ZooKeeper也像Redis一样，指定过期时间，只需把**ZkTemplateImplLock** 类的常量**TIMEOUT**提取出来当做参数传给**ZkClient**即可。

**3. zkClient常用的两个方法：**

zkClient.subscribeChildChanges ： 监听子节点是否发生变化

zkClient.subscribeDataChanges ： 监听节点数据是否发生变化（zookeeper分布式锁就用到了该方法）

/**
     * dubbo服务注册就是靠的这个原理（事件通知）
     * @date 2019/9/21 20:39
     */
    public class Test006 {
        // 参数1 连接地址
        private static final String ADDRES = "192.168.0.8:2181";
        // 参数2 zk超时时间
        private static final int TIMEOUT = 5000;
    
        public static void main(String[] args) {
            // 1.创建我们的zk连接
            ZkClient zkClient = new ZkClient(ADDRES, TIMEOUT);
            String parentPath = "/my-service";
            // 2.【监听子节点是否有发生变化】如果发生变化都可以获取到回调通知
            zkClient.subscribeChildChanges(parentPath, new IZkChildListener() {
                @Override
                public void handleChildChange(String s, List<String> list) throws Exception {
                    System.out.println("s:" + s + ",节点发生了变化");
                    list.forEach((t) -> System.out.println(t));
                }
            });
            // 2.【监听节点的value值是否发生变化】
            zkClient.subscribeDataChanges(parentPath + "/8080", new IZkDataListener() {
                @Override
                // 监听节点的内容是否发生变化
                public void handleDataChange(String s, Object o) throws Exception {
                    System.out.println("s:" + s + ",o:" + o);
                }
                @Override
                // 监听该节点是否被删除
                public void handleDataDeleted(String s) throws Exception {
                    System.out.println("s被删除:" + s);
                }
            });
            // 死循环，让程序一直停在这,不终止，从而一直监听（ZkTemplateImplLock是用countDownLatch.await()实现，原理一个样）
            while (true) {
    
            }
        }
    }

### 个人总结： ###

**1. CountDownLatch**

CountDownLatch是一个同步辅助类，它允许一个或多个线程一直等待直到其他线程执行完毕才开始执行。  
  每次调用CountDown()，计数减1  
  主程序执行到await()函数会阻塞等待线程的执行，直到计数为0  
  计数器通过使用锁（共享锁、排它锁）实现

2. 业务逻辑卡了，导致根本无法关闭连接，导致锁无法释放，这时其它几个JVM都会进行等待，产生死锁。如何解决？

答案：每个连接拿到锁后，都应该有超时时间的，即**TIMEOUT**常量（在规定时间内如果没有释放锁，则自动进行关闭连接）。

那么问题又来了，如果zk超时了，会自动关闭连接，对一些幂等性要求比较高的操作（如任务调度）等业务逻辑（写操作）如何保证原子性问题？

答案：统一直接回滚，手动rollback事务（一般的，如果任务调度等用了分布式锁，超时时间应该大于任务调度需要的时间，但一般任务调度都用elastic-job或者xxl-job等框架去实现）

3. 简述zookeeper获取锁的原理

多个jvm同时创建临时节点，只要谁能够创建成功，谁就能够获取到锁

场景分析：  集群环境下，如任务调度做了集群，那么这时候就可以使用zookeeper分布式锁，谁能创建节点成功，谁就能执行任务调度

4. ① 节点如果是文件夹，也是可以有value    ② Lock锁与synchronized锁区别：手动挡，自动挡

5. **Z****ookeeper****实现****分布式锁的思路：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 2][]

特性：节点保证唯一、事件通知、临时节点（生命周期和Session会关联）

**创建分布式锁原理：**

① 多个jvm同时在Zookeeper上创建相同的临时节点(lockPath)

② 因为临时节点路径保证唯一的性，只要谁能够创建成功谁就能够获取锁，就可以开始执行业务逻辑；

③ 如果节点已经给其他请求创建的话或者是创建节点失败，当前的请求实现等待；

**释放锁的原理：**

.因为我们采用**临时节点**，当前节点创建成功，表示获取锁成功；正常执行完业务逻辑。调用Session关闭连接方法，当前的节点会删除；----释放锁

其他正在等待请求，采用**事件监听****，**如果当前节点被删除的话，有重新进入到获取锁流程；

简言之：临时节点+事件通知

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70]: /images/20220404/736ded07620d4342bf5f9f7d871064bf.png
[20181216221225596.png]: /images/20220404/d3937f56333f4dea9f03c05a38340468.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 1]: /images/20220404/f8113fd27be44c77ae0e51dc2d364c3a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 2]: /images/20220404/602c2287bcc04004882475819ad1cc6a.png