Curator 源码初探（二） Semaphore

亦凉 2022-11-03 11:56 140阅读 0赞

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    		CuratorFramework client = CuratorFrameworkFactory.newClient(
    				"127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183",
    				retryPolicy);
    		client.start();
    
    		InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(
    				client, "/semaphores/semaphore_01", 3); 
    		Lease lease = semaphore.acquire();
    		Thread.sleep(3000); 
    		semaphore.returnLease(lease);

锁初始化

private InterProcessSemaphoreV2(CuratorFramework client, String path, int maxLeases, SharedCountReader count)
        {
            this.client = client;
            //注意这里，发现底层依赖于了InterProcessMutex 
            lock = new InterProcessMutex(client, ZKPaths.makePath(path, LOCK_PARENT));
            //设置maxLeases 就是信号量 数量
            this.maxLeases = (count != null) ? count.getCount() : maxLeases;
            leasesPath = ZKPaths.makePath(path, LEASE_PARENT);
    
            if ( count != null )
            {
                count.addListener
                    (
                        new SharedCountListener()
                        {
                            @Override
                            public void countHasChanged(SharedCountReader sharedCount, int newCount) throws Exception
                            {
                                InterProcessSemaphoreV2.this.maxLeases = newCount;
                            }
    
                            @Override
                            public void stateChanged(CuratorFramework client, ConnectionState newState)
                            {
                                // no need to handle this here - clients should set their own connection state listener
                            }
                        }
                    );
            }
        }

尝试获取锁

public Lease acquire() throws Exception
        {
            Collection<Lease> leases = acquire(1, 0, null);
            return leases.iterator().next();
        }

public Collection<Lease> acquire(int qty, long time, TimeUnit unit) throws Exception
        {
            long startMs = System.currentTimeMillis();
            boolean hasWait = (unit != null);
            long waitMs = hasWait ? TimeUnit.MILLISECONDS.convert(time, unit) : 0;
    
            Preconditions.checkArgument(qty > 0, "qty cannot be 0");
    
            ImmutableList.Builder<Lease> builder = ImmutableList.builder();
            boolean success = false;
            try
            {
                while ( qty-- > 0 )
                {
                    int retryCount = 0;
                    long startMillis = System.currentTimeMillis();
                    boolean isDone = false;
                    while ( !isDone )
                    {
                        switch ( internalAcquire1Lease(builder, startMs, hasWait, waitMs) )
                        {
                            case CONTINUE:
                            {
                                isDone = true;
                                break;
                            }
    
                            case RETURN_NULL:
                            {
                                return null;
                            }
    
                            case RETRY_DUE_TO_MISSING_NODE:
                            {
                                // gets thrown by internalAcquire1Lease 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()) )
                                {
                                    throw new KeeperException.NoNodeException("Sequential path not found - possible session loss");
                                }
                                // try again
                                break;
                            }
                        }
                    }
                }
                success = true;
            }
            finally
            {
                if ( !success )
                {
                    returnAll(builder.build());
                }
            }
    
            return builder.build();
        }

private InternalAcquireResult internalAcquire1Lease(ImmutableList.Builder<Lease> builder, long startMs, boolean hasWait, long waitMs) throws Exception
        {
            if ( client.getState() != CuratorFrameworkState.STARTED )
            {
                return InternalAcquireResult.RETURN_NULL;
            }
    
            if ( hasWait )
            {
                long thisWaitMs = getThisWaitMs(startMs, waitMs);
                if ( !lock.acquire(thisWaitMs, TimeUnit.MILLISECONDS) )
                {
                    return InternalAcquireResult.RETURN_NULL;
                }
            }
            else
            {
                //private final InterProcessMutex lock; 
                //底层还是基于一般锁来实现的
                //逻辑是什么 就是底层去判断创建的node节点的排序是否< maxLeases ，否则添加一个watcher 然后wait（）
                lock.acquire();
            }
            try
            {
                PathAndBytesable<String> createBuilder = client.create().creatingParentsIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL);
                String path = (nodeData != null) ? createBuilder.forPath(ZKPaths.makePath(leasesPath, LEASE_BASE_NAME), nodeData) : createBuilder.forPath(ZKPaths.makePath(leasesPath, LEASE_BASE_NAME));
                //然后又去创建了一个 /semaphores/semaphore_01/leases/sdasdadasdasdas-lease-00000 节点
                String nodeName = ZKPaths.getNodeFromPath(path);
                builder.add(makeLease(path));
    
                synchronized(this)
                {
                    for(;;)
                    {
                        //获取/semaphores/semaphore_01/leases 下的所有节点
                        List<String> children = client.getChildren().usingWatcher(watcher).forPath(leasesPath);
                        if ( !children.contains(nodeName) )
                        {
                            log.error("Sequential path not found: " + path);
                            return InternalAcquireResult.RETRY_DUE_TO_MISSING_NODE;
                        }
                        //如果超过了 maxLease 数量
                        if ( children.size() <= maxLeases )
                        {
                            break;
                        }
                        
                        if ( hasWait )
                        {
                            long thisWaitMs = getThisWaitMs(startMs, waitMs);
                            if ( thisWaitMs <= 0 )
                            {
                                return InternalAcquireResult.RETURN_NULL;
                            }
                            wait(thisWaitMs);
                        }
                        else
                        {
                            wait();
                        }
                    }
                }
            }
            finally
            
            {
                   //如果小于等于3 就会将之前加的锁释放掉
                lock.release();
            }
            return InternalAcquireResult.CONTINUE;
        }

大家思考一下，为什么要先去获取/semaphores/semaphore\_01/lock 锁之后，才能继续去/semaphores/semaphore\_01/leases/ 节点下去操作，去进入信号量获取？？？？？

为什么不直接通过/locks/lock\_01 中允许前3个序号的元素去直接操作？

因为如果第4个线程进来，或发现size > maxLeases数量，然后添加一个watcher然后进入阻塞，不会释放lock 的锁，剩下的 5 6 7...后续线程都会被阻塞在lock锁那边，而不是尝试获取lease 信号线这边,这样的好处是什么？

（1）能够保证 公平的顺序的获取信号量，

（2）信号量释放的时候不必通知过多的watcher，进行回调，只需要回调一个就ok

（3）使得程序的实现简单清晰，这充分体现了框架编写者的水平真的很优秀

释放信号量，其实很简单，直接将连接关闭就ok了，zk会自动的删除节点信息，然后触发watcher 去通知lease中的阻塞node，然后尝试获取信号量

/**
         * Convenience method. Closes the lease
         *
         * @param lease lease to close
         */
        public void returnLease(Lease lease)
        {
            Closeables.closeQuietly(lease);
        }