Zookeeper之分布式锁

àì夳堔傛蜴生んèń 2023-05-29 02:59 12阅读 0赞

基于zk的特性，我们很容易得出使用zk实现分布式锁的落地方案：

1.  使用zk的临时节点和有序节点，每个线程获取锁就是在zk创建一个临时有序的节点，比如在/lock/目录下。
2.  创建节点成功后，获取/lock目录下的所有临时节点，再判断当前线程创建的节点是否是所有的节点的序号最小的节点
3.  如果当前线程创建的节点是所有节点序号最小的节点，则认为获取锁成功。
4.  如果当前线程创建的节点不是所有节点序号最小的节点，则对节点序号的前一个节点添加一个事件监听。 比如当前线程获取到的节点序号为`/lock/003`,然后所有的节点列表为`[/lock/001,/lock/002,/lock/003]`,则对`/lock/002`这个节点添加一个事件监听器。

如果锁释放了，会唤醒下一个序号的节点，然后重新执行第3步，判断是否自己的节点序号是最小。

比如`/lock/001`释放了，`/lock/002`监听到时间，此时节点集合为`[/lock/002,/lock/003]`,则`/lock/002`为最小序号节点，获取到锁。

整个过程如下：

![format_png][]

具体的实现思路就是这样，至于代码怎么写，这里比较复杂就不贴出来了。

### Curator介绍 ###

Curator是一个zookeeper的开源客户端，也提供了分布式锁的实现。

*  InterProcessMutex：分布式可重入排它锁
 *  InterProcessSemaphoreMutex：分布式排它锁
 *  InterProcessReadWriteLock：分布式读写锁
 *  InterProcessMultiLock：将多个锁作为单个实体管理的容器

他的使用方式也比较简单：

InterProcessMutex interProcessMutex = new InterProcessMutex(client,"/anyLock");
    interProcessMutex.acquire();
    interProcessMutex.release();

我们来看获取锁的代码:

public void acquire() throws Exception{
            if ( !internalLock(-1, null) ) {
                throw new IOException("Lost connection while trying to acquire lock: " + basePath);
            }
        }

private boolean internalLock(long time, TimeUnit unit) throws Exception {
            Thread currentThread = Thread.currentThread();
            LockData lockData = threadData.get(currentThread);
            if ( lockData != null )
            {
                // re-entering
                lockData.lockCount.incrementAndGet();
                return true;
            }
    
            String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
            if ( lockPath != null )
            {
                LockData newLockData = new LockData(currentThread, lockPath);
                threadData.put(currentThread, newLockData);
                return true;
            }
    
            return false;
        }

逻辑如下:  
每次获取锁时会直接从本地缓存中先获取锁的元数据,如果存在,则在原有的计数器基础上+1,直接返回;  
否则,尝试去获取锁,逻辑如下

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
        {
             
            final long  startMillis = System.currentTimeMillis();
             //等待时间
            final Long   millisToWait = (unit != null) ? unit.toMillis(time) : null;
            final byte[]    localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
            int             retryCount = 0;
    
            String          ourPath = null;
            boolean         hasTheLock = false;
            boolean         isDone = false;
            while ( !isDone )
            {
                isDone = true;
    
                try
                {
                  
                    ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
                    hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
                }
                catch ( KeeperException.NoNodeException e )
                {
                    // gets thrown by StandardLockInternalsDriver when it can't find the lock node
                    // this can happen when the session expires, etc. So, if the retry allows, just try it all again
                    if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
                    {
                        isDone = false;
                    }
                    else
                    {
                        throw e;
                    }
                }
            }
    
            if ( hasTheLock )
            {
                return ourPath;
            }
    
            return null;
        }

首先设置一个是否有锁的标志`hasTheLock = false`,然后  
`ourPath = driver.createsTheLock(client, path, localLockNodeBytes);`这个地方主要是通过`StandardLockInternalsDriver`在锁目录下创建EPHEMERAL\_SEQUENTIAL节点,  
`hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);`这里主要是循环获取锁的过程,代码看下面,首先是判断是否实现了revocable接口,如果实现了那么就对这个path设置监听,否则的话通过StandardLockInternalsDriver尝试得到PredicateResults(主要是否得到锁及需要监视的目录的两个属性);

private boolean internalLockLoop(long startMillis,Long millisToWait, String ourPath) throws Exception{
            boolean     haveTheLock = false;
            boolean     doDelete = false;
            try{
                if ( revocable.get() != null ){   client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
                }
                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();
    
                        synchronized(this)
                        {
                            try 
                            {
                                // use getData() instead of exists() to avoid leaving unneeded watchers which is a type of resource leak
                                client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                                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 ) 
                            {
                                // it has been deleted (i.e. lock released). Try to acquire again
                            }
                        }
                    }
                }
            }
            catch ( Exception e )
            {
                ThreadUtils.checkInterrupted(e);
                doDelete = true;
                throw e;
            }
            finally
            {
                if ( doDelete )
                {
                    deleteOurPath(ourPath);
                }
            }
            return haveTheLock;
        }

然后判断PredicateResults中的pathToWatch(主要保存sequenceNode)是否是最小的节点,如果是,则得到锁,getsTheLock为true,否则得到该序列的前一个节点,设为pathToWatch,并监控起来;再判断获取锁的时间是否超时,超时则删除节点,不竞争下次锁,否则,睡眠等待获取锁;最后把获取的锁对象的锁路径等信息封装成LockData存储在本地缓存中.

下面是释放锁的过程

public void release() throws Exception
        {
            /*
                Note on concurrency: a given lockData instance
                can be only acted on by a single thread so locking isn't necessary
             */
    
            Thread currentThread = Thread.currentThread();
            LockData lockData = threadData.get(currentThread);
            if ( lockData == null )
            {
                throw new IllegalMonitorStateException("You do not own the lock: " + basePath);
            }
    
            int newLockCount = lockData.lockCount.decrementAndGet();
            if ( newLockCount > 0 )
            {
                return;
            }
            if ( newLockCount < 0 )
            {
                throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + basePath);
            }
            try
            {
                internals.releaseLock(lockData.lockPath);
            }
            finally
            {
                threadData.remove(currentThread);
            }
        }

代码很简单,从本地缓存中拿到锁对象,计数器-1,只有到那个计数器=0的时候才会去执`internals.releaseLock(lockData.lockPath);`

final void releaseLock(String lockPath) throws Exception
        {
            client.removeWatchers();
            revocable.set(null);
            deleteOurPath(lockPath);
        }

只要逻辑见名知意,首先移除watcher监听,这个监听可能是在循环获取锁的时候创建的,然后取消动作时触发动作时间置空,最后就是删除path

## 两种方案的优缺点比较 ##

学完了两种分布式锁的实现方案之后，本节需要讨论的是redis和zk的实现方案中各自的优缺点。

对于redis的分布式锁而言，它有以下缺点：

*  **redis的设计定位决定了它的数据并不是强一致性的，只能保证最终一致性**，在某些极端情况下，可能会出现问题。锁的模型不够健壮
 *  即便使用redlock算法来实现，在某些复杂场景下，也无法保证其实现100%没有问题，关于redlock的讨论可以看How to do distributed locking
 *  利用redis做锁的时候,一般都需要做锁的有效时间限定。而curator则利用了zookeeper的临时顺序节点特性,一旦客户端失去连接后,则就会自动清除该节点

但是另一方面使用redis实现分布式锁在很多企业中非常常见，而且大部分情况下都不会遇到所谓的“极端复杂场景”

所以使用redis作为分布式锁也不失为一种好的方案，最重要的一点是redis的性能很高，可以支撑高并发的获取、释放锁操作。

对于zk分布式锁而言:

*  **zookeeper天生设计定位就是分布式协调，强一致性**。锁的模型健壮、简单易用、适合做分布式锁。
 *  如果获取不到锁，只需要添加一个监听器就可以了，不用一直轮询，性能消耗较小。

但是zk也有其缺点：如果有较多的客户端频繁的申请加锁、释放锁，对于zk集群的压力会比较大，Redis的读写性能要比ZK要高。

CAP理论中，zookeeper强调的是CP（一致性），redis强调的是AP(可用性)

## 小结 ##

通过前面的分析，实现分布式锁的两种常见方案：redis和zookeeper，他们各有千秋。应该如何选型呢？

就个人而言的话，**我比较推崇zk实现的锁：**

因为redis是有可能存在隐患的，可能会导致数据不对的情况。但是，怎么选用要看具体在公司的场景了。

如果公司里面有zk集群条件，优先选用zk实现，但是如果说公司里面只有redis集群，没有条件搭建zk集群。

那么其实用redis来实现也可以，另外还可能是系统设计者考虑到了系统已经有redis，但是又不希望再次引入一些外部依赖的情况下，可以选用redis。

这个是要系统设计者基于架构的考虑了。

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9hc2sucWNsb3VkaW1nLmNvbS9odHRwLXNhdmUveWVoZS0xMzE0MDQ3L2w4N3I3ZjU1b3YuanBlZw?x-oss-process=image/format,png