`

锁机制(三)(转)

 
阅读更多

不同的角度理解(^_^)

在理解J.U.C原理以及锁机制之前,我们来介绍J.U.C框架最核心也是最复杂的一个基础类:java.util.concurrent.locks.AbstractQueuedSynchronizer

 

AQS

AbstractQueuedSynchronizer,简称AQS,是J.U.C最复杂的一个类,导致绝大多数讲解并发原理或者实战的时候都不会提到此类。但是虚心的作者愿意借助自己有限的能力和精力来探讨一二(参考资源中也有一些作者做了部分的分析。)。

首先从理论知识开始,在了解了相关原理后会针对源码进行一些分析,最后加上一些实战来描述。

image

上面的继承体系中,AbstractQueuedSynchronizer是CountDownLatch/FutureTask/ReentrantLock/RenntrantReadWriteLock/Semaphore的基础,因此AbstractQueuedSynchronizer是Lock/Executor实现的前提。公平锁、不公平锁、Condition、CountDownLatch、Semaphore等放到后面的篇幅中说明。

完整的设计原理可以参考Doug Lea的论文 The java.util.concurrent Synchronizer Framework ,这里做一些简要的分析。

基本的思想是表现为一个同步器,支持下面两个操作:

获取锁:首先判断当前状态是否允许获取锁,如果是就获取锁,否则就阻塞操作或者获取失败,也就是说如果是独占锁就可能阻塞,如果是共享锁就可能失败。另外如果是阻塞线程,那么线程就需要进入阻塞队列。当状态位允许获取锁时就修改状态,并且如果进了队列就从队列中移除。

while(synchronization state does not allow acquire){

    enqueue current thread if not already queued;

    possibly block current thread;

}

dequeue current thread if it was queued;

释放锁:这个过程就是修改状态位,如果有线程因为状态位阻塞的话就唤醒队列中的一个或者更多线程。

update synchronization state;

if(state may permit a blocked thread to acquire)

    unlock one or more queued threads;

要支持上面两个操作就必须有下面的条件:

  • 原子性操作同步器的状态位
  • 阻塞和唤醒线程
  • 一个有序的队列

目标明确,要解决的问题也清晰了,那么剩下的就是解决上面三个问题。

状态位的原子操作

这里使用一个32位的整数来描述状态位,前面章节的原子操作的理论知识整好派上用场,在这里依然使用CAS操作来解决这个问题。事实上这里还有一个64位版本的同步器(AbstractQueuedLongSynchronizer),这里暂且不谈。

阻塞和唤醒线程

标准的JAVA API里面是无法挂起(阻塞)一个线程,然后在将来某个时刻再唤醒它的。JDK 1.0的API里面有Thread.suspend和Thread.resume,并且一直延续了下来。但是这些都是过时的API,而且也是不推荐的做法。

在JDK 5.0以后利用JNI在LockSupport类中实现了此特性。

LockSupport.park()
LockSupport.park(Object)
LockSupport.parkNanos(Object, long)
LockSupport.parkNanos(long)
LockSupport.parkUntil(Object, long)
LockSupport.parkUntil(long)
LockSupport.unpark(Thread)

上面的API中park()是在当前线程中调用,导致线程阻塞,带参数的Object是挂起的对象,这样监视的时候就能够知道此线程是因为什么资源而阻塞的。由于park()立即返回,所以通常情况下需要在循环中去检测竞争资源来决定是否进行下一次阻塞。park()返回的原因有三:

  • 其他某个线程调用将当前线程作为目标调用 unpark
  • 其他某个线程中断当前线程;
  • 该调用不合逻辑地(即毫无理由地)返回。

其实第三条就决定了需要循环检测了,类似于通常写的while(checkCondition()){Thread.sleep(time);}类似的功能。

有序队列

在AQS中采用CHL列表来解决有序的队列的问题。

imageAQS采用的CHL模型采用下面的算法完成FIFO的入队列和出队列过程。

对于入队列(enqueue):采用CAS操作,每次比较尾结点是否一致,然后插入的到尾结点中。

do {

        pred = tail;

}while ( !compareAndSet(pred,tail,node) );

对于出队列(dequeue):由于每一个节点也缓存了一个状态,决定是否出队列,因此当不满足条件时就需要自旋等待,一旦满足条件就将头结点设置为下一个节点。

while (pred.status != RELEASED) ;

head  = node;

实际上这里自旋等待也是使用LockSupport.park()来实现的。

AQS里面有三个核心字段:

private volatile int state;

private transient volatile Node head;

private transient volatile Node tail;

其中state描述的有多少个线程取得了锁,对于互斥锁来说state<=1。head/tail加上CAS操作就构成了一个CHL的FIFO队列。下面是Node节点的属性。

volatile int waitStatus; 节点的等待状态,一个节点可能位于以下几种状态:

  • CANCELLED = 1: 节点操作因为超时或者对应的线程被interrupt。节点不应该留在此状态,一旦达到此状态将从CHL队列中踢出。
  • SIGNAL = -1: 节点的继任节点是(或者将要成为)BLOCKED状态(例如通过LockSupport.park()操作),因此一个节点一旦被释放(解锁)或者取消就需要唤醒(LockSupport.unpack())它的继任节点。
  • CONDITION = -2:表明节点对应的线程因为不满足一个条件(Condition)而被阻塞。
  • 0: 正常状态,新生的非CONDITION节点都是此状态。
  • 非负值标识节点不需要被通知(唤醒)。

volatile Node prev;此节点的前一个节点。节点的waitStatus依赖于前一个节点的状态。

volatile Node next;此节点的后一个节点。后一个节点是否被唤醒(uppark())依赖于当前节点是否被释放。

volatile Thread thread;节点绑定的线程。

Node nextWaiter;下一个等待条件(Condition)的节点,由于Condition是独占模式,因此这里有一个简单的队列来描述Condition上的线程节点。

 

AQS 在J.U.C里面是一个非常核心的工具,而且也非常复杂,里面考虑到了非常多的逻辑实现,所以在后面的章节中总是不断的尝试介绍AQS的特性和实现。

这一个小节主要介绍了一些理论背景和相关的数据结构,在下一个小节中将根据以上知识来了解Lock.lock/unlock是如何实现的。

 

参考资料:

(1)ReentrantLock代码剖析之ReentrantLock.lock ReentrantLock代码剖析之ReentrantLock.unlock ReentrantLock代码剖析之ReentrantLock.lockInterruptibly

(2)java多线程--java.util.concurrent.locks.AbstractQueuedSynchronizer解析(只包含多线程同步示例)

(3)处理 InterruptedException

(4)AbstractQueuedSynchronizer源码解析之ReentrantLock(一)  AbstractQueuedSynchronizer源码解析之ReentrantLock(二)

(5)The java.util.concurrent Synchronizer Framework

 

接上篇,这篇从Lock.lock/unlock开始。特别说明在没有特殊情况下所有程序、API、文档都是基于JDK 6.0的。

public void java.util.concurrent.locks.ReentrantLock.lock()

获取锁。

如果该锁没有被另一个线程保持,则获取该锁并立即返回,将锁的保持计数设置为 1。

如果当前线程已经保持该锁,则将保持计数加 1,并且该方法立即返回。

如果该锁被另一个线程保持,则出于线程调度的目的,禁用当前线程,并且在获得锁之前,该线程将一直处于休眠状态,此时锁保持计数被设置为 1。

从上面的文档可以看出ReentrantLock是可重入锁的实现。而内部是委托java.util.concurrent.locks.ReentrantLock.Sync.lock()实现的。java.util.concurrent.locks.ReentrantLock.Sync是抽象类,有java.util.concurrent.locks.ReentrantLock.FairSync和java.util.concurrent.locks.ReentrantLock.NonfairSync两个实现,也就是常说的公平锁和不公平锁。

公平锁和非公平锁

如果获取一个锁是按照请求的顺序得到的,那么就是公平锁,否则就是非公平锁。

在没有深入了解内部机制及实现之前,先了解下为什么会存在公平锁和非公平锁。公平锁保证一个阻塞的线程最终能够获得锁,因为是有序的,所以总是可以按照请求的顺序获得锁。不公平锁意味着后请求锁的线程可能在其前面排列的休眠线程恢复前拿到锁,这样就有可能提高并发的性能。这是因为通常情况下挂起的线程重新开始与它真正开始运行,二者之间会产生严重的延时。因此非公平锁就可以利用这段时间完成操作。这是非公平锁在某些时候比公平锁性能要好的原因之一。

二者在实现上的区别会在后面介绍,我们先从公平锁(FairSync)开始。

前面说过java.util.concurrent.locks.AbstractQueuedSynchronizer (AQS)是Lock的基础,对于一个FairSync而言,lock()就直接调用AQS的acquire(int arg);

public final void acquire(int arg) 以独占模式获取对象,忽略中断。通过至少调用一次 tryAcquire(int) 来实现此方法,并在成功时返回。否则在成功之前,一直调用tryAcquire(int) 将线程加入队列,线程可能重复被阻塞或不被阻塞。

在介绍实现之前先要补充上一节的知识,对于一个AQS的实现而言,通常情况下需要实现以下方法来描述如何锁定线程。

  • tryAcquire(int) 试图在独占模式下获取对象状态。此方法应该查询是否允许它在独占模式下获取对象状态,如果允许,则获取它。

    此方法总是由执行 acquire 的线程来调用。如果此方法报告失败,则 acquire 方法可以将线程加入队列(如果还没有将它加入队列),直到获得其他某个线程释放了该线程的信号。也就是说此方法是一种尝试性方法,如果成功获取锁那最好,如果没有成功也没有关系,直接返回false。

  • tryRelease(int) 试图设置状态来反映独占模式下的一个释放。 此方法总是由正在执行释放的线程调用。释放锁可能失败或者抛出异常,这个在后面会具体分析。
  • tryAcquireShared(int) 试图在共享模式下获取对象状态。
  • tryReleaseShared(int) 试图设置状态来反映共享模式下的一个释放。
  • isHeldExclusively() 如果对于当前(正调用的)线程,同步是以独占方式进行的,则返回 true
  • 除了tryAcquire(int)外,其它方法会在后面具体介绍。首先对于ReentrantLock而言,不管是公平锁还是非公平锁,都是独占锁,也就是说同时能够有一个线程持有锁。因此对于acquire(int arg)而言,arg==1。在AQS中acquire的实现如下:

    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

    这个看起来比较复杂,我们分解以下4个步骤。

    1.  
      1. 如果tryAcquire(arg)成功,那就没有问题,已经拿到锁,整个lock()过程就结束了。如果失败进行操作2。
      2. 创建一个独占节点(Node)并且此节点加入CHL队列末尾。进行操作3。
      3. 自旋尝试获取锁,失败根据前一个节点来决定是否挂起(park()),直到成功获取到锁。进行操作4。
      4. 如果当前线程已经中断过,那么就中断当前线程(清除中断位)。

    这是一个比较复杂的过程,我们按部就班一个一个分析。

    tryAcquire(acquires)

    对于公平锁而言,它的实现方式如下:

        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (isFirst(current) &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }

    在这段代码中,前面说明对于AQS存在一个state来描述当前有多少线程持有锁。由于AQS支持共享锁(例如读写锁,后面会继续讲),所以这里state>=0,但是由于ReentrantLock是独占锁,所以这里不妨理解为0<=state,acquires=1。isFirst(current)是一个很复杂的逻辑,包括踢出无用的节点等复杂过程,这里暂且不提,大体上的意思是说判断AQS是否为空或者当前线程是否在队列头(为了区分公平与非公平锁)。

    1.  
      1. 如果当前锁有其它线程持有,c!=0,进行操作2。否则,如果当前线程在AQS队列头部,则尝试将AQS状态state设为acquires(等于1),成功后将AQS独占线程设为当前线程返回true,否则进行2。这里可以看到compareAndSetState就是使用了CAS操作。
      2. 判断当前线程与AQS的独占线程是否相同,如果相同,那么就将当前状态位加1(这里+1后结果为负数后面会讲,这里暂且不理它),修改状态位,返回true,否则进行3。这里之所以不是将当前状态位设置为1,而是修改为旧值+1呢?这是因为ReentrantLock是可重入锁,同一个线程每持有一次就+1。
      3. 返回false。

    比较非公平锁的tryAcquire实现java.util.concurrent.locks.ReentrantLock.Sync.nonfairTryAcquire(int),公平锁多了一个判断当前节点是否在队列头,这个就保证了是否按照请求锁的顺序来决定获取锁的顺序(同一个线程的多次获取锁除外)。

    现在再回头看公平锁和非公平锁的lock()方法。公平锁只有一句acquire(1);而非公平锁的调用如下:

    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

    很显然,非公平锁在第一次获取锁,或者其它线程释放锁后(可能等待),优先采用compareAndSetState(0,1)然后设置AQS独占线程而持有锁,这样有时候比acquire(1)顺序检查锁持有而要高效。即使在重入锁上,也就是compareAndSetState(0,1)失败,但是是当前线程持有锁上,非公平锁也没有问题。

    addWaiter(mode)

    tryAcquire失败就意味着入队列了。此时AQS的队列中节点Node就开始发挥作用了。一般情况下AQS支持独占锁和共享锁,而独占锁在Node中就意味着条件(Condition)队列为空(上一篇中介绍过相关概念)。在java.util.concurrent.locks.AbstractQueuedSynchronizer.Node中有两个常量,

    static final Node EXCLUSIVE = null; //独占节点模式

    static final Node SHARED = new Node(); //共享节点模式

    addWaiter(mode)中的mode就是节点模式,也就是共享锁还是独占锁模式。

    前面一再强调ReentrantLock是独占锁模式。

    private Node addWaiter(Node mode) {
         Node node = new Node(Thread.currentThread(), mode);
         // Try the fast path of enq; backup to full enq on failure
         Node pred = tail;
         if (pred != null) {
             node.prev = pred;
             if (compareAndSetTail(pred, node)) {
                 pred.next = node;
                 return node;
             }
         }
         enq(node);
         return node;
    }

    上面是节点如队列的一部分。当前仅当队列不为空并且将新节点插入尾部成功后直接返回新节点。否则进入enq(Node)进行操作。

    private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                Node h = new Node(); // Dummy header
                h.next = node;
                node.prev = h;
                if (compareAndSetHead(h)) {
                    tail = node;
                    return h;
                }
            }
            else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

    enq(Node)去队列操作实现了CHL队列的算法,如果为空就创建头结点,然后同时比较节点尾部是否是改变来决定CAS操作是否成功,当且仅当成功后才将为不节点的下一个节点指向为新节点。可以看到这里仍然是CAS操作。

    acquireQueued(node,arg)

    自旋请求锁,如果可能的话挂起线程,直到得到锁,返回当前线程是否中断过(如果park()过并且中断过的话有一个interrupted中断位)。

    final boolean acquireQueued(final Node node, int arg) {
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } catch (RuntimeException ex) {
            cancelAcquire(node);
            throw ex;
        }
    }

    下面的分析就需要用到上节节点的状态描述了。acquireQueued过程是这样的:

    1.  
      1. 如果当前节点是AQS队列的头结点(如果第一个节点是DUMP节点也就是傀儡节点,那么第二个节点实际上就是头结点了),就尝试在此获取锁tryAcquire(arg)。如果成功就将头结点设置为当前节点(不管第一个结点是否是DUMP节点),返回中断位。否则进行2。
      2. 检测当前节点是否应该park(),如果应该park()就挂起当前线程并且返回当前线程中断位。进行操作1。

    一个节点是否该park()是关键,这是由方法java.util.concurrent.locks.AbstractQueuedSynchronizer.shouldParkAfterFailedAcquire(Node, Node)实现的。

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int s = pred.waitStatus;
        if (s < 0) return true;
        if (s > 0) {
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else compareAndSetWaitStatus(pred, 0, Node.SIGNAL);
        return false;
    }

    1.  
      1. 如果前一个节点的等待状态waitStatus<0,也就是前面的节点还没有获得到锁,那么返回true,表示当前节点(线程)就应该park()了。否则进行2。
      2. 如果前一个节点的等待状态waitStatus>0,也就是前一个节点被CANCELLED了,那么就将前一个节点去掉,递归此操作直到所有前一个节点的waitStatus<=0,进行4。否则进行3。
      3. 前一个节点等待状态waitStatus=0,修改前一个节点状态位为SINGAL,表示后面有节点等待你处理,需要根据它的等待状态来决定是否该park()。进行4。
      4. 返回false,表示线程不应该park()。

    selfInterrupt()

    private static void selfInterrupt() {
        Thread.currentThread().interrupt();
    }

    如果线程曾经中断过(或者阻塞过)(比如手动interrupt()或者超时等等,那么就再中断一次,中断两次的意思就是清除中断位)。

    大体上整个Lock.lock()就这样一个流程。除了lock()方法外,还有lockInterruptibly()/tryLock()/unlock()/newCondition()等,在接下来的章节中会一一介绍。

     

     

    本小节介绍锁释放Lock.unlock()。

    Release/TryRelease

    unlock操作实际上就调用了AQS的release操作,释放持有的锁。

    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

    前面提到过tryRelease(arg)操作,此操作里面总是尝试去释放锁,如果成功,说明锁确实被当前线程持有,那么就看AQS队列中的头结点是否为空并且能否被唤醒,如果可以的话就唤醒继任节点(下一个非CANCELLED节点,下面会具体分析)。

    对于独占锁而言,java.util.concurrent.locks.ReentrantLock.Sync.tryRelease(int)展示了如何尝试释放锁(tryRelease)操作。

    protected final boolean tryRelease(int releases) {
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }

    整个tryRelease操作是这样的:

    1.  
      1. 判断持有锁的线程是否是当前线程,如果不是就抛出IllegalMonitorStateExeception(),因为一个线程是不能释放另一个线程持有的锁(否则锁就失去了意义)。否则进行2。
      2. 将AQS状态位减少要释放的次数(对于独占锁而言总是1),如果剩余的状态位0(也就是没有线程持有锁),那么当前线程就是最后一个持有锁的线程,清空AQS持有锁的独占线程。进行3。
      3. 将剩余的状态位写回AQS,如果没有线程持有锁就返回true,否则就是false。

    参考上一节的分析就可以知道,这里c==0决定了是否完全释放了锁。由于ReentrantLock是可重入锁,因此同一个线程可能多重持有锁,那么当且仅当最后一个持有锁的线程释放锁是才能将AQS中持有锁的独占线程清空,这样接下来的操作才需要唤醒下一个需要锁的AQS节点(Node),否则就只是减少锁持有的计数器,并不能改变其他操作。

    tryRelease操作成功后(也就是完全释放了锁),release操作才能检查是否需要唤醒下一个继任节点。这里的前提是AQS队列的头结点需要锁(waitStatus!=0),如果头结点需要锁,就开始检测下一个继任节点是否需要锁操作。

    在上一节中说道acquireQueued操作完成后(拿到了锁),会将当前持有锁的节点设为头结点,所以一旦头结点释放锁,那么就需要寻找头结点的下一个需要锁的继任节点,并唤醒它。

    private void unparkSuccessor(Node node) {
            //此时node是需要是需要释放锁的头结点

            //清空头结点的waitStatus,也就是不再需要锁了
            compareAndSetWaitStatus(node, Node.SIGNAL, 0);

            //从头结点的下一个节点开始寻找继任节点,当且仅当继任节点的waitStatus<=0才是有效继任节点,否则将这些waitStatus>0(也就是CANCELLED的节点)从AQS队列中剔除  
           //这里并没有从head->tail开始寻找,而是从tail->head寻找最后一个有效节点。
           //解释在这里 http://www.blogjava.net/xylz/archive/2010/07/08/325540.html#377512

            Node s = node.next;
            if (s == null || s.waitStatus > 0) {
                s = null;
                for (Node t = tail; t != null && t != node; t = t.prev)
                    if (t.waitStatus <= 0)
                        s = t;
            }

            //如果找到一个有效的继任节点,就唤醒此节点线程
            if (s != null)
                LockSupport.unpark(s.thread);
        }

    这里再一次把acquireQueued的过程找出来。对比unparkSuccessor,一旦头节点的继任节点被唤醒,那么继任节点就会尝试去获取锁(在acquireQueued中node就是有效的继任节点,p就是唤醒它的头结点),如果成功就会将头结点设置为自身,并且将头结点的前任节点清空,这样前任节点(已经过时了)就可以被GC释放了。

    final boolean acquireQueued(final Node node, int arg) {
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } catch (RuntimeException ex) {
            cancelAcquire(node);
            throw ex;
        }
    }

    setHead中,将头结点的前任节点清空并且将头结点的线程清空就是为了更好的GC,防止内存泄露。

    private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    }

    对比lock()操作,unlock()操作还是比较简单的,主要就是释放响应的资源,并且唤醒AQS队列中有效的继任节点。这样所就按照请求的顺序去尝试获取锁了。

    整个lock()/unlock()过程完成了,我们再回头看公平锁(FairSync)和非公平锁(NonfairSync)。

    公平锁和非公平锁只是在获取锁的时候有差别,其它都是一样的。

    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

    在上面非公平锁的代码中总是优先尝试当前是否有线程持有锁,一旦没有任何线程持有锁,那么非公平锁就霸道的尝试将锁“占为己有”。如果在抢占锁的时候失败就和公平锁一样老老实实的去排队。

    也即是说公平锁和非公平锁只是在入AQSCLH队列之前有所差别,一旦进入了队列,所有线程都是按照队列中先来后到的顺序请求锁。

    Condition

    条件变量很大一个程度上是为了解决Object.wait/notify/notifyAll难以使用的问题。

    条件(也称为条件队列 或条件变量)为线程提供了一个含义,以便在某个状态条件现在可能为 true 的另一个线程通知它之前,一直挂起该线程(即让其“等待”)。因为访问此共享状态信息发生在不同的线程中,所以它必须受保护,因此要将某种形式的锁与该条件相关联。等待提供一个条件的主要属性是:以原子方式 释放相关的锁,并挂起当前线程,就像 Object.wait 做的那样。

    上述API说明表明条件变量需要与锁绑定,而且多个Condition需要绑定到同一锁上。前面的Lock中提到,获取一个条件变量的方法是Lock.newCondition()

    void await() throws InterruptedException;
    void awaitUninterruptibly();
    long awaitNanos(long nanosTimeout) throws InterruptedException;
    boolean await(long time, TimeUnit unit) throws InterruptedException;
    boolean awaitUntil(Date deadline) throws InterruptedException;
    void signal();
    void signalAll();

    以上是Condition接口定义的方法,await*对应于Object.waitsignal对应于Object.notifysignalAll对应于Object.notifyAll。特别说明的是Condition的接口改变名称就是为了避免与Object中的wait/notify/notifyAll的语义和使用上混淆,因为Condition同样有wait/notify/notifyAll方法。

    每一个Lock可以有任意数据的Condition对象,Condition是与Lock绑定的,所以就有Lock的公平性特性:如果是公平锁,线程为按照FIFO的顺序从Condition.await中释放,如果是非公平锁,那么后续的锁竞争就不保证FIFO顺序了。

    一个使用Condition实现生产者消费者的模型例子如下。

    package xylz.study.concurrency.lock;

    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;

    public class ProductQueue<T> {

        private final T[] items;

        private final Lock lock = new ReentrantLock();

        private Condition notFull = lock.newCondition();

        private Condition notEmpty = lock.newCondition();

        //
        private int head, tail, count;

        public ProductQueue(int maxSize) {
            items = (T[]) new Object[maxSize];
        }

        public ProductQueue() {
            this(10);
        }

        public void put(T t) throws InterruptedException {
            lock.lock();
            try {
                while (count == getCapacity()) {
                    notFull.await();
                }
                items[tail] = t;
                if (++tail == getCapacity()) {
                    tail = 0;
                }
                ++count;
                notEmpty.signalAll();
            } finally {
                lock.unlock();
            }
        }

        public T take() throws InterruptedException {
            lock.lock();
            try {
                while (count == 0) {
                    notEmpty.await();
                }
                T ret = items[head];
                items[head] = null;//GC
                //
                if (++head == getCapacity()) {
                    head = 0;
                }
                --count;
                notFull.signalAll();
                return ret;
            } finally {
                lock.unlock();
            }
        }

        public int getCapacity() {
            return items.length;
        }

        public int size() {
            lock.lock();
            try {
                return count;
            } finally {
                lock.unlock();
            }
        }

    }

    在这个例子中消费take()需要 队列不为空,如果为空就挂起(await()),直到收到notEmpty的信号;生产put()需要队列不满,如果满了就挂起(await()),直到收到notFull的信号。

    可能有人会问题,如果一个线程lock()对象后被挂起还没有unlock,那么另外一个线程就拿不到锁了(lock()操作会挂起),那么就无法通知(notify)前一个线程,这样岂不是“死锁”了?

     

    await* 操作

    上一节中说过多次ReentrantLock是独占锁,一个线程拿到锁后如果不释放,那么另外一个线程肯定是拿不到锁,所以在lock.lock()lock.unlock()之间可能有一次释放锁的操作(同样也必然还有一次获取锁的操作)。我们再回头看代码,不管take()还是put(),在进入lock.lock()后唯一可能释放锁的操作就是await()了。也就是说await()操作实际上就是释放锁,然后挂起线程,一旦条件满足就被唤醒,再次获取锁!

    public final void await() throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        Node node = addConditionWaiter();
        int savedState = fullyRelease(node);
        int interruptMode = 0;
        while (!isOnSyncQueue(node)) {
            LockSupport.park(this);
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        if (node.nextWaiter != null)
            unlinkCancelledWaiters();
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }

    上面是await()的代码片段。上一节中说过,AQS在获取锁的时候需要有一个CHL的FIFO队列,所以对于一个Condition.await()而言,如果释放了锁,要想再一次获取锁那么就需要进入队列,等待被通知获取锁。完整的await()操作是安装如下步骤进行的:

    1.  
      1. 将当前线程加入Condition锁队列。特别说明的是,这里不同于AQS的队列,这里进入的是Condition的FIFO队列。后面会具体谈到此结构。进行2。
      2. 释放锁。这里可以看到将锁释放了,否则别的线程就无法拿到锁而发生死锁。进行3。
      3. 自旋(while)挂起,直到被唤醒或者超时或者CACELLED等。进行4。
      4. 获取锁(acquireQueued)。并将自己从Condition的FIFO队列中释放,表明自己不再需要锁(我已经拿到锁了)。

    这里再回头介绍Condition的数据结构。我们知道一个Condition可以在多个地方被await*(),那么就需要一个FIFO的结构将这些Condition串联起来,然后根据需要唤醒一个或者多个(通常是所有)。所以在Condition内部就需要一个FIFO的队列。

    private transient Node firstWaiter;
    private transient Node lastWaiter;

    上面的两个节点就是描述一个FIFO的队列。我们再结合前面提到的节点(Node)数据结构。我们就发现Node.nextWaiter就派上用场了!nextWaiter就是将一系列的Condition.await*串联起来组成一个FIFO的队列。

     

    signal/signalAll 操作

    await*()清楚了,现在再来看signal/signalAll就容易多了。按照signal/signalAll的需求,就是要将Condition.await*()中FIFO队列中第一个Node唤醒(或者全部Node)唤醒。尽管所有Node可能都被唤醒,但是要知道的是仍然只有一个线程能够拿到锁,其它没有拿到锁的线程仍然需要自旋等待,就上上面提到的第4步(acquireQueued)。

    private void doSignal(Node first) {
        do {
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            first.nextWaiter = null;
        } while (!transferForSignal(first) &&
                 (first = firstWaiter) != null);
    }

    private void doSignalAll(Node first) {
        lastWaiter = firstWaiter  = null;
        do {
            Node next = first.nextWaiter;
            first.nextWaiter = null;
            transferForSignal(first);
            first = next;
        } while (first != null);
    }

    上面的代码很容易看出来,signal就是唤醒Condition队列中的第一个非CANCELLED节点线程,而signalAll就是唤醒所有非CANCELLED节点线程。当然了遇到CANCELLED线程就需要将其从FIFO队列中剔除。

    final boolean transferForSignal(Node node) {
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;

        Node p = enq(node);
        int c = p.waitStatus;
        if (c > 0 || !compareAndSetWaitStatus(p, c, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

    上面就是唤醒一个await*()线程的过程,根据前面的小节介绍的,如果要unpark线程,并使线程拿到锁,那么就需要线程节点进入AQS的队列。所以可以看到在LockSupport.unpark之前调用了enq(node)操作,将当前节点加入到AQS队列。

    整个锁机制的原理就介绍完了,从下一节开始就进入了锁机制的应用了。

    分享到:
    评论

    相关推荐

      SQL server锁的机制

      SQL Server的锁机制是数据库管理系统中用于控制并发访问的关键组件,确保数据的完整性和一致性。在SQL Server中,锁用于管理事务之间的数据访问,防止数据的不一致性,并提高多用户的并发性。以下是对SQL Server锁...

      面向Java锁机制的字节码自动重构框架.pdf

      根据提供的文件内容,以下是关于“面向Java锁机制的字节码自动重构框架”的详细知识点解析: 1. Java锁机制 Java语言提供了多种锁机制,包括同步锁(synchronized),可重入锁(ReentrantLock)和读写锁(ReadWriteLock)...

      MS-SQL 锁机制

      ### MS-SQL 锁机制详解 #### 一、锁的概述 锁是在多用户数据库系统中用于实现并发控制的关键机制之一。它可以帮助防止多个用户同时访问相同的数据时导致的数据不一致性问题。MS-SQL Server通过不同的锁机制来确保...

      DB数据库锁机制及问题PPT课件.pptx

      ### DB数据库锁机制详解 #### 一、锁的基本概念与分类 锁是数据库管理系统(DBMS)为了控制并发访问数据的完整性和一致性而采用的一种机制。它确保了在多用户环境中,多个事务能够按照一定的顺序执行,避免数据冲突...

      WinForm(C#,.net3.5)三维图形旋转练习(自己做的)

      欧拉角是指三个旋转角度,而四元数则是一种更高级的数学工具,用于避免万向锁问题(即某些情况下连续旋转会导致图形锁定)。在没有扎实的数学基础的情况下,实现这些旋转可能确实会带来挑战。 在项目中,"Windows...

      对于Oracle锁的一些理论总结

      Oracle数据库中的锁机制是确保数据一致性、避免并发访问冲突的关键技术。本文主要探讨Oracle的锁理论,特别是DML锁,即数据操纵语言锁。 首先,锁的基本概念是用于控制共享资源的并发访问。在Oracle中,锁分为三类...

      深入解析MS-SQL锁机制

      MS-SQL锁机制是数据库管理系统中用于控制并发访问和维护数据一致性的重要机制。在多用户环境中,为了防止并发操作导致的数据不一致问题,如丢失更新、脏读和不可重复读等,引入了锁的概念。锁可以限制不同事务对同一...

      安卓转动解锁

      这种解锁机制不仅增加了操作的乐趣,同时也提升了用户体验。在安卓系统中,转动解锁通常涉及到图形绘制和SurfaceView的运用,这两个知识点在Android应用开发中至关重要。 一、图形绘制 在安卓转动解锁中,图形绘制...

      DB2和 Oracle的并发控制(锁)比较

      为了实现这些目标,DBMS引入了并发控制机制,其中最常用的就是锁机制。 #### 二、DB2的多粒度锁机制 **2.1 锁的对象** DB2支持多种粒度的锁机制,包括对表空间、表、行以及索引的锁定。在实际应用中,表锁和行锁...

      电子功用-带有电机锁止器的助力转向装置

      四、电机锁止器的工作机制 电机锁止器通常在两种情况下发挥作用:一是车辆静止时,为了防止电动机在无驾驶意图下产生助力,锁止器会切断电动机的电源;二是高速行驶时,为了确保车辆的直线稳定性,锁止器会限制电动...

      数电-密码电子锁-Multisim(密码电子锁的设计与调试课程设计)

      2. **了解电子门铃的基本知识:**在掌握密码电子锁设计的同时,了解电子门铃的工作机制,并将其作为密码电子锁的一项附加功能。 3. **进一步熟悉D触发器的基本功能:**通过对D触发器的操作实践,巩固学生对于该元件...

      三维交互操作(缩放,旋转,平移)

      为了实现漫游功能,需要设计一套完整的输入处理机制,并结合矩阵变换来更新视点位置。 在编程实现这些功能时,还需要考虑性能优化,如使用矩阵堆栈进行变换累积,以及利用硬件加速特性。此外,为了提供流畅的用户...

      《CMOS集成电路闩锁效应》第三章课件.pptx

      《CMOS集成电路闩锁效应》第三章主要探讨了CMOS集成电路中常见的问题——闩锁效应。闩锁效应是指在特定条件下,电路中的寄生PNPN结构由于正反馈机制导致电路进入不稳定状态,甚至可能导致器件损坏。以下是本章内容的...

      转盘式自动锁螺丝机_零件图_机械工程图_机械三维3D设计图打包下载.rar

      在实际运作中,转盘式自动锁螺丝机的工作原理是基于转盘连续旋转和分度定位的机制。转盘上设有多个工位,每个工位对应一个螺丝锁付的步骤。螺丝由输送系统精密送至特定位置后,再由锁付机构准确锁紧。整个过程由控制...

      InformInformix数据库的锁技术的锁技术

      Informix 的锁机制为并发控制提供了灵活性,允许开发者根据应用程序的需求选择最适合的锁定策略。通过调整锁的类型和范围,可以平衡数据的完整性和系统的并发性能。然而,正确地管理和使用锁是至关重要的,以避免...

      android和linux的唤醒机制

      #### 三、Android层源码解析 ##### 1. HAL层分析 HAL层是Android中与硬件直接交互的一层,它的实现通常位于`hardware/libhardware_legacy/power/power.c`文件中。该文件中的代码主要负责管理和控制硬件级别的唤醒...

      密码锁(VHDL)EDA技术

      1. 锁的状态管理:锁可以处于开锁、上锁和错误状态,这些状态可以通过一组布尔变量表示,并在进程中根据条件进行转换。 2. 输入检测:VHDL提供了多种信号操作符,如‘=’(等于)、‘/=’(不等于)等,用于检测密码...

      Android 图案解锁之九宫解锁源码.zip

      - **路径编码**:将用户绘制的图案转换为一串数字序列,比如连接(1,2,4)三个点,编码为"124"。 - **存储路径**:将用户的解锁图案编码后存储到设备的安全存储区域,如KeyStore或SharedPreferences。 - **解锁验证...

      电信设备-一种靶牌转动及锁紧装置.zip

      本文将从靶牌转动机制、锁紧装置的设计与功能、以及装置的附加防护特性三个方面,探讨这项创新装置的结构和技术细节。 首先,靶牌转动机制是这一装置的核心所在。合理的转动设计使得靶牌能够在特定范围内自由旋转,...

      oracle锁讲解笔记

      ### Oracle锁机制详解 #### 锁的基本概念与作用 锁机制是Oracle数据库中用于管理并发访问共享资源的关键组件。在多用户环境中,确保数据的一致性和完整性至关重要,尤其是在多个会话试图同时修改相同数据的情况下...

    Global site tag (gtag.js) - Google Analytics