深度解析Curator分布式锁InterProcessMutex源码

淩亂°似流年 2023-01-16 15:23 280阅读 0赞

> 原创不易，转载请注明出处

### 文章目录 ###

*   *   *  前言
         *  1.使用demo
         *  2.zk的节点类型介绍
         *  3.顺序节点演示
         *  4.核心原理
         *  5.源码分析
         *   *  初始化
             *  获取锁
             *  释放锁

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

### 前言 ###

本文将介绍下基于zk实现的分布式锁InterProcessMutex实现原理，InterProcessMutex 是zk java 客户端curator提供的一个分布式锁工具，本文主要分为5大部分，分别是使用该锁的demo，zk节点类型的介绍，顺序节点介绍，实现该锁的核心原理，最后是源码分析。

### 1.使用demo ###

curator maven依赖引入

<dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>4.2.0</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>4.2.0</version>
    </dependency>

//创建
    CuratorFramework curatorFramework = CuratorFrameworkFactory
                .newClient("127.0.0.1:2181", new ExponentialBackoffRetry(1000, 3));
        curatorFramework.start();
            
    // 分布式锁
    InterProcessMutex lock = new InterProcessMutex(curatorFramework, "/lock");
    //获取锁
    if ( lock.acquire(1000, TimeUnit.MILLISECONDS) ){ 
        try{ 
            // 获取到了分布式锁
            // 获取所有子节点
            List<String> childNames = curatorFramework.getChildren().forPath("/lock");
            // 打印查看
            System.out.println(childNames);
            // do some work inside of the critical section here
        }finally { 
            //释放锁
            lock.release();
        }
    }

打印结果

[_c_54fe8bae-b49c-4a7b-b99e-733f463e4bed-lock-0000000000]

### 2.zk的节点类型介绍 ###

PERSISTENT 持久化节点，客户端断开连接，节点还会存在

PERSISTENT\_SEQUENTIAL 持久化顺序节点，除了有持久化节点的特性，同时创建的时候zk会为节点添加个序列

EPHEMERAL 临时节点，客户端断开连接，节点就会被删除

EPHEMERAL\_SEQUENTIAL 临时顺序节点，有临时节点的特性，同时创建的时候zk会为节点添加个序列

### 3.顺序节点演示 ###

String s = curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/test/node");
            List<String> strings = curatorFramework.getChildren().forPath("/test");
            System.out.println(strings);

我这里使用的持久顺序节点，多执行几次，查看结果

[node0000000002, node0000000003, node0000000000, node0000000001]

可以看到，zk为你创建的节点是 你设置的节点名+序列

### 4.核心原理 ###

1.  获取锁，最最最核心的就是创建临时顺序节点，然后看看创建的那个节点的序列是不是排在第一个，如果是排在第一个，就说明获取了锁，如果不是第一个，就对前一个节点添加监听器，然后等待，如果前一个节点发生变化，就获取唤醒等待线程，继续这个逻辑。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
    多个进程/线程同时创建顺序节点， 然后就会生成如上图的各个带有序列的节点，排在一个的节点对应的就会获取到锁。比如说进程 某线程1创建的xxx000000节点，然后这个进程某线程1就会成功获取到锁。
2.  释放锁，最最最核心的就是删除自己创建的那个临时循序节点。

### 5.源码分析 ###

#### 初始化 ####

InterProcessMutex lock = new InterProcessMutex(curatorFramework, "/lock");

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

这里StandardLockInternalsDriver 需要注意下，后面会用到driver

public InterProcessMutex(CuratorFramework client, String path, LockInternalsDriver driver){ 
        this(client, path, LOCK_NAME, 1, driver);
    }

这里的这个1是maxLeases。

InterProcessMutex(CuratorFramework client, String path, String lockName, int maxLeases, LockInternalsDriver driver){ 
        // 我们设置的那个path只是个base path
        basePath = PathUtils.validatePath(path);
        internals = new LockInternals(client, driver, path, lockName, maxLeases);
    }

#### 获取锁 ####

public boolean acquire(long time, TimeUnit unit) throws Exception{ 
        return internalLock(time, unit);
    }

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 )
        { 
    
            // 如果获取到了，就封装放到map中，这一块可以做重入锁
            LockData newLockData = new LockData(currentThread, lockPath);
            threadData.put(currentThread, newLockData);
            return true;
        }
    
        return false;
    }

这个地方有个并发map存线程与lockData的对应关系，先去map中获取当前线程对应的lockData,如果存在说明当前线程已经获取到了分布式锁，而且是还没有释放的，这个时候获取锁其实就是对lockData的计数器+1，这里其实就是为了实现重入锁的功能，在释放锁的时候，也是先减这个计数器。如果并发map中没有缓存当前线程的lockData数据，就会调用internals.attemptLock(time, unit, getLockNodeBytes())方法获取分布式锁，这块我们先假设已经获取到了锁，再往后看，lockPath不是null，说明获取到了锁，就会封装一个lockData，然后缓存到map中。

接下来我们看下这个internals.attemptLock(time, unit, getLockNodeBytes()) 方法是怎样获取锁的。

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{ 
                //先是创建节点 EPHEMERAL_SEQUENTIAL 临时顺序节点
                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;
    }

这段代码有点长，前面定义的一堆变量看看就好，就是一堆标识，hasTheLock 是否获取到了锁，接着下面的while循环，首先是调用driver.createsTheLock()方法创建EPHEMERAL\_SEQUENTIAL临时顺序节点，返回创建的path，接着就是调用internalLockLoop 方法本地获取锁，该方法其实就是看看你创建的这个path的序列是不是排在第一个，如果排在第一个说明已经获取到了锁，如果不是第一个的话，就获取到它前面一个序列的path，然后添加监听器，，该线程进行wait等待，如果前一个path发生了变化，就会通知到过来唤醒该线程，同时返回是否获取到锁，如果获取到锁直接返回自己的那个path就行了。  
接下来我们看下createsTheLock方法是怎样创建这个所节点的。这个方法是在StandardLockInternalsDriver类里面，直接看下

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

这里这个判断不用刻意关心，直接看里面的就好，可以看到是创建了一个EPHEMERAL\_SEQUENTIAL类型的节点，也就是临时顺序节点，最后返回这个path。

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();
                // 获取到自己的那个序列
                // 这里就是取到zk生成的哪个_c_d6df7f45-200e-4935-89d5-0e6c96bf1cdc-lock-0000000002
                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;
                }else{ 
                    // 如果没有的话，就获取下前一个序列的 path
                    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 ) { 
                                    // timed out - delete our node 超时，删除我们的节点
                                    // 删除节点
                                    doDelete = true;
                                    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 ){ 
                // 删除我们创建的path
                deleteOurPath(ourPath);
            }
        }
        return haveTheLock;
    }

这个方法也很长，但是内容不难，主要是干了3件事，1是获取到我们设置的/lock节点下面的顺序子节点全部获取到，然后根据序列进行从小到大排序， 2就是判断我们创建的节点在哪个子节点集合的哪个位置，如果是在0索引位置处，就说明已经获取到锁了，如果不在，就将它前一个序列的节点path返回。3是监听前一个序列的节点，然后当前线程wait等待，如果前一个节点发生了变化，就会通知过来，接着就会唤醒该线程。就可以继续执行了。

#### 释放锁 ####

我们再来看下释放锁的release方法实现。

@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);
    
        // 获取到锁，就会缓存到map中
        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);
        }
        // =0的情况
        try{ 
            // 释放锁
            internals.releaseLock(lockData.lockPath);
        }finally{ 
            // 从map中移除
            threadData.remove(currentThread);
        }
    }

先从并发map中根据当前线程获取对应的lockData，如果不存在抛出异常，先对lockData里面的计数器-1，如果大于0直接return，这种情况就是有重入的，如果是小于0 的话，抛出异常，这种情况一般不会存在，接着再往下走就是=0的情况了，这个时候就是需要与zk交互，调用releaseLock方法删除对应的节点，最后从并发map中移除这个线程对应的lockData。  
接下来我们看下调用releaseLock方法干了啥，其实就是删除对应path的节点。

final void releaseLock(String lockPath) throws Exception { 
        // 移除watcher
        client.removeWatchers();
        // 重置为空
        revocable.set(null);
        // 删除这个我们这个path
        deleteOurPath(lockPath);
    }

这里我们主要看下deleteOurPath方法。

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

走到这里，就能看到删除节点的代码了。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221022/66de8a03eabf47db866515de4c297db8.png