curator 实现分布式一致性锁

最近准备在项目中引入分布式锁，故而研究基于zookeeper的curator框架。

网上资料不多，自己研究其源码发现，这个框架已经帮我做了很多现成的实现。

下面介绍下锁的实现:

通过源码中LockingExample例子作为切入（推荐多利用现有资源，最快切入），为减小篇幅，代码仅保留关键部分。

curator已经为我们准备好了锁的实现 ----InterProcessMutex，基于zookeeper跨jvm的公平互斥锁实现.

-----------------------------------------------------------------------------------------------------------------------------

1.看下锁的定义，将线程对象和锁对象（线程、路径、锁的数量）关联

    private final ConcurrentMap<Thread, LockData> threadData = Maps.newConcurrentMap();

    private static class LockData

    {

        final Thread owningThread;

        final String lockPath;

        final AtomicInteger lockCount = new AtomicInteger(1);

        private LockData(Thread owningThread, String lockPath)

        {

            this.owningThread = owningThread;

            this.lockPath = lockPath;

        }

    }

2.获得锁：

如果线程已经有锁，则增加锁的数量，返回 String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
否则尝试获取锁，获得则加入线程持有锁的MAP，否则返回未获得锁。

    private final LockInternals internals;    

    @Override

    public boolean acquire(long time, TimeUnit unit) throws Exception

    {

        return internalLock(time, unit);

    }

    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;

    }

3.可以看到，逻辑是基于LockInternals，来看一下他是怎么做到的

通过dirver在锁目录下创建EPHEMERAL_SEQUENTIAL节点
循环尝试获取,基于监听,wait,notifyAll

获取锁目录下子节点的有序集合
通过dirver尝试得到PredicateResults(含有是否得到锁及需要监视的目录)
更新监听信息，开始下一轮尝试

    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);

            }

        }

    }

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();

                    System.out.println("ourpath : " + ourPath + " : " + previousSequencePath);

                    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 )

                        {

                        }

                    }

                }

            }

        }

        catch ( Exception e )

        {

            doDelete = true;

            throw e;

        }

        finally

        {

            if ( doDelete )

            {

                deleteOurPath(ourPath);

            }

        }

        return haveTheLock;

    }

4.分析下前文中的driver实现，即StandardLockInternalsDriver 关键代码

创建锁是基于目录下创建的EPHEMERAL_SEQUENTIAL节点，即与客户端生命周期相同，并且名字后自动加创建的序列号

得到PredicateResult,如果当前节点为最小节点，则得到锁，getsTheLock为true，否则得到该序列的前一个节点，设为pathToWatch，监控之

    @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;

        String          pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);

        return new PredicateResults(pathToWatch, getsTheLock);

    }

    @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;

    }

先发布，后续有心得再续。

巴特西

curator 实现分布式一致性锁

最新文章

热门文章