Curator 源码初探（四）读写锁

太过爱你忘了你带给我的痛 2022-11-03 13:29 159阅读 0赞

InterProcessReadWriteLock lock = new InterProcessReadWriteLock(
    				client, "/locks/lock_01");
    		lock.writeLock().acquire();
    		lock.readLock().acquire();
    		lock.readLock().release();  
    		lock.writeLock().release();

创建读写锁，底层是基于InternalInterProcessMutex，创建一个写锁，创建了一个读锁

public InterProcessReadWriteLock(CuratorFramework client, String basePath)
        {
            writeMutex = new InternalInterProcessMutex
            (
                client,
                basePath,
                WRITE_LOCK_NAME,
                1,
                new SortingLockInternalsDriver()
                {
                    @Override
                    public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
                    {
                        return super.getsTheLock(client, children, sequenceNodeName, maxLeases);
                    }
                }
            );
    
            readMutex = new InternalInterProcessMutex
            (
                client,
                basePath,
                READ_LOCK_NAME,
                Integer.MAX_VALUE,
                new SortingLockInternalsDriver()
                {
                    @Override
                    public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
                    {
                        return readLockPredicate(children, sequenceNodeName);
                    }
                }
            );
        }

InternalInterProcessMutex 这个类又是InterProcessMutex 普通锁的子类，

private static class InternalInterProcessMutex extends InterProcessMutex
        {
            private final String lockName;
    
            InternalInterProcessMutex(CuratorFramework client, String path, String lockName, int maxLeases, LockInternalsDriver driver)
            {
                super(client, path, lockName, maxLeases, driver);
                this.lockName = lockName;
            }
    
            @Override
            public Collection<String> getParticipantNodes() throws Exception
            {
                Collection<String>  nodes = super.getParticipantNodes();
                Iterable<String>    filtered = Iterables.filter
                (
                    nodes,
                    new Predicate<String>()
                    {
                        @Override
                        public boolean apply(String node)
                        {
                            return node.contains(lockName);
                        }
                    }
                );
                return ImmutableList.copyOf(filtered);
            }
        }

尝试获取锁我们先来看下读锁

基本逻辑跟之前的普通锁差不多，区别在哪？

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

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 )
                    {
                        //locks/lock_01/dsadsadasd-READ-00000001
                        ourPath = client.create().creatingParentsIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, localLockNodeBytes);
                    }
                    else
                    {
                        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;
        }

根据 locks/lock\_01/XXXXXXXXXXXXXX-READ-00000001，创建一个临时顺序节点

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

重点区别在这里 ，读锁，自己实现了一套SortingLockInternalsDrive.getsTheLock逻辑

readMutex = new InternalInterProcessMutex
            (
                client,
                basePath,
                READ_LOCK_NAME,
                Integer.MAX_VALUE,
                new SortingLockInternalsDriver()
                {
                    @Override
                    public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
                    {
                        return readLockPredicate(children, sequenceNodeName);
                    }
                }
            );

private PredicateResults readLockPredicate(List<String> children, String sequenceNodeName) throws Exception
        {
            //这里是什么意思？其实就是允许先写锁 后读锁，判断释放是当前线程加的写锁
            if ( writeMutex.isOwnedByCurrentThread() )
            {
                return new PredicateResults(null, true);
            }
    
            int         index = 0;
            int         firstWriteIndex = Integer.MAX_VALUE;
            int         ourIndex = Integer.MAX_VALUE;
            for ( String node : children )
            {
                if ( node.contains(WRITE_LOCK_NAME) )
                {
                    firstWriteIndex = Math.min(index, firstWriteIndex);
                }
                else if ( node.startsWith(sequenceNodeName) )
                {
                    ourIndex = index;
                    break;
                }
    
                ++index;
            }
            StandardLockInternalsDriver.validateOurIndex(sequenceNodeName, ourIndex);
    
            boolean     getsTheLock = (ourIndex < firstWriteIndex);
            String      pathToWatch = getsTheLock ? null : children.get(firstWriteIndex);
            return new PredicateResults(pathToWatch, getsTheLock);
        }

那么写锁是怎么加锁的？

肯定也是创建一个临时顺序节点，也就是 /locks/lock\_01/xxxxxxx-WRIT-0000002

writeMutex = new InternalInterProcessMutex
            (
                client,
                basePath,
                WRITE_LOCK_NAME,
                1,
                new SortingLockInternalsDriver()
                {
                    @Override
                    public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
                    {
                        return super.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;
            String          pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);
    
            return new PredicateResults(pathToWatch, getsTheLock);
        }

必须排第一位 才能加写锁成功，否则就会创建一个watcher 去监听前面一个节点，然后阻塞

那么我们来分析一下先读锁 后写锁 会如何？

其实一定会互斥，然后直接导致程序卡死，为什么？

加读锁肯定没问题，那么尝试加写锁的时候就会导致阻塞，然后就会导致程序直接阻塞，

我们来分析一下几种加锁的方式

同一线程的可重入性分析

（1）同一线程的先读锁后写锁是互斥的，不支持的，会导致程序卡死

（2）统一线程的先写锁后读锁，是允许的，加读锁的时候，会判断写锁是否是当前线程持有，其实就是去判断writeMutex 中的concurrentHashMap中是否有当前线程对应的 lockData，注意 readMutex 与 writeMutex 是两个分别独立的InterProcessMutex，其实就是判断是否有lockData 就OK 了

（3）同一线程的读锁重入性，其实就是普通锁的逻辑，直接将readMutex 也就是InterProcessMutex 中的 lockData.lockCount.incrementAndGet(); +1 就Ok，并不会去操作ZK node节点

（4）同一线程的写锁可重入性，其实就是普通锁的逻辑，直接将writeMutex 也就是InterProcessMutex 中的 lockData.lockCount.incrementAndGet(); +1 就Ok，并不会去操作ZK node节点

不同线程的读写锁分析

（1）先读 后写，线程1会创建一个/locks/lock\_01/xxxxx-READ-0000000, 线程2 尝试获取写锁，创建一个/locks/lock\_01/xxxxxxxxx-WRIT-000001节点，然后判断自己的index是否是0，如果是 肯定就是获取写锁，这里肯定不是对吧，那么就会创建一个watcher 去监听前面一个节点的变化，这里自己思考一下，如果前面有多个READ节点会怎么操作？

（2）先写 后读，写线程会创建一个/locks/lock\_01/xxxxxxxxxx-WRIT-00000,然后读节点就创建一个临时顺序节点，同时会监听前面的/locks/lock\_01/xxxxxxxxxx-WRIT-00000 写节点

（3）先读 后读，两个线程的加读锁，读锁的判断逻辑其实是1）先判断是否是当前线程持有写锁（不是我们这考虑的）2）看自己前面的顺序节点是否都是READ节点，如果存在WRIT节点，那么就会监听自己前面的那个写节点，创建一个watcher 进行监听

（4）先写 后写，其实这个逻辑应该是最简单的，因为写锁会判断是否是index，就是自己是否是第一位，否则就监听前面一个节点的变化