Curator实现分布式锁的基本原理

Posted 朱培

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Curator实现分布式锁的基本原理相关的知识,希望对你有一定的参考价值。

curator是Netflix公司开源的一个ZooKeeper客户端封装。curator 对于锁这块做了一些封装,curator 提供了InterProcessMutex 这样一个 api。除了分布式锁之外,还提供了 leader 选举、分布式队列等常用的功能。本文主要以InterProcessMutex为例,介绍一下这个分布式可重入排它锁的实现原理。

Curator的几种锁方案:

  • InterProcessMutex:分布式可重入排它锁
  • InterProcessSemaphoreMutex:分布式排它锁
  • InterProcessReadWriteLock:分布式读写锁
  • InterProcessMultiLock:将多个锁作为单个实体管理的容器

接下来我们以InterProcessMutex为例,介绍一下这个分布式可重入排它锁的实现原理

1、实例化InterProcessMutex

Zookeeper 利用 path 创建临时顺序节点,实现公平锁的核心,也是最常用的一个构造函数。

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

maxLeases=1,表示可以获得分布式锁的线程数量(跨 JVM)为 1,即为互斥锁

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

internals 的类型为 LockInternals ,InterProcessMutex 将分布式锁的申请和释放操作委托给internals 执行

InterProcessMutex(CuratorFramework client, String path, String lockName, int maxLeases, LockInternalsDriver driver) 
    basePath = PathUtils.validatePath(path);
    internals = new LockInternals(client, driver, path, lockName, maxLeases);

2、InterProcessMutex.acquire

实例化完成的InterProcessMutex对象,开始调用acquire()方法来尝试加锁:

当时间为-1,表示无限等待

public void acquire() throws Exception

    if ( !internalLock(-1, null) )
    
        throw new IOException("Lost connection while trying to acquire lock: " + basePath);
    

指定获得锁最大的等待的时间,抢夺时,如果出现堵塞,会在超过该时间后,返回false。通过设置整个超时时间,避免出现大量的临时节点累积以及线程堵塞的问题。

public boolean acquire(long time, TimeUnit unit) throws Exception

    return internalLock(time, unit);

3、InterProcessMutex.internalLock

锁的可重入。每个InterProcessMutex实例,都会持有一个ConcurrentMap类型的threadData对象,以线程对象作为Key,以LockData作为Value值。

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

    Thread currentThread = Thread.currentThread();

    LockData lockData = threadData.get(currentThread);
    if ( lockData != null )
    
       // 实现可重入,同一线程再次 acquire,首先判断当前的映射表内(threadData)是否有该线程的锁信息,如果有则原子+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;

4 LockInternals.attemptLock

抢夺锁,尝试获取锁,并返回锁对应的 Zookeeper 临时顺序节点的路径。在 Zookeeper 中创建的临时顺序节点的路径,相当于一把待激活的分布式锁, 激活条件:同级目录子节点,名称排序最小(排队,公平锁)。

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception

    final long      startMillis = System.currentTimeMillis();
	// 无限等待时,millisToWait 为 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;

    String          ourPath = null;
    boolean         hasTheLock = false;  // 是否已经持有分布式锁
    boolean         isDone = false;   // 是否已经完成尝试获取分布式锁的操作
    while ( !isDone )
    
        isDone = true;

        try
        
			// 从 InterProcessMutex 的构造函数可知实际 driver 为 StandardLockInternalsDriver 的实例
			// 在Zookeeper中创建临时顺序节点
            ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
			// 循环等待来激活分布式锁,实现锁的公平性
            hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
        
        catch ( KeeperException.NoNodeException e )
        
            // 因 为 会 话 过 期 等 原 因 ,StandardLockInternalsDriver 因为无法找到创建的临时顺序节点而抛出 NoNodeException 异常
            if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
            
				// 满足重试策略尝试重新获取锁
                isDone = false;
            
            else
            
				// 不满足重试策略则继续抛出NoNodeException
                throw e;
            
        
    

    if ( hasTheLock )
    
		// 如果成功获得分布式锁,返回临时顺序节点的路径,上层将其封装成锁信息记录在映射表,方便锁重入
        return ourPath;
    
	// 如果获取分布式锁失败,返回 null
    return null;

5 driver.createsTheLock

这个driver的createsTheLock方法就是在创建这个锁,即在zookeeper的指定路径上,创建一个临时序列节点。注意:此时只是纯粹的创建了一个节点,不是说线程已经持有了锁。

public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) throws Exception

    String ourPath;
	// lockNodeBytes 不为 null 则作为数据节点内容,否则采用默认内容(IP 地址)
    if ( lockNodeBytes != null )
    
		// creatingParentContainersIfNeeded:用于创建父节点,如果不支持 CreateMode.CONTAINER,那么将采用 CreateMode.PERSISTENT,
		//CreateMode.EPHEMERAL_SEQUENTIAL:临时顺序节点,Zookeeper 能保证在节点产生的顺序性
		// 依据顺序来激活分布式锁,从而也实现了分布式锁的公平性,
        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;

6 LockInternals.internalLockLoop

循环等待来激活分布式锁,实现锁的公平性,判断自身是否能够持有锁。如果不能,进入wait,等待被唤醒。

private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception

    boolean     haveTheLock = false;  // 是否已经持有分布式锁,默认为false
    boolean     doDelete = false;  // 是否需要删除子节点
    try
    
        if ( revocable.get() != null )
        
            client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
        

        while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )
        
			// 获取排序后的子节点列表,并且从小到大根据节点名称后10位数字进行排序。
            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 
                    
                       // 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 )
    
		// 标记删除,在 finally 删除之前创建的临时顺序节点(后台不断尝试)
        ThreadUtils.checkInterrupted(e);
        doDelete = true;
		// 重新抛出,尝试重新获取锁
        throw e;
    
    finally
    
        if ( doDelete )
        
            deleteOurPath(ourPath);
        
    
    return haveTheLock;

7 driver.getsTheLock

进入 StandardLockInternalsDriver中。判断是否可以持有锁,判断规则:当前创建的节点是否在上一步获取到的子节点列表的首位。

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;
	// 如果已经获得了锁,则无需监听任何节点,否则需要监听上一顺序节点(ourIndex-1)
	// 因 为 锁 是 公 平 的 , 因 此 无 需 监 听 除 了(ourIndex-1)以外的所有节点,这是为了减少羊群效应
    String          pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);
	// 返回获取锁的结果,交由上层继续处理(添加监听等操作)
    return new PredicateResults(pathToWatch, getsTheLock);


static void validateOurIndex(String sequenceNodeName, int ourIndex) throws KeeperException

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

以上就是整个锁的抢夺过程,下面来分析锁释放的逻辑

8 释放锁,InterProcessMutex.release

释放的逻辑比较简单,首先减少重入锁的计数,直到变成0。然后释放锁,即移除移除Watchers & 删除创建的节点,最后从threadData中,删除自己线程的缓存。

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,原子-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);
    

扩展:Zookeeper 数据的同步流程

当整个集群在启动时,或者当 leader 节点出现网络中断、崩溃等情况时,ZAB 协议就会进入恢复模式并选举产生新的 Leader,当 leader 服务器选举出来后,并且集群中有过半的机器和该 leader 节点完成数据同步后(同步指的是数据同步,用来保证集群中过半的机器能够和 leader 服务器的数据状态保持一致),ZAB 协议就会退出恢复模式。当集群中已经有过半的 Follower 节点完成了和 Leader 状态同步以后,那么整个集群就进入了消息广播模式。这个时候,在 Leader 节点正常工作时,启动一台新的服务器加入到集群,那这个服务器会直接进入数据恢复模式,和leader 节点进行数据同步。同步完成后即可正常对外提供非事务请求的处理。需要注意的是:leader 节点可以处理事务请求和非事务请求,follower 节点只能处理非事务请求,如果 follower 节点接收到非事务请求,会把这个请求转发给 Leader 服务器。

消息广播的实现原理:

消息广播的过程实际上是一个简化版本的二阶段提交过程

  1. leader 接收到消息请求后,将消息赋予一个全局唯一的64 位自增 id,叫:zxid,通过 zxid 的大小比较既可以实现因果有序这个特征
  2. leader 为每个 follower 准备了一个 FIFO 队列(通过 TCP协议来实现,以实现了全局有序这一个特点)将带有 zxid的消息作为一个提案(proposal)分发给所有的 follower
  3. 当 follower 接收到 proposal,先把 proposal 写到磁盘,写入成功以后再向 leader 回复一个 ack
  4. 当 leader 接收到合法数量(超过半数节点)的 ACK 后,leader 就会向这些 follower 发送 commit 命令,同时会在本地执行该消息
  5. 当 follower 收到消息的 commit 命令以后,会提交该消息

leader 的投票过程,不需要 Observer 的 ack,也就是Observer 不需要参与投票过程,但是 Observer 必须要同步 Leader 的数据从而在处理请求的时候保证数据的一致性

总结

总结:InterProcessMutex通过在zookeeper的某路径节点下创建临时序列节点来实现分布式锁,即每个线程(跨进程的线程)获取同一把锁前,都需要在同样的路径下创建一个节点,节点名字由uuid + 递增序列组成。而通过对比自身的序列数是否在所有子节点中最小的一位,来判断是否成功获取到了锁。当获取锁失败时,它会添加watcher来监听前一个节点的变动情况,然后进行等待状态。直到watcher的事件生效将自己唤醒,或者超时时间异常返回。

以上是关于Curator实现分布式锁的基本原理的主要内容,如果未能解决你的问题,请参考以下文章

Curator实现分布式锁的基本原理

Curator实现zookeeper分布式锁的基本原理

从Curator实现分布式锁的源码再到羊群效应

浅谈Zookeeper客户端库Curator实现加锁的原理

浅谈Zookeeper客户端库Curator实现加锁的原理

curator 分布式锁InterProcessMutex