Java并发编程-再谈 AbstractQueuedSynchronizer 1 ：独占模式

关于AbstractQueuedSynchronizer

JDK1.5以后引入了并发包java.util.concurrent，大大提升了Java程序的并发性能。关于java.util.concurrent包我总结以下：

• AbstractQueuedSynchronizer是并发类诸如ReentrantLock、CountDownLatch、Semphore的核心
• CAS算法是AbstractQueuedSynchronizer的核心

能够说AbstractQueuedSynchronizer是并发类的重中之重。其实以前在ReentrantLock实现原理深刻探究一文中已经有结合ReentrantLock详细解读过AbstractQueuedSynchronizer，但限于当时水平缘由，回看一年半前的此文，感受对于AbstractQueuedSynchronizer的解读理解还不够深，所以这里更新一篇文章，再次解读AbstractQueuedSynchronizer的数据结构即相关源码实现，本文基于JDK1.7版本。 

AbstactQueuedSynchronizer的基本数据结构

AbstractQueuedSynchronizer的基本数据结构为Node，关于Node，JDK做者写了详细的注释，这里我大体总结几点：

1. AbstractQueuedSynchronizer的等待队列是CLH队列的变种，CLH队列一般用于自旋锁，AbstractQueuedSynchronizer的等待队列用于阻塞同步器
2. 每一个节点中持有一个名为”status”的字段用因而否一条线程应当阻塞的追踪，可是status字段并不保证加锁
3. 一条线程若是它处于队列的头，那么他会尝试去acquire，可是成为头并不保证成功，它只是有权利去竞争
4. 要进入队列，你只须要自动将它拼接在队列尾部便可；要从队列中移除，你只须要设置header字段

下面我用一张表格总结一下Node中持有哪些变量且每一个变量的含义：

关于SIGNAL、CANCELLED、CONDITION、PROPAGATE四个状态，JDK源码的注释中一样有了详细的解读，再用一张表格总结一下：

 

AbstractQueuedSynchronizer供子类实现的方法

AbstractQueuedSynchzonizer是基于模板模式的实现，不过它的模板模式写法有点特别，整个类中没有任何一个abstract的抽象方法，取而代之的是，须要子类去实现的那些方法经过一个方法体抛出UnsupportedOperationException异常来让子类知道。

AbstractQueuedSynchronizer类中一共有五处方法供子类实现，用表格总结一下：

这里的acquire很差翻译，因此就直接原词放上来了，由于acquire是一个动词，后面并无带宾语，所以不知道具体acquire的是什么。按照我我的理解，acquire的意思应当是根据状态字段state去获取一个执行当前动做的资格。

好比ReentrantLock的lock()方法最终会调用acquire方法，那么：

1. 线程1去lock()，执行acquire，发现state=0，所以有资格执行lock()的动做，将state设置为1，返回true
2. 线程2去lock()，执行acquire，发现state=1，所以没有资格执行lock()的动做，返回false

这种理解我认为应当是比较准确的。

独占模式acquire实现流程

有了上面的这些基础，咱们看一下独占式acquire的实现流程，主要是在线程acquire失败后，是如何构建数据结构的，先看理论，以后再用一个例子画图说明。

看一下AbstractQuueuedSynchronizer的acquire方法实现流程，acquire方法是用于独占模式下进行操做的：

1 2 3 4 5

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

tryAcquire方法前面说过了，是子类实现的一个方法，若是tryAcquire返回的是true（成功），即代表当前线程得到了一个执行当前动做的资格，天然也就不须要构建数据结构进行阻塞等待。

若是tryAcquire方法返回的是false，那么当前线程没有得到执行当前动做的资格，接着执行”acquireQueued(addWaiter(Node.EXCLUSIVE), arg))”这句代码，这句话很明显，它是由两步构成的：

1. addWaiter，添加一个等待者
2. acquireQueued，尝试从等待队列中去获取执行一次acquire动做

分别看一下每一步作了什么。

addWaiter

先看第一步，addWaiter作了什么，从传入的参数Node.EXCLUSIVE咱们知道这是独占模式的：

1 2 3 4 5 6 7 8 9 10 11 12 13 14

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 prev = tail;      if (prev != null ) {          node.prev = prev;          if (compareAndSetTail(prev, node)) {              prev.next = node;              return node;          }      }      enq(node);      return node; }

首先看第4行~第11行的代码，得到当前数据结构中的尾节点，若是有尾节点，那么先获取这个节点认为它是前驱节点prev，而后：

• 新生成的Node的前驱节点指向prev
• 并发下只有一条线程能够经过CAS算法让本身的Node成为尾节点，此时将此prev的next指向该线程对应的Node

所以在数据结构中有节点的状况下，全部新增节点都是做为尾节点插入数据结构。从注释上来看，这段逻辑的存在的意义是以最短路径O(1)的效果完成快速入队，以最大化减少开销。

假如当前节点没有被设置为尾节点，那么执行enq方法：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

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

这段代码的逻辑为：

1. 若是尾节点为空，即当前数据结构中没有节点，那么new一个不带任何状态的Node做为头节点
2. 若是尾节点不为空，那么并发下使用CAS算法将当前Node追加成为尾节点，因为是一个for(;;)循环，所以全部没有成功acquire的Node最终都会被追加到数据结构中

看完了代码，用一张图表示一下AbstractQueuedSynchronizer的总体数据结构（比较简单，就不本身画了，网上随便找了一张图）：

 

acquireQueued

队列构建好了，下一步就是在必要的时候从队列里面拿出一个Node了，这就是acquireQueued方法，顾名思义，从队列里面acquire。看下acquireQueued方法的实现：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

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

这段代码描述了几件事：

1. 从第6行的代码获取节点的前驱节点p，第7行的代码判断p是前驱节点并tryAcquire咱们知道，只有当前第一个持有Thread的节点才会尝试acquire，若是节点acquire成功，那么setHead方法，将当前节点做为head、将当前节点中的thread设置为null、将当前节点的prev设置为null，这保证了数据结构中头结点永远是一个不带Thread的空节点
2. 若是当前节点不是前驱节点或者tryAcquire失败，那么执行第13行~第15行的代码，作了两步操做，首先判断在acquie失败后是否应该park，其次park并检查中断状态

看一下第一步shouldParkAfterFailedAcquire代码作了什么：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27

private static boolean shouldParkAfterFailedAcquire(Node prev, Node node) {      int ws = prev.waitStatus;      if (ws == Node.SIGNAL)          /*           * This node has already set status asking a release           * to signal it, so it can safely park.           */          return true;      if (ws > 0) {          /*           * prevecessor was cancelled. Skip over prevecessors and           * indicate retry.           */          do {              node.prev = prev = prev.prev;          } while (prev.waitStatus > 0);          prev.next = node;      } else {          /*           * waitStatus must be 0 or PROPAGATE.  Indicate that we           * need a signal, but don't park yet.  Caller will need to           * retry to make sure it cannot acquire before parking.           */          compareAndSetWaitStatus(prev, ws, Node.SIGNAL);      }      return false ; }

这里每一个节点判断它前驱节点的状态，若是：

1. 它的前驱节点是SIGNAL状态的，返回true，表示当前节点应当park
2. 它的前驱节点的waitStatus>0，至关于CANCELLED（由于状态值里面只有CANCELLED是大于0的），那么CANCELLED的节点做废，当前节点不断向前找并从新链接为双向队列，直到找到一个前驱节点waitStats不是CANCELLED的为止
3. 它的前驱节点不是SIGNAL状态且waitStatus<=0，此时执行第24行代码，利用CAS机制，若是waitStatus的前驱节点是0那么更新为SIGNAL状态

若是判断判断应当park，那么parkAndCheckInterrupt方法：

1 2 3 4

private final boolean parkAndCheckInterrupt() {        LockSupport.park( this );        return Thread.interrupted();   }

利用LockSupport的park方法让当前线程阻塞。 

独占模式release流程

上面整理了独占模式的acquire流程，看到了等待的Node是如何构建成一个数据结构的，下面看一下释放的时候作了什么，release方法的实现为：

1 2 3 4 5 6 7 8 9

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一样是子类去实现的，表示当前动做我执行完了，要释放我执行当前动做的资格，讲这个资格让给其它线程，而后tryRelease释放成功，获取到head节点，若是head节点的waitStatus不为0的话，执行unparkSuccessor方法，顾名思义unparkSuccessor意为unpark头结点的继承者，方法实现为：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

private void unparkSuccessor(Node node) {          /*           * If status is negative (i.e., possibly needing signal) try           * to clear in anticipation of signalling.  It is OK if this           * fails or if status is changed by waiting thread.           */          int ws = node.waitStatus;          if (ws < 0)              compareAndSetWaitStatus(node, ws, 0);            /*           * Thread to unpark is held in successor, which is normally           * just the next node.  But if cancelled or apparently null,           * traverse backwards from tail to find the actual           * non-cancelled successor.           */          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); }

这段代码比较好理解，整理一下流程：

1. 头节点的waitStatus<0，将头节点的waitStatus设置为0
2. 拿到头节点的下一个节点s，若是s==null或者s的waitStatus>0（被取消了），那么从队列尾巴开始向前寻找一个waitStatus<=0的节点做为后继要唤醒的节点

最后，若是拿到了一个不等于null的节点s，就利用LockSupport的unpark方法让它取消阻塞。

实战举例：数据结构构建

上面的例子讲解地过于理论，下面利用ReentrantLock举个例子，可是这里不讲ReentrantLock实现原理，只是利用ReentrantLock研究AbstractQueuedSynchronizer的acquire和release。示例代码为：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49

/**   * @author 五月的仓颉http://www.cnblogs.com/xrq730/p/7056614.html   */ public class AbstractQueuedSynchronizerTest {        @Test      public void testAbstractQueuedSynchronizer() {          Lock lock = new ReentrantLock();            Runnable runnable0 = new ReentrantLockThread(lock);          Thread thread0 = new Thread(runnable0);          thread0.setName( "线程0" );            Runnable runnable1 = new ReentrantLockThread(lock);          Thread thread1 = new Thread(runnable1);          thread1.setName( "线程1" );            Runnable runnable2 = new ReentrantLockThread(lock);          Thread thread2 = new Thread(runnable2);          thread2.setName( "线程2" );            thread0.start();          thread1.start();          thread2.start();            for (;;);      }        private class ReentrantLockThread implements Runnable {            private Lock lock;            public ReentrantLockThread(Lock lock) {              this .lock = lock;          }            @Override          public void run() {              try {                  lock.lock();                  for (;;);              } finally {                  lock.unlock();              }          }        }   }

所有是死循环，至关于第一条线程（线程0）acquire成功以后，后两条线程（线程一、线程2）阻塞，下面的代码就不考虑后两条线程谁先谁后的问题，就一条线程（线程1）流程执行到底、另外一条线程（线程2）流程执行到底这么分析了。

这里再把addWaiter和enq两个方法源码贴一下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14

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 prev = tail;      if (prev != null ) {          node.prev = prev;          if (compareAndSetTail(prev, node)) {              prev.next = node;              return node;          }      }      enq(node);      return node; }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

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

首先第一个acquire失败的线程1，因为此时整个数据结构中么没有任何数据，所以addWaiter方法第4行中拿到的prev=tail为空，执行enq方法，首先第3行获取tail，第4行判断到tail是null，所以头结点new一个Node出来经过CAS算法设置为数据结构的head，tail一样也是这个Node，此时数据结构为：

为了方便描述，prev和next，我给每一个Node随便加了一个地址。接着继续enq，由于enq内是一个死循环，因此继续第3行获取tail，new了一个空的Node以后tail就有了，执行else判断，经过第8行~第10行代码将当前线程对应的Node追加到数据结构尾部，那么当前构建的数据结构为：

这样，线程1对应的Node被加入数据结构，成为数据结构的tail，而数据结构的head是一个什么都没有的空Node。

接着线程2也acquire失败了，线程2既然acquire失败，那也要准备被加入数据结构中，继续先执行addWaiter方法，因为此时已经有了tail，所以不须要执行enq方法，能够直接将当前Node添加到数据结构尾部，那么当前构建的数据结构为：

至此，两个阻塞的线程构建的三个Node已经所有归位。

实战举例：线程阻塞

上述流程只是描述了构建数据结构的过程，并无描述线程一、线程2阻塞的流程，所以接着继续用实际例子看一下线程一、线程2如何阻塞。贴一下acquireQueued、shouldParkAfterFailedAcquire两个方法源码：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

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

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27

private static boolean shouldParkAfterFailedAcquire(Node prev, Node node) {      int ws = prev.waitStatus;      if (ws == Node.SIGNAL)          /*           * This node has already set status asking a release           * to signal it, so it can safely park.           */          return true;      if (ws > 0) {          /*           * prevecessor was cancelled. Skip over prevecessors and           * indicate retry.           */          do {              node.prev = prev = prev.prev;          } while (prev.waitStatus > 0);          prev.next = node;      } else {          /*           * waitStatus must be 0 or PROPAGATE.  Indicate that we           * need a signal, but don't park yet.  Caller will need to           * retry to make sure it cannot acquire before parking.           */          compareAndSetWaitStatus(prev, ws, Node.SIGNAL);      }      return false ; }

首先是线程1，它的前驱节点是head节点，在它tryAcquire成功的状况下，执行第8行~第11行的代码。作几件事情：

1. head为线程1对应的Node
2. 线程1对应的Node的thread置空
3. 线程1对应的Node的prev置空
4. 原head的next置空，这样原head中的prev、next、thread都为空，对象内没有引用指向其余地方，GC能够认为这个Node是垃圾，对这个Node进行回收，注释”Help GC”就是这个意思
5. failed=false表示没有失败

所以，若是线程1执行tryAcquire成功，那么数据结构将变为：

从上述流程能够总结到：只有前驱节点为head的节点会尝试tryAcquire，其他都不会，结合后面的release选继承者的方式，保证了先acquire失败的线程会优先从阻塞状态中解除去从新acquire。这是一种公平的acquire方式，由于它遵循”先到先得”原则，可是咱们能够动动手脚让这种公平变为非公平，好比ReentrantLock默认的非公平模式，这个留在后面说。

那若是线程1执行tryAcquire失败，那么要执行shouldParkAfterFailedAcquire方法了，shouldParkAfterFailedAcquire拿线程1的前驱节点也就是head节点的waitStatus作了一个判断，由于waitStatus=0，所以执行第18行~第20行的逻辑，将head的waitStatus设置为SIGNAL即-1，而后方法返回false，数据结构变为：

看到这里就一个变化：head的waitStatus从0变成了-1。既然shouldParkAfterFailedAcquire返回false，acquireQueued的第13行~第14行的判断天然不经过，继续走for(;;)循环，若是tryAcquire失败显然又来到了shouldParkAfterFailedAcquire方法，此时线程1对应的Node的前驱节点head节点的waitStatus已经变为了SIGNAL即-1，所以执行第4行~第8行的代码，直接返回true出去。

shouldParkAfterFailedAcquire返回true，parkAndCheckInterrupt直接调用LockSupport的park方法：

1 2 3 4

private final boolean parkAndCheckInterrupt() {       LockSupport.park( this );       return Thread.interrupted();   }

至此线程1阻塞，线程2阻塞的流程与线程1阻塞的流程相同，能够本身分析一下。

另外再提一个问题，不知道你们会不会想：

1. 为何线程1对应的Node构建完毕不直接调用LockSupport的park方法进行阻塞？
2. 为何不直接把head的waitStatus直接设置为Signal而要从0设置为Signal？

我认为这是AbstractQueuedSynchronizer开发人员作了相似自旋的操做。由于不少时候获取acquire进行操做的时间很短，阻塞会引发上下文的切换，而很短期就从阻塞状态解除，这样相对会比较耗费性能。

所以咱们看到线程1自构建完毕Node加入数据结构到阻塞，一共尝试了两次tryAcquire，若是其中有一次成功，那么线程1就没有必要被阻塞，提高了性能。