Apache Curator之InterProcessMutex源碼分析（四）

上篇文章通過秒購的例子對InterProcessMutex鎖有了初步認識，本文將通過對源碼進行分析帶你進入分布式鎖的世界。

老規矩先上圖，為了更清晰的了解獲取鎖，釋放鎖的過程，下圖簡化了一些細節，使整個流程更加通暢。

下面將逐個方法去分析。

InterProcessMutex.acquire()

 

    @Override
       //獲得分布式鎖，阻塞
    public void acquire() throws Exception {
        if (!internalLock(-1, null)) {
            throw new IOException("Lost connection while trying to acquire lock: " + basePath);
        }
    }

       //獲得分布式鎖，在指定的時間內阻塞，推薦使用
    @Override
    public boolean acquire(long time, TimeUnit unit) throws Exception {
        return internalLock(time, unit);
    }

 

InterProcessMutex.internalLock(long time, TimeUnit unit)

    private boolean internalLock(long time, TimeUnit unit) throws Exception {

        Thread currentThread = Thread.currentThread();
        //獲得當前線程的鎖對象
        LockData lockData = threadData.get(currentThread);
        if (lockData != null) {
            // re-entering
               //如果鎖不為空，當前線程已經獲得鎖，可重入鎖，lockCount++   
            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.attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes)

    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 {
                   //在當前path下創建臨時有序節點
                ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
                   //判斷是不是序號最小的節點，如果是返回true，否則阻塞等待
                hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
            } catch (KeeperException.NoNodeException e) {
                if (client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper())) {
                    isDone = false;
                } else {
                    throw e;
                }
            }
        }

        if (hasTheLock) {
            return ourPath;
        }

        return null;
    }

StandardLockInternalsDriver.createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes)

    @Override
    public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) throws Exception {
        String ourPath;
           //在zookeeper的指定路徑上，創建一個臨時序列節點。只是純粹的創建了一個節點，並不是說線程已經持有了鎖。　　
        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.internalLockLoop(long startMillis, Long millisToWait, String 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();
　　　　　　　　　　　　//同步，是為了實現公平鎖
                    synchronized (this) {
                        try {
                               //給比自己小的節點設置監聽器
                            client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                            if (millisToWait != null) {
                                millisToWait -= (System.currentTimeMillis() - startMillis);
                                startMillis = System.currentTimeMillis();
                                   //等待超時，超時刪除臨時節點，超時時間是acquire方法傳入的參數
                                if (millisToWait <= 0) {
                                    doDelete = true;    // timed out - delete our node
                                    break;
                                }
                                //沒有超時，繼續等待
                                wait(millisToWait);
                            } else {
                                   //如果等待時間==null，一直阻塞等待
                                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;
    }

LockInternals.getSortedChildren()

    List<String> getSortedChildren() throws Exception {
        return getSortedChildren(client, basePath, lockName, driver);
    }

    public static List<String> getSortedChildren(CuratorFramework client, String basePath, final String lockName, final LockInternalsSorter sorter) throws Exception {
        try {
               //獲得basePath下所有子節點，進行排序
            List<String> children = client.getChildren().forPath(basePath);
            List<String> sortedList = Lists.newArrayList(children);
            Collections.sort(sortedList, new Comparator<String>() {
                @Override
                public int compare(String lhs, String rhs) {
                    return sorter.fixForSorting(lhs, lockName).compareTo(sorter.fixForSorting(rhs, lockName));
                }
            });
            return sortedList;
        } catch (KeeperException.NoNodeException ignore) {
            return Collections.emptyList();
        }
    }

StandardLockInternalsDriver.getsTheLock(CuratorFramework c, List<String> child, String nodeName, int max)

    @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.deleteOurPath(String ourPath)

    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.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);
           //如果當前線程沒有持有鎖，不能釋放
        if (lockData == null) {
            throw new IllegalMonitorStateException("You do not own the lock: " + basePath);
        }

           //重入鎖計數減一，如果還大於0，不能釋放。直到所有重入業務完成，計數為0才能釋放
        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);
        }
    }

LockInternals.release()

    final void releaseLock(String lockPath) throws Exception {
        client.removeWatchers();//刪除監聽
        revocable.set(null);
        deleteOurPath(lockPath);//刪除Path
    }