1 前言

最近在做项目的时候，遇到一个多线程访问申请一个资源的问题。需要每个线程都能够最终在有效的时间内申请到或者超时失败。以前一般这种方式用的是redis做枷锁机制，每个线程都去redis加一把共同的锁，如果枷锁成功，则执行资源申请操作。而没有枷锁成功的线程，则在有效时间内循环尝试去枷锁，并且每次木有加锁成功，则就Thread.sleep()一会。 

通过这种方式可以看出，这里对于sleep的时间间隔要求设置很严格，如果太小，则就会增加大规模的redis请求操作；而如果太长，当资源可用的时候，但是线程若都在sleep，则会出现资源空闲等待，线程不能及时的使用资源。

基于上面这种原因，开始换种用curator的分布式锁机制来实现互斥锁。

2 简介

curator的特性，不必多说，可以说是 curator是zookeeper中的类似于guava对于java的意义一样，提供了丰富的封装，异常处理，提供了fluent编程模型，提供了master选举，分布式锁，分布式基数，分布式barrier，可以很方便的为日常生产所使用。

 这次主要用了curator的InterProcessMutex这种互斥锁来做，也借于此机会，阅读了它的代码，分享一下它的机制，共大家一起来交流学习下。


3 InterProcessMutex基本简介

InterProcessMutex一个可重入锁，提供分布式锁的入口服务。基本的构造过程如下:

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

构造器内部最终会构造一个

internals = new LockInternals(client, driver, path, lockName, maxLeases);

LockInternals这个是所有申请锁与释放锁的核心实现


4 InterProcessMutex的获取锁

InterProcessMutex.internalLock()提供两种机制来加锁，

第一种是无限等待，直到获取到锁。第二种是有限等待，在规定的时间内获取锁，如果木有失败。

该方法内部简略如下:

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 )
{
    LockData   newLockData = new LockData(currentThread, lockPath);
    threadData.put(currentThread, newLockData);
    return true;
}
return false;

对于同一个线程再次获取锁的时候，会判断当前线程是否已经拥有了，

如果拥有了，则直接做原子操作加1，然后返回true，这样就实现了可重入锁。

对于其他情况，则都会调用 LockInternals.attemptLock();

4.1  LockInternals.attemptLock()

1）根据传入的超时做判断，是否需要millisToWait设置

2）创建临时顺序节点路径:

ourPath = client.create().creatingParentsIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);

假如basepath=/zklock/activityno,这个是活动no的根路径,这个path是构建InterProcessMutex设置的。

则path是/zklock/activityno/lock-，注意这个path是在basepath下创建的，lock-是curator自动添加的

则创建的顺序节点如: /zklock/activityno/_c_f4a49d75-86f8-40b2-8b9c-d813392aa1db-lock-0000000008

尤其要注意这里，LockInternals会对所有请求获取锁的线程都会创建一个临时顺序节点，节点后缀顺序由zk来保证，

同时zk客户端底层能够保证同一个客户端发送的请求是按照顺序的，这样就能够保证同一个客户端先申请锁创建的后缀序号比后申请的编号小。

3）循环等待尝试枷锁internalLockLoop(startMillis, millisToWait, ourPath);

内部核心代码流程如下:

3.1) 获取所有的子节点

List<String>        children = getSortedChildren();

排序:获取basepath下所有的子节点,然后截取lock-后面的编号，做升序排序,注意这里的升序排序，保证了最先申请锁的排在最开始，公平策略是根据谁先申请那么你的优先级就应该最高

String  sequenceNodeName = ourPath.substring(basePath.length() + 1);

获取节点名如: _c_f4a49d75-86f8-40b2-8b9c-d813392aa1db-lock-0000000008

3.2 核心获取锁的判断.

根据拿到的所有子节点路径以及当前子节点去尝试获取锁。

maxLeases代表是租赁个数，对于分布式互斥锁，这里值为0，保证了只允许租赁一个。

这里就是获取锁的核心实现,若最终获取成功，则直接return，否则进行无限wait或者有限wait()

PredicateResults    predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
if ( predicateResults.getsTheLock() )
{
      haveTheLock = true;
}else{
    //wait等待，有限或者无限等待，注意这里用了线程的wait来做等待
    client.getData().usingWatcher(watcher).forPath(previousSequencePath);
}

根据driver.getsTheLock的结果，如果木有获取到，则就会watcher返回的path，然后根据传入的时间来做wait操作。


注意这里wait是互斥信号量是LockInternals. 自定义的watcher很简单，一旦监听的到的节点数据变更或删除，则就直接notifyFromWatcher();

driver.getsTheLock内部代码如下:

public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
 {
        //先根据子节点名获取在已经升序的list中的索引位置。
        int             ourIndex = children.indexOf(sequenceNodeName);
        validateOurIndex(sequenceNodeName, ourIndex);
        //比较索引位置，由于maxLeases=1,则只有ourIndex=0才成立，这样就可以用来判断当前子节点是否是升序第一个节点，并且也有很好的扩展性
        //可以改变maxLeases来允许同时租赁的数量
        //注意，这里升序是根据zk生成的顺序编号排序的，申请越早编号越小。
        boolean         getsTheLock = ourIndex < maxLeases;
        //若getsTheLock=true，表示获取到锁，否则获取它上一个位置的路径，注意这个路径会用来做watche的
        String          pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);
        return new PredicateResults(pathToWatch, getsTheLock);
}

5 InterProcessMutex的释放锁

主要是判断是否是当前线程，或者非当前线程。最终会根据线程号找到对应的path路径，然后直接删除该临时节点。

6 InterProcessMutex总结

1) curator的InterProcessMutex提供了多种锁机制，互斥锁，读写锁，以及可定时数的互斥锁

2）所有申请锁都会创建临时顺序节点，保证了都能够有机会去获取锁。

3）内部用了线程的wait()和notifyAll()这种等待机制，可以及时的唤醒最渴望得到锁的线程。

避免常规利用Thread.sleep()这种无用的间隔等待机制.

4) 利用redis做锁的时候，一般都需要做锁的有效时间限定。而curator则利用了zookeeper的临时顺序节点特性，

一旦客户端失去连接后，则就会自动清除该节点。