【Zookeeper】应用场景：分布式锁 InterProcessMutex 源码分析

拼搏现实的明天。 2022-12-20 11:15 401阅读 0赞

在[上一篇][Link 1]文章，我们介绍了通过 zookeeper 的临时有序节点实现分布式锁，而 Curator 已经实现好了一个线程分布式锁 InterProcessMutex。所以，本篇我们就来看看它的源码是怎么做的。

### InterProcessMutex ###

public class InterProcessMutex implements InterProcessLock, Revocable<InterProcessMutex> {
        
        
        // 核心，LockInternals类似AQS，负责锁的申请和释放
        private final LockInternals internals;
        private final String basePath;
    	// 记录线程与锁的映射关系
        private final ConcurrentMap<Thread, LockData> threadData = Maps.newConcurrentMap();
        
        // 锁信息，Zookeeper中一个临时顺序节点对应一个锁，但要让锁生效需要排队（公平锁）
        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;
            }
        }
        
    	// 最常用的构造函数
    	public InterProcessMutex(CuratorFramework client, String path)
    	{
        	
    		// zookeeper利用path创建临时顺序节点，实现公平锁的核心
        	this(client, path, new StandardLockInternalsDriver());
    	}
    
    	public InterProcessMutex(CuratorFramework client, String path, LockInternalsDriver driver)
        {
        		
            // maxLeases=1表示可以获得分布式锁的线程数量（跨JVM）为 1， 即互斥锁
            this(client, path, LOCK_NAME, 1, driver);
        }
        
        // protected构造函数，最终的构造函数
        InterProcessMutex(CuratorFramework client, String path, String lockName, int maxLeases, 
                          LockInternalsDriver driver)
        {
        
            basePath = PathUtils.validatePath(path);
            // InterProcessMutex将分布式锁的申请释放委托给internals执行
            internals = new LockInternals(client, driver, path, lockName, maxLeases);
        }
    }

### InterProcessMutex.acquire() ###

// 无限等待
    public void acquire() throws Exception
    {
        
        if ( !internalLock(-1, null) )
        {
        
            throw new IOException("Lost connection while trying to acquire lock: " + basePath);
        }
    }
    
    // 限时等待
    public boolean acquire(long time, TimeUnit unit) throws Exception
    {
        
        return internalLock(time, unit);
    }

### InterProcessMutex.internalLock() ###

private boolean internalLock(long time, TimeUnit unit) throws Exception
    {
        
    	// 拿到当前线程
        Thread currentThread = Thread.currentThread();
    	// 查映射表threadData，判断是否已经拿到锁
        LockData lockData = threadData.get(currentThread);
        if ( lockData != null )
        {
        
            // 实现可重入，即已拿锁线程再要拿锁时原子+1
            lockData.lockCount.incrementAndGet();
            return true;
        }
    	
    	// 没拿到锁，尝试通过LockInternals获取锁
        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.attemptLock() ###

// 尝试获取锁，并返回锁对应的Zookeeper临时顺序节点路径
    String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
    {
        
        final long      startMillis = System.currentTimeMillis();
        // 无线等待时为null
        final Long      millisToWait = (unit != null) ? unit.toMillis(time) : null; 
        // 创建ZNode节点时的数据内容，无关紧要，这里为null，采用默认值（IP地址）
        final byte[]    localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
        // 当前已经重试次数，与CuratorFramework的重试策略有关
        int             retryCount = 0;
    	
    	// 在Zookeeper创建临时顺序节点的路径，相当于一把待激活的分布式锁
    	// 激活条件：同级目录子节点，名称排序最小（排队，公平锁）
        String          ourPath = null;
        // 是否已经持有分布式锁
        boolean         hasTheLock = false;
        // 是否已经完成尝试获取分布式锁的操作
        boolean         isDone = false;
        while ( !isDone )
        {
        
            isDone = true;
    
            try
            {
        
            	// 在Zookeeper中创建临时顺序节点
            	// 从InterProcessMutex构造函数可知，实际driver为StandardLockInternalsDriver实例
                ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
                // 循环等待来激活分布式锁，实现锁的公平性
                hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
            }
            catch ( KeeperException.NoNodeException e )
            {
        
            	// 容错处理，因为会话过期等原因，driver因无法找到创建的节点而抛出noNodeException异常
                if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, 
                System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
                {
        	
                	// 满足重试策略尝试重新获取锁
                    isDone = false;
                }
                else
                {
        
                	// 不满足重试策略，则继续抛出NoNodeException
                    throw e;
                }
            }
        }
    
        if ( hasTheLock )
        {
        
        	// 成功获得分布式锁，返回临时顺序节点路径，上层将其封装在LockData记录在threadData映射表，好重入
            return ourPath;
        }
    	// 获取分布式锁失败，返回null
        return null;
    }

### StandardLockInternalsDriver.createsTheLock() ###

// 在Zookeeper中创建临时节点
    public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) 
        throws Exception
    {
        
        String ourPath;
        // lockNodeBytes部位null则作为数据节点内容，否则默认采用IP地址
        if ( lockNodeBytes != null )
        {
        
        	// creatingParentContainersIfNeeded，用于创建父节点，
            // withProtection，临时子节点会添加GUID前缀
            // CreateMode.EPHEMERAL_SEQUENTIAL，临时顺序节点，保证节点产生的顺序性
            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.internalLockLoop() ###

// 循环等待来激活分布式锁，实现锁的公平性
    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); 
    		   // 实现锁的公平性核心逻辑
                PredicateResults    predicateResults = driver.getsTheLock(client, children, 
                	sequenceNodeName, maxLeases);
                if ( predicateResults.getsTheLock() )
                {
        
                	// 获得了锁，中断循环，继续返回上层
                    haveTheLock = true;
                }
                else
                {
        	
                	// 没获得锁，监听上一临时顺序节点
                    String  previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
    
                    synchronized(this)
                    {
        
                        try 
                        {
        
    					  // exists()会导致资源泄露，因此exists()可以监听不存在ZNode，因此采用getData()
    					  // 上一临时顺序节点如果被删除，会唤醒当前线程继续竞争锁，，正常情况下会直接获得
                            client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                            if ( millisToWait != null )
                            {
        
                                millisToWait -= (System.currentTimeMillis() - startMillis);
                                startMillis = System.currentTimeMillis();
                                if ( millisToWait <= 0 )
                                {
        
                                    doDelete = true;    // 获取锁超时 ，标记删除之前创建的临时顺序节点
                                    break;
                                }
    
                                wait(millisToWait);
                            }
                            else
                            {
        
                                wait(); // 等待别唤醒，无限等待
                            }
                        }
                        catch ( KeeperException.NoNodeException e ) 
                        {
        
                            // 容错处理
                            // client.getData可能抛出NoNodeException，因为可能是锁释放或会话过期(链接丢失)
                            // 这里没做任何处理，因为外层是while循环，再次执行driver.getsTheLock时会调用
                            // validateOurIndex，此时会爬出NoNodeException，从而进入下面catch和finally
                            // 重新抛出上层尝试重试获取锁并删除临时顺序节点 
                        }
                    }
                }
            }
        }
        catch ( Exception e )
        {
        
            ThreadUtils.checkInterrupted(e);
            // 标记删除，在finally删除之前创建的临时顺序节点（后台不断尝试）
            doDelete = true;
            throw e; // 重新抛出，尝试重新获取锁
        }
        finally
        {
        
            if ( doDelete )
            {
        
                deleteOurPath(ourPath);
            }
        }
        return haveTheLock;
    }

### StandardLockInternalsDriver.getsTheLock() ###

public PredicateResults getsTheLock(CuratorFramework client, List<String> children,
                                        String sequenceNodeName, int maxLeases) throws Exception
    {
        
    	// 之前创建的临时顺序节点再排序后的节点列表中的索引
        int             ourIndex = children.indexOf(sequenceNodeName);
        // 校验之前创建的临时顺序节点是否有效
        validateOurIndex(sequenceNodeName, ourIndex);
    	// 锁公平性的核心逻辑
    	// 由InterProcessMutex构造函数式可知，maxLeases为1，即只有ourIndex为0时，线程才能持有锁，
    	// 或说该线程创建的临时顺序节点激活了锁
    	// Zookeeper的临时顺序节点特性保证了跨多个JVM的新城并发创建节点时的顺序性，越早创建越早获得锁
        boolean         getsTheLock = ourIndex < maxLeases;
        String          pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);
    	
    	// 返回获取锁的结果，交由上层继续处理（添加监听等操作）
        return new PredicateResults(pathToWatch, getsTheLock);
    }

### StandardLockInternalsDriver.validateOurIndex() ###

static void validateOurIndex(String sequenceNodeName, int ourIndex) throws KeeperException
    {
        
        if ( ourIndex < 0 )
        {
        
        	// 容错处理
            // 由于会话过期或连接丢失等原因，该线程创建的临时顺序节点被 Zookeeper 服务端删除，
            // 往外抛出NoNodeException         
            // 如果在重试策略允许范围内，则进行重 新尝试获取锁，这会重新重新生成临时顺序节点 
            throw new KeeperException.NoNodeException("Sequential path not found: " + 
            sequenceNodeName);
        }
    }

## 3.Curator释放锁源码分析 ##

### InterProcessMutex.release() ###

public void release() throws Exception
    {
        
     
        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 )
        {
        
        	// 锁是可重入等待，初始值为1，原子--到0，锁才能释放
            return;
        }
        if ( newLockCount < 0 )
        {
        
        	// 理论上无法执行该路径
            throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + 、
            basePath);
        }
        try
        {
        
        	 // lockData != null && newLockCount == 0，释放锁资源
            internals.releaseLock(lockData.lockPath);
        }
        finally
        {
        
        	// 最后从映射表中移除当前线程的锁信息
            threadData.remove(currentThread);
        }
    }

### LockInternals.releaseLock() ###

final void releaseLock(String lockPath) throws Exception
    {
        
        client.removeWatchers();
        // 删除临时顺序节点，只会触发后一节点去获取锁，理论上不存在竞争
        // 只排队，非抢占，公平锁，先到先得到
        revocable.set(null);
        deleteOurPath(lockPath);
    }

### LockInternals.deleteOurPath() ###

private void deleteOurPath(String ourPath) throws Exception
    {
        
        try
        {
        
        	// 后台不断尝试删除
            client.delete().guaranteed().forPath(ourPath);
        }
        catch ( KeeperException.NoNodeException e )
        {
        
            // 已经删除(可能会话过期导致)，不做处理         
            // 实际使用 Curator-2.12.0 时，并不会抛 出该异常 
        }
    }

[Link 1]: https://blog.csdn.net/weixin_43935927/article/details/114242116