Curator 源码初探（一）公平锁

秒速五厘米 2022-11-03 11:40 136阅读 0赞

Curator 是一个基于ZK实现的分布式锁框架 ，如何实现分布式锁的？我们来看看

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    		CuratorFramework client = CuratorFrameworkFactory.newClient(
    				"127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183",  
    				retryPolicy);
    		client.start();
    		
    		InterProcessMutex lock = new InterProcessMutex(client, "/locks/lock_01");
    		lock.acquire();
    		lock.release();

public InterProcessMutex(CuratorFramework client, String path)
        {
            this(client, path, LOCK_NAME, 1, new StandardLockInternalsDriver());
        }

InterProcessMutex(CuratorFramework client, String path, String lockName, int maxLeases, LockInternalsDriver driver)
        {
            basePath = path;
            internals = new LockInternals(client, driver, path, lockName, maxLeases);
        }

/**
         * Given a parent path and a child node, create a combined full path
         *
         * @param parent the parent
         * @param child  the child
         * @return full path
         */
        public static String makePath(String parent, String child)

/**
         * Acquire the mutex - blocking until it's available. Note: the same thread
         * can call acquire re-entrantly. Each call to acquire must be balanced by a call
         * to {@link #release()}
         * 获取一个互斥锁 阻塞直到锁有效，也就是当前线程获取到锁，同一个线程是可重入的，意思就是
         * 可重入锁，每次获取锁之后要使用 release()方法去释放掉锁
         *
         * @throws Exception ZK errors, interruptions, another thread owns the lock
         */
        @Override
        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
        {
            /*
               Note on concurrency: a given lockData instance
               can be only acted on by a single thread so locking isn't necessary
                一个给定的lockData实例只能被一个单线程操作，所以不需要去锁定
            */
            //获取线程
            Thread          currentThread = Thread.currentThread();
            //
            LockData        lockData = threadData.get(currentThread);
            //如果lockData 存在，意思就代表重入
            if ( lockData != null )
            {
                // re-entering 这里是可重入锁机制，我把属性拿过来看看
                //其实就是一个AotmicInteger 并发包里面的对象对吧
                //其实就是记录重入的次数，第一次默认初始化为1
                //final AtomicInteger lockCount = new AtomicInteger(1);
                lockData.lockCount.incrementAndGet();
                return true;
            }
            //尝试去获取锁 其实就是去创建节点
            String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
            if ( lockPath != null )
            {
                //如果获取到锁 ，就将当前线程作为key  path 作为value 放入到 concurrentHashMap中
                LockData        newLockData = new LockData(currentThread, lockPath);
                threadData.put(currentThread, newLockData);
                return true;
            }
    
            return false;
        }

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
                {
                    if ( localLockNodeBytes != null )
                    {
                        ourPath = client.create().creatingParentsIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, localLockNodeBytes);
                    }
                    else
                    {
                        //去创建临时顺序节点
                        //path  /locks/lock_01/lock-
                        ourPath = client.create().creatingParentsIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
                    }
                    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;
        }

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);
                    //如果返回获取锁成功，那么就返回true
                    if ( predicateResults.getsTheLock() )
                    {
                        haveTheLock = true;
                    }
                    //否则，怎么？添加一个watcher
                    else
                    {
                        //上一个节点的序号进行监听上一个节点
                        ///locks/lock_01/lock-0000001
                        String  previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
    
                        synchronized(this)
                        {
                            //对上一个node 节点进行监听是否还存在
                            Stat stat = client.checkExists().usingWatcher(watcher).forPath(previousSequencePath);
                            if ( stat != null )
                            {
                                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();
                                }
                            }
                        }
                        // else it may have been deleted (i.e. lock released). Try to acquire again
                    }
                }
            }
            catch ( Exception e )
            {
                doDelete = true;
                throw e;
            }
            finally
            {
                if ( doDelete )
                {
                    deleteOurPath(ourPath);
                }
            }
            return haveTheLock;
        }

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)
                        {
                            Stat stat = client.checkExists().usingWatcher(watcher).forPath(previousSequencePath);
                            if ( stat != null )
                            {
                                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();
                                }
                            }
                        }
                        // else it may have been deleted (i.e. lock released). Try to acquire again
                    }
                }
            }
            catch ( Exception e )
            {
                doDelete = true;
                throw e;
            }
            finally
            {
                if ( doDelete )
                {
                    deleteOurPath(ourPath);
                }
            }
            return haveTheLock;
        }

获取对应锁下面的所有子节点，然后按照顺序进行排序

public static List<String> getSortedChildren(CuratorFramework client, String basePath, final String lockName, final LockInternalsSorter sorter) throws Exception
        {
            //basePath /locks/lock_01
            List<String> children = client.getChildren().forPath(basePath);
            List<String> sortedList = Lists.newArrayList(children);
            Collections.sort
            (
                sortedList,
                new Comparator<String>()
                {
                    @Override
                    public int compare(String lhs, String rhs)
                    {
                        return sorter.fixForSorting(lhs, lockName).compareTo(sorter.fixForSorting(rhs, lockName));
                    }
                }
            );
            return sortedList;
        }

@Override
        public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
        {
            //获取当前创建的节点 在 数组中的位置
            //什么意思？在顺序节点中排第几号
            int             ourIndex = children.indexOf(sequenceNodeName);
            validateOurIndex(sequenceNodeName, ourIndex);
            //判断当前线程创建的节点的index 是否小于maxKeases 个数
               //这里一般是为1，但是为什么会设置成这样的呢？？？
            //大家想一想，如果要通过ZK实现信号量模式，是不是只需要将maxLeases 的个数调成我们需要的
            //数字就行了 对把 
            boolean         getsTheLock = ourIndex < maxLeases;
            String          pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);
    
            return new PredicateResults(pathToWatch, getsTheLock);
        }

客户端释放锁

@Override
        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);
            }
            //将对应的loackData 中的原子类 存储的数字-1
            int newLockCount = lockData.lockCount.decrementAndGet();
              //如果大于0 说明还有重入
            if ( newLockCount > 0 )
            {
                return;
            }
              //小于0 说明有问题
            if ( newLockCount < 0 )
            {
                throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + basePath);
            }
    
            try
            {    
                //移除掉当前线程持有锁的node节点
                
                internals.releaseLock(lockData.lockPath);
            }
            finally
            {
                //从concurrentHashMap中移除 lockData
                threadData.remove(currentThread);
            }
        }

移除path

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

那么如果释放锁之后。怎么通知其他的客户端去获取锁？

ZK 的watcher 机制，对吧，第二个node节点监听第一个node节点被删除掉了，就会触发回调，ZK 通知client其执行watcher中的逻辑

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

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

唤醒所有的阻塞线程，那么之前获取锁失败，使用wait（）线程就会被唤醒，然后尝试去 锁，获取锁如何获取？还是之前的逻辑，判断当前节点是否为第一个节点，如果是 获取节点，对吧

到这我们大致可以看出来其是ZK 实现的分布式锁是公平锁，并没有非公平锁的实现