Zookeeper分布式锁的原理

上一篇写了zk分布式锁的使用，现在我们来看一下Curator是怎么实现分布式锁的。

简单的来说：

（1）各个线程在当前path下生成顺序节点；

（2）序号为0的节点成功拿到锁；

（3）没有拿到锁的节点会增加一个对上一个节点的Watch，并阻塞；

（4）当第一个节点删除时，下一个节点被唤醒，重新去拿锁。（或者阻塞一定时间后删除自身节点，返回获取锁失败）

 

 

首先来看一下获取锁的代码：

 

lock.acquire();//会一直阻塞到获得锁成功

或者

boolean locked= lock.acquire(2000, TimeUnit.MILLISECONDS);

1.点进去可以看到，调用的方法是InterProcessMutex.internalLock()

 

 

private boolean internalLock(long time, TimeUnit unit) throws Exception
    {
        Thread currentThread = Thread.currentThread();
        //先看当前线程是否已经获得过锁
        LockData lockData = threadData.get(currentThread);
        if ( lockData != null )
        {
            //有则直接重入
            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;
    }

 2.再看一下internals.attemptLock(time, unit, getLockNodeBytes())

 

 

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
            {
                //关键看这两句
                //先成当前的path;
                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;
    }

 3.再看 driver.createsTheLock(client, path, localLockNodeBytes)

 

 @Override
    public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) throws Exception
    {
        String ourPath;
        if ( lockNodeBytes != null )
        {
            ourPath = client.create().creatingParentsIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, lockNodeBytes);
        }
        else
        {
            ourPath = client.create().creatingParentsIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
        }
        return ourPath;
    }

 可以看到，只是在path下面生成临时的顺序节点。

 

 

4.判断是否获得到了锁：hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);

 

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
                {
                    //获取失败时，Watch前一个节点
                    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
                            //注册Watch
                            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;
                                }
                                //阻塞，等待watch事件触发后的通知
                                wait(millisToWait);
                            }
                            else
                            {
                                wait();
                            }
                        }
                        catch ( KeeperException.NoNodeException e ) 
                        {
                            // it has been deleted (i.e. lock released). Try to acquire again
                        }
                    }
                }
            }
        }
        catch ( Exception e )
        {
            doDelete = true;
            throw e;
        }
        finally
        {
            if ( doDelete )
            {
                deleteOurPath(ourPath);
            }
        }
        return haveTheLock;
    }

 5. PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);

  

 @Override
    public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
    {
        int             ourIndex = children.indexOf(sequenceNodeName);
        validateOurIndex(sequenceNodeName, ourIndex);

        boolean         getsTheLock = ourIndex < maxLeases;//从前面可以看到maxLeases的值是1，所以只有当前节点是第一个顺序节点时，才能成功获得锁，其他都会失败。
        String          pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases); //如果获取锁失败了，要继续等待，那必须要对前面的节点进行watch,

        return new PredicateResults(pathToWatch, getsTheLock);
    }

 6.被watch的节点发生改变时，会唤醒阻塞的线程去重新竞争锁;

 private final Watcher watcher = new Watcher()
    {
        @Override
        public void process(WatchedEvent event)
        {
            notifyFromWatcher();
        }
    };

  private synchronized void notifyFromWatcher()
    {
        notifyAll();
    }