深入浅出 Java Concurrency (3): 锁机制原理

前面的章节主要谈谈原子操作，至于与原子操作一些相关的问题或者说陷阱就放到最后的总结篇来整体说明。从这一章开始花少量的篇幅谈谈锁机制。

上一个章节中谈到了锁机制，并且针对于原子操作谈了一些相关的概念和设计思想。接下来的文章中，尽可能的深入研究锁机制，并且理解里面的原理和实际应用场合。

尽管synchronized在语法上已经足够简单了，在JDK 5之前只能借助此实现，但是由于是独占锁，性能却不高，因此JDK 5以后就开始借助于JNI来完成更高级的锁实现。

JDK 5中的锁是接口java.util.concurrent.locks.Lock。另外java.util.concurrent.locks.ReadWriteLock提供了一对可供读写并发的锁。根据前面的规则，我们从java.util.concurrent.locks.Lock的API开始。

?

void lock();

获取锁。

如果锁不可用，出于线程调度目的，将禁用当前线程，并且在获得锁之前，该线程将一直处于休眠状态。

void lockInterruptibly() throws InterruptedException;

如果当前线程未被中断，则获取锁。

如果锁可用，则获取锁，并立即返回。

如果锁不可用，出于线程调度目的，将禁用当前线程，并且在发生以下两种情况之一以前，该线程将一直处于休眠状态：

锁由当前线程获得；或者其他某个线程中断当前线程，并且支持对锁获取的中断。

如果当前线程：

在进入此方法时已经设置了该线程的中断状态；或者在获取锁时被中断，并且支持对锁获取的中断，

则将抛出???? //
???? System.out.println();
???? costTime = 0;
???? //
???? final Object lock = new Object();
???? for (int m = 0; m < max; m++) {
???????? staticValue=0;
???????? long start1 = System.nanoTime();
???????? Thread[] ts = new Thread[max];
???????? for(int i=0;i<max;i++) {
???????????? ts[i] = new Thread() {
???????????????? public void run() {
???????????????????? for (int i = 0; i < loopCount; i++) {
???????????????????????? synchronized(lock) {
???????????????????????????? ++staticValue;
???????????????????????? }
???????????????????? }
???????????????? }
???????????? };
???????? }
???????? for(Thread t:ts) {
???????????? t.start();
???????? }
???????? for(Thread t:ts) {
???????????? t.join();
???????? }
???????? long end1 = System.nanoTime();
???????? costTime += (end1-start1);
???? }
???? //
???? System.out.println("cost2: " + (costTime));
}


static int staticValue = 0;

?

在这个例子中每次启动10个线程，每个线程计算100000次自增操作，重复测试10次，下面是某此测试的结果：

cost1: 624071136

cost2: 2057847833

尽管上面的例子不是非常正式的测试案例，但上面的例子在于说明，Lock的性能比synchronized的要好得多。如果可以的话总是使用Lock替代synchronized是一个明智的选择。

?

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

?

AQS

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

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



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

完整的设计原理可以参考Doug Lea的论文AQS采用的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个步骤。

?
如果tryAcquire(arg)成功，那就没有问题，已经拿到锁，整个lock()过程就结束了。如果失败进行操作2。创建一个独占节点（Node）并且此节点加入CHL队列末尾。进行操作3。自旋尝试获取锁，失败根据前一个节点来决定是否挂起（park()），直到成功获取到锁。进行操作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是否为空或者当前线程是否在队列头（为了区分公平与非公平锁）。

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

?
如果当前节点是AQS队列的头结点（如果第一个节点是DUMP节点也就是傀儡节点，那么第二个节点实际上就是头结点了），就尝试在此获取锁tryAcquire(arg)。如果成功就将头结点设置为当前节点（不管第一个结点是否是DUMP节点），返回中断位。否则进行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;
}

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

?
判断持有锁的线程是否是当前线程，如果不是就抛出IllegalMonitorStateExeception()，因为一个线程是不能释放另一个线程持有的锁（否则锁就失去了意义）。否则进行2。将AQS状态位减少要释放的次数（对于独占锁而言总是1），如果剩余的状态位0（也就是没有线程持有锁），那么当前线程就是最后一个持有锁的线程，清空AQS持有锁的独占线程。进行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);
}

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

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

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.wait，signal对应于Object.notify，signalAll对应于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()操作是安装如下步骤进行的：

?
将当前线程加入Condition锁队列。特别说明的是，这里不同于AQS的队列，这里进入的是Condition的FIFO队列。后面会具体谈到此结构。进行2。释放锁。这里可以看到将锁释放了，否则别的线程就无法拿到锁而发生死锁。进行3。自旋(while)挂起，直到被唤醒或者超时或者CACELLED等。进行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队列。

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

转 http://www.blogjava.net/xylz/archive/2010/07/08/325540.html

?

?