分布式专题（六）Zookeeper实现分布式锁

╰+哭是因爲堅強的太久メ 2023-07-23 14:57 11阅读 0赞

参考博客：[https://www.jianshu.com/p/a974eec257e6][https_www.jianshu.com_p_a974eec257e6]  
                  [https://www.jianshu.com/p/91976b27a188][https_www.jianshu.com_p_91976b27a188]

在许多场景中，**数据一致性**是一个比较重要的话题，在单机环境中，我们可以通过Java提供的**并发API**来解决；而在分布式环境(会遇到网络故障、消息重复、消息丢失等各种问题)下要复杂得多，常见的解决方案是**分布式事务**、**分布式锁**等。

本文主要探讨如何利用Zookeeper来实现分布式锁。

# 一、关于分布式锁 #

分布式锁是控制分布式系统之间**同步访问共享资源**的一种方式。

在**实现**分布式锁的过程中需要注意的：

*  锁的可重入性(递归调用不应该被阻塞、避免死锁)
 *  锁的超时(避免死锁、死循环等意外情况)
 *  锁的阻塞(保证原子性等)
 *  锁的特性支持(阻塞锁、可重入锁、公平锁、联锁、信号量、读写锁)

在**使用**分布式锁时需要注意：

*  分布式锁的开销(分布式锁一般能不用就不用，有些场景可以用乐观锁代替)
 *  加锁的粒度(控制加锁的粒度，可以优化系统的性能)
 *  加锁的方式

# 二、分布式锁的方案及其优缺点 #

## 基于数据库 ##

### **1. 基于数据库表** ###

最简单的方式可能就是直接创建一张锁表，当我们要锁住某个方法或资源时，我们就在该表中增加一条记录，想要释放锁的时候就删除这条记录。给某字段添加唯一性约束，如果有多个请求同时提交到数据库的话，**数据库会保证只有一个操作可以成功**，那么我们就可以认为操作成功的那个线程获得了该方法的锁，可以执行方法体内容。

会引入数据库单点、无失效时间、不阻塞、不可重入等问题。

### **2. 基于数据库排他锁** ###

如果使用的是MySql的InnoDB引擎，在查询语句后面增加`for update`，数据库会在查询过程中(须通过唯一索引查询)给数据库表增加排他锁，我们可以认为获得排它锁的线程即可获得分布式锁，通过 connection.commit() 操作来释放锁。

会引入数据库单点、不可重入、无法保证一定使用行锁(部分情况下MySQL自动使用表锁而不是行锁)、排他锁长时间不提交导致占用数据库连接等问题。

### **3. 数据库实现分布式锁总结** ###

优点：

*  直接借助数据库，容易理解。

缺点：

*  会引入更多的问题，使整个方案变得越来越复杂
 *  操作数据库需要一定的开销，有一定的性能问题
 *  使用数据库的行级锁并不一定靠谱，尤其是当我们的锁表并不大的时候

## 基于缓存 ##

相比较于基于数据库实现分布式锁的方案来说，基于缓存来实现在性能方面会表现的更好一点。目前有很多成熟的缓存产品，包括Redis、memcached、tair等。

这里以Redis为例举出几种实现方法：

### 1. 基于 redis 的 setnx()、expire() 方法做分布式锁 ###

setnx 的含义就是 `SET if Not Exists`，其主要有两个参数 `setnx(key, value)`。该方法是原子的，如果 key 不存在，则设置当前 key 成功，返回 1；如果当前 key 已经存在，则设置当前 key 失败，返回 0。

expire 设置过期时间，要注意的是 setnx 命令不能设置 key 的超时时间，只能通过 expire() 来对 key 设置。

### 2. 基于 redis 的 setnx()、get()、getset()方法做分布式锁 ###

getset 这个命令主要有两个参数 `getset(key，newValue)`，该方法是原子的，对 key 设置 newValue 这个值，并且返回 key 原来的旧值。

### 3. 基于 Redlock 做分布式锁 ###

Redlock 是 Redis 的作者 antirez 给出的集群模式的 Redis 分布式锁，它基于 N 个完全独立的 Redis 节点（通常情况下 N 可以设置成 5）

### 4. 基于 redisson 做分布式锁 ###

redisson 是 redis 官方的分布式锁组件，GitHub 地址：[https://github.com/redisson/redisson][https_github.com_redisson_redisson]

### 基于缓存实现分布式锁总结 ###

优点：

*  性能好

缺点：

*  实现中需要考虑的因素太多
 *  通过超时时间来控制锁的失效时间并不是十分的靠谱

## 基于Zookeeper ##

> 基于Zookeeper实现分布式锁有两种：一种是基于**强一致性的排他锁**，节点的创建具有全局唯一性；另一种是基于**临时有序性的共享锁**，序号最小的节点最先获取锁，临时节点删除后释放锁。

### Zookeeper 如何实现分布式锁？ ###

下面讲如何实现排他锁和共享锁，以及如何解决羊群效应。

**排他锁**

排他锁，又称写锁或**独占锁**。如果事务T1对数据对象O1加上了排他锁，那么在整个加锁期间，只允许事务T1对O1进行读取或更新操作，其他任务事务都不能对这个数据对象进行任何操作，直到T1释放了排他锁。

排他锁核心是**保证当前有且仅有一个事务获得锁，并且锁释放之后，所有正在等待获取锁的事务都能够被通知到**。

Zookeeper 的**强一致性**特性，能够很好地保证在分布式高并发情况下**节点的创建一定能够保证全局唯一性**，即Zookeeper将会保证客户端无法重复创建一个已经存在的数据节点。可以利用Zookeeper这个特性，实现排他锁。

*  **定义锁**：通过Zookeeper上的数据节点来表示一个锁
 *  **获取锁**：客户端通过调用 `create` 方法创建表示锁的临时节点，可以认为创建成功的客户端获得了锁，同时可以让没有获得锁的节点在该节点上注册Watcher监听，以便实时监听到lock节点的变更情况
 *  **释放锁**：以下两种情况都可以让锁释放
    
     *  当前获得锁的客户端发生宕机或异常，那么Zookeeper上这个临时节点就会被删除
     *  正常执行完业务逻辑，客户端主动删除自己创建的临时节点

基于Zookeeper实现排他锁流程：

![format_png][]

**共享锁**

共享锁，如果事务T1对数据对象O1加上了共享锁，那么当前事务只能对O1进行**读取操作**，其他事务也只能对这个数据对象加共享锁，直到该数据对象上的所有共享锁都被释放。

共享锁与排他锁的区别在于，加了排他锁之后，数据对象只对当前事务可见，而加了共享锁之后，数据对象对所有事务都可见。

算法流程如下：

*  每个客户端创建**临时有序**节点
 *  客户端获取节点列表，判断自己是否列表中的第一个节点，如果是就获得锁，如果不是就**监听自己前面的节点**，等待前面节点被删除
 *  如果获取锁就进行正常的业务流程，执行完释放锁

同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题

为什么要监听前一个节点而不是所有的节点呢？这是因为如果监听所有的子节点，那么任意一个子节点状态改变，其它所有子节点都会收到通知（**羊群效应**），而我们只希望它的后一个子节点收到通知。

Zookeeper羊群效应改进前后Watcher监听图

![format_png 1][]

### **Zookeeper实现分布式锁总结** ###

优点：

*  有效的解决单点问题，不可重入问题，非阻塞问题以及锁无法释放的问题
 *  实现较为简单

缺点：

*  性能上不如使用缓存实现的分布式锁，因为每次在创建锁和释放锁的过程中，都要动态创建、销毁临时节点来实现锁功能
 *  需要对Zookeeper的原理有所了解

# 三、基于Curator客户端实现分布式锁 #

采用zk的原生API实现会比较复杂，所以这里就直接用Curator这个轮子，采用Curator的`acquire`和`release`两个方法就能实现分布式锁。

import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.framework.recipes.locks.InterProcessMutex;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    
    public class CuratorDistributeLock {
    
        public static void main(String[] args) {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
            CuratorFramework client = CuratorFrameworkFactory.newClient("111.231.83.101:2181",retryPolicy);
            client.start();
            CuratorFramework client2 = CuratorFrameworkFactory.newClient("111.231.83.101:2181",retryPolicy);
            client2.start();
            //创建分布式锁, 锁空间的根节点路径为/curator/lock
            InterProcessMutex mutex  = new InterProcessMutex(client,"/curator/lock");
            final InterProcessMutex mutex2  = new InterProcessMutex(client2,"/curator/lock");
            try {
                mutex.acquire();
            } catch (Exception e) {
                e.printStackTrace();
            }
            //获得了锁, 进行业务流程
            System.out.println("clent Enter mutex");
             Thread client2Th = new Thread(new Runnable() {
                 @Override
                 public void run() {
                     try {
                         mutex2.acquire();
                         System.out.println("client2 Enter mutex");
                         mutex2.release();
                         System.out.println("client2 release lock");
    
                     }catch (Exception e){
                         e.printStackTrace();
                     }
    
                 }
             });
            client2Th.start();
            //完成业务流程, 释放锁
            try {
                Thread.sleep(5000);
                mutex.release();
                System.out.println("client release lock");
                client2Th.join();
            } catch (Exception e) {
                e.printStackTrace();
            }
    
            //关闭客户端
            client.close();
        }
    }

上述代码的执行结果如下：

![format_png 2][]

可以看到`client`客户端首先拿到锁再执行业务，然后再轮到`client2`尝试获取锁并执行业务。

## 源码分析 ##

一直追踪`acquire()`的加锁方法，可以追踪到加锁的核心函数为`attemptLock`。

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
        {
            .....
            while ( !isDone )
            {
                isDone = true;
    
                try
                {
                    //创建临时有序节点
                    ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
                    //判断自己是否最小序号的节点，如果不是添加监听前面节点被删的通知
                    hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
                }
            }
            //如果获取锁返回节点路径
            if ( hasTheLock )
            {
                return ourPath;
            }
            ....
        }

深入`internalLockLoop`函数源码：

private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception
        {
              .......
                while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )
                {
                   //获取子节点列表按照序号从小到大排序
                    List<String>        children = getSortedChildren();
                    String              sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
                    //判断自己是否是当前最小序号节点
                    PredicateResults    predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
                    if ( predicateResults.getsTheLock() )
                    {
                        //成功获取锁
                        haveTheLock = true;
                    }
                    else
                    {
                       //拿到前一个节点
                        String  previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
                     //如果没有拿到锁,调用wait，等待前一个节点删除时,通过回调notifyAll唤醒当前线程
                        synchronized(this)
                        {
                            try 
                            {
                               //设置监听器，getData会判读前一个节点是否存在，不存在就会抛出异常从而不会设置监听器
                                client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                                //如果设置了millisToWait，等一段时间，到了时间删除自己跳出循环
                                if ( millisToWait != null )
                                {
                                    millisToWait -= (System.currentTimeMillis() - startMillis);
                                    startMillis = System.currentTimeMillis();
                                    if ( millisToWait <= 0 )
                                    {
                                        doDelete = true;    // timed out - delete our node
                                        break;
                                    }
                                    //等待一段时间
                                    wait(millisToWait);
                                }
                                else
                                {
                                    //一直等待下去
                                    wait();
                                }
                            }
                            catch ( KeeperException.NoNodeException e ) 
                            {
                              //getData发现前一个子节点被删除，抛出异常
                            }
                        }
                    }
                }
            }
            .....
        }

[https_www.jianshu.com_p_a974eec257e6]: https://www.jianshu.com/p/a974eec257e6
[https_www.jianshu.com_p_91976b27a188]: https://www.jianshu.com/p/91976b27a188
[https_github.com_redisson_redisson]: https://github.com/redisson/redisson
[format_png]: /images/20230528/5f7109223aef4f1db4c1c08edd8b5dd8.png
[format_png 1]: /images/20230528/8fd11f8d42944e298278cdf35ae50480.png
[format_png 2]: /images/20230528/f64ae75864bb423292418dc27f3bb881.png