Multi Shared Lock

一个管理着多把锁的容器。当这个容器的acquire()方法调用的时候，相当于申请容器内所有的锁。如果申请失败，那么已持有的锁会被释放掉。

而调用容器的release()方法，则意味着释放容器内所有的锁。（失败时则被忽略）

1. 关键 API

org.apache.curator.framework.recipes.locks.InterProcessMultiLock

org.apache.curator.framework.recipes.locks.InterProcessLock

2. 机制说明

多把锁放入一个容器中，统一管理。

适用于一个任务需要持有多个锁才可进行的场景。

3. 用法

3.1 创建

托管已经创建好的锁对象
- 可以是任何锁类型

public InterProcessMultiLock(List
    
      locks){}

自动创建锁对象
- InterProcessMutexes
- 可重入分布式锁

public InterProcessMultiLock(CuratorFramework client,                             List
    
      paths)

4. 错误处理

在实际使用中，必须考虑链接问题。强烈建议：添加一个ConnectionStateListener用以处理链接中断或者丢失的情况

如果遇到链接中断SUSPENDED，在恢复链接RECONNECTED之前,就不能保证是不是还持有锁了。而如果链接丢失LOST，那就意味着不再持有锁了。

5. 源码分析

5.1 类定义

public class InterProcessMultiLock implements InterProcessLock {}

虽然也是实现了org.apache.curator.framework.recipes.locks.InterProcessLock,但是InterProcessMultiLock并不是一个实际意义上的锁。

实现这个接口，更多意义上可以当作一个委托或者代理模式。

5.2 成员变量

public class InterProcessMultiLock implements InterProcessLock{    private final List
    
      locks;}

只有一个被管理锁对象的集合

5.3 构造器

public InterProcessMultiLock(List
    
      locks){    this.locks = ImmutableList.copyOf(locks);}public InterProcessMultiLock(CuratorFramework client, List
     
       paths){    // paths get checked in each individual InterProcessMutex, so trust them here    this(makeLocks(client, paths));}

已有锁集合
- 做了一份不可变的集合，赋值

创建一组锁
- 由makeLocks方法负责创建

private static List
    
      makeLocks(CuratorFramework client, List
     
       paths){    ImmutableList.Builder
      
        builder = ImmutableList.builder();    for ( String path : paths )    {        InterProcessLock        lock = new InterProcessMutex(client, path);        builder.add(lock);    }    return builder.build();}

注意此方法是私用静态方法，也算固定套路
- 极限情况下，静态方法的调用效率更高（省略了this检查）

对paths逐一构建InterProcessMutex对象

将锁对象加入管理的集合中

5.4 加锁

public void acquire() throws Exception{    acquire(-1, null);}public boolean acquire(long time, TimeUnit unit) throws Exception{    Exception                   exception = null;    List
    
           acquired = Lists.newArrayList();    boolean                     success = true;    for ( InterProcessLock lock : locks )    {        try        {            if ( unit == null )            {                lock.acquire();                acquired.add(lock);            }            else            {                if ( lock.acquire(time, unit) )                {                    acquired.add(lock);                }                else                {                    success = false;                    break;                }            }        }        catch ( Exception e )        {            ThreadUtils.checkInterrupted(e);            success = false;            exception = e;        }    }    if ( !success )    {        for ( InterProcessLock lock : reverse(acquired) )        {            try            {                lock.release();            }            catch ( Exception e )            {                ThreadUtils.checkInterrupted(e);                // ignore            }        }    }    if ( exception != null )    {        throw exception;    }        return success;}

逐一对锁对象进行加锁操作
1. 如果加锁成功，记录状态
2. 如果加锁失败，则中止加锁

检查状态
1. 如果加锁过程不是全部成功
2. 则释放已经持有的锁
  - 此过程中的异常会被忽略

如果加锁过程出现异常，也需要对已经持有的锁进行释放

由于锁对象本身就可进行互斥
- 所以此方法没有进行并发控制

5.5 释放锁

public synchronized void release() throws Exception{    Exception       baseException = null;    for ( InterProcessLock lock : reverse(locks) )    {        try        {            lock.release();        }        catch ( Exception e )        {            ThreadUtils.checkInterrupted(e);            if ( baseException == null )            {                baseException = e;            }            else            {                baseException = new Exception(baseException);            }        }    }    if ( baseException != null )    {        throw baseException;    }}

逐一释放的过程

锁集合是先进过倒序处理的
- 先获得的锁，后释放
- 保障多个锁的依赖安全

释放锁的过程加上了synchronized
- 这个时候其实对于所有锁的状态是不确保统一的（有些锁可能已经被释放）
- 一个批次的释放过程，异常需要处理

5.6 锁状态判断

public synchronized boolean isAcquiredInThisProcess(){    // it's subjective what the correct meaning is here - I choose to return true    // only if all of the locks are acquired    for ( InterProcessLock lock : locks )    {        if ( !lock.isAcquiredInThisProcess() )        {            return false;        }    }    return true;}

同样，这个方法也是加上了synchronized的。需要保障方法内部锁状态的统一。

只有所有锁都加锁成功，才认为获得了锁。

转载于:https://my.oschina.net/roccn/blog/912701

你可能感兴趣的文章