zookeeper源码-分布式锁

逃离我推掉我的手 2021-12-04 07:21 193阅读 0赞

curator2.12，zookeeper3.4版本

加锁举例

public class Test {
        public static void main(String[] args) throws Exception {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(100, 3);
            CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", retryPolicy);
            client.start();
    
            //锁节点路径必须以/开始
            InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");
    
            mutex.acquire();
            System.out.println("拿到锁");
    
            mutex.release();
            System.out.println("释放锁");
    
            //client.close();
    
        }
    }

## 建立连接 ##

CuratorFrameworkFactory

private static final int DEFAULT_SESSION_TIMEOUT_MS = Integer.getInteger("curator-default-session-timeout", 60 * 1000);
        private static final int DEFAULT_CONNECTION_TIMEOUT_MS = Integer.getInteger("curator-default-connection-timeout", 15 * 1000);
    
        public static CuratorFramework newClient(String connectString, RetryPolicy retryPolicy)
        {
            return newClient(connectString, DEFAULT_SESSION_TIMEOUT_MS, DEFAULT_CONNECTION_TIMEOUT_MS, retryPolicy);
        }
    
        public static CuratorFramework newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)
        {
            return builder().
                connectString(connectString).
                sessionTimeoutMs(sessionTimeoutMs).
                connectionTimeoutMs(connectionTimeoutMs).
                retryPolicy(retryPolicy).
                build();
        }
    
        public static Builder builder()
        {
            return new Builder();
        }
    
            public CuratorFramework build()
            {
                return new CuratorFrameworkImpl(this);
            }

CuratorFrameworkImpl

public CuratorFrameworkImpl(CuratorFrameworkFactory.Builder builder)
        {
            ZookeeperFactory localZookeeperFactory = makeZookeeperFactory(builder.getZookeeperFactory());
            this.client = new CuratorZookeeperClient(localZookeeperFactory, builder.getEnsembleProvider(), builder.getSessionTimeoutMs(), builder.getConnectionTimeoutMs(), new Watcher()
            {
                @Override
                public void process(WatchedEvent watchedEvent)
                {
                    CuratorEvent event = new CuratorEventImpl(CuratorFrameworkImpl.this, CuratorEventType.WATCHED, watchedEvent.getState().getIntValue(), unfixForNamespace(watchedEvent.getPath()), null, null, null, null, null, watchedEvent, null);
                    processEvent(event);
                }
            }, builder.getRetryPolicy(), builder.canBeReadOnly());
    
            listeners = new ListenerContainer<CuratorListener>();
            unhandledErrorListeners = new ListenerContainer<UnhandledErrorListener>();
            backgroundOperations = new DelayQueue<OperationAndData<?>>();
            namespace = new NamespaceImpl(this, builder.getNamespace());
            threadFactory = getThreadFactory(builder);
            maxCloseWaitMs = builder.getMaxCloseWaitMs();
            connectionStateManager = new ConnectionStateManager(this, builder.getThreadFactory());
            compressionProvider = builder.getCompressionProvider();
            aclProvider = builder.getAclProvider();
            state = new AtomicReference<CuratorFrameworkState>(CuratorFrameworkState.LATENT);
            useContainerParentsIfAvailable = builder.useContainerParentsIfAvailable();
    
            byte[] builderDefaultData = builder.getDefaultData();
            defaultData = (builderDefaultData != null) ? Arrays.copyOf(builderDefaultData, builderDefaultData.length) : new byte[0];
            authInfos = buildAuths(builder);
    
            failedDeleteManager = new FailedDeleteManager(this);
            namespaceFacadeCache = new NamespaceFacadeCache(this);
        }

## 建立锁节点 ##

InterProcessMutex

private static final String LOCK_NAME = "lock-";
    
        /**
         * @param client client
         * @param path   the path to lock
         */
        public InterProcessMutex(CuratorFramework client, String path)
        {
            this(client, path, new StandardLockInternalsDriver());
        }
        public InterProcessMutex(CuratorFramework client, String path, LockInternalsDriver driver)
        {
            this(client, path, LOCK_NAME, 1, driver);
        }
        
        private final LockInternals internals;
        private final String basePath;
    
        InterProcessMutex(CuratorFramework client, String path, String lockName, int maxLeases, LockInternalsDriver driver)
        {
            basePath = PathUtils.validatePath(path);
            internals = new LockInternals(client, driver, path, lockName, maxLeases);
        }

## 加锁 ##

/**
         * 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()}
         *
         * @throws Exception ZK errors, connection interruptions
         */
        @Override
        public void acquire() throws Exception
        {
            if ( !internalLock(-1, null) )
            {
                throw new IOException("Lost connection while trying to acquire lock: " + basePath);
            }
        }
    
    
        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;
            }
        }
    
        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
            */
    
            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;
        }

LockInternals

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

StandardLockInternalsDriver

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

LockInternals

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

StandardLockInternalsDriver

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

失败删除节点  
LockInternals

private void deleteOurPath(String ourPath) throws Exception
        {
            try
            {
                client.delete().guaranteed().forPath(ourPath);
            }
            catch ( KeeperException.NoNodeException e )
            {
                // ignore - already deleted (possibly expired session, etc.)
            }
        }

## 释放锁 ##

InterProcessMutex

/**
         * Perform one release of the mutex if the calling thread is the same thread that acquired it. If the
         * thread had made multiple calls to acquire, the mutex will still be held when this method returns.
         *
         * @throws Exception ZK errors, interruptions, current thread does not own the lock
         */
        @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);
            }
    
            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);
            }
        }

计数器减一后，大于0，说明重入锁，直接结束  
如果等于0，说明释放锁，删除节点即可

LockInternals

private final AtomicReference<RevocationSpec>   revocable = new AtomicReference<RevocationSpec>(null);
        void releaseLock(String lockPath) throws Exception
        {
            revocable.set(null);
            deleteOurPath(lockPath);///curator/lock/_c_b967a79f-8052-435e-818b-b32b6fbb9d16-lock-0000000007
        }