Java并发系列之AbstractQueuedSynchronizer源码分析（条件队列）

通过前面三篇的分析，我们深入了解了abstractqueuedsynchronizer的内部结构和一些设计理念，知道了abstractqueuedsynchronizer内部维护了一个同步状态和两个排队区，这两个排队区分别是同步队列和条件队列。我们还是拿公共厕所做比喻，同步队列是主要的排队区，如果公共厕所没开放，所有想要进入厕所的人都得在这里排队。而条件队列主要是为条件等待设置的，我们想象一下如果一个人通过排队终于成功获取锁进入了厕所，但在方便之前发现自己没带手纸，碰到这种情况虽然很无奈，但是它也必须接受这个事实，这时它只好乖乖的出去先准备好手纸(进入条件队列等待)，当然在出去之前还得把锁给释放了好让其他人能够进来，在准备好了手纸(条件满足)之后它又得重新回到同步队列中去排队。当然进入房间的人并不都是因为没带手纸，可能还有其他一些原因必须中断操作先去条件队列中去排队，所以条件队列可以有多个，依不同的等待条件而设置不同的条件队列。条件队列是一条单向链表，condition接口定义了条件队列中的所有操作，abstractqueuedsynchronizer内部的conditionobject类实现了condition接口，下面我们看看condition接口都定义了哪些操作。

public interface condition {
  
  //响应线程中断的条件等待
  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开头的是线程进入条件队列等待的方法，以signal开头的是将条件队列中的线程“唤醒”的方法。这里要注意的是，调用signal方法可能唤醒线程也可能不会唤醒线程，什么时候会唤醒线程这得看情况，后面会讲到，但是调用signal方法一定会将线程从条件队列中移到同步队列尾部。这里为了叙述方便，我们先暂时不纠结这么多，统一称signal方法为唤醒条件队列线程的操作。大家注意看一下，await方法分为5种，分别是响应线程中断等待，不响应线程中断等待，设置相对时间不自旋等待，设置相对时间自旋等待，设置绝对时间等待；signal方法只有2种，分别是只唤醒条件队列头结点和唤醒条件队列所有结点的操作。同一类的方法基本上是相通的，由于篇幅所限，我们不可能也不需要将这些方法全部仔细的讲到，只需要将一个代表方法搞懂了再看其他方法就能够触类旁通。所以在本文中我只会细讲await方法和signal方法，其他方法不细讲但会贴出源码来以供大家参考。

1. 响应线程中断的条件等待

//响应线程中断的条件等待
public final void await() throws interruptedexception {
  //如果线程被中断则抛出异常
  if (thread.interrupted()) {
    throw new interruptedexception();
  }
  //将当前线程添加到条件队列尾部
  node node = addconditionwaiter();
  //在进入条件等待之前先完全释放锁
  int savedstate = fullyrelease(node);
  int interruptmode = 0;
  //线程一直在while循环里进行条件等待
  while (!isonsyncqueue(node)) {
    //进行条件等待的线程都在这里被挂起, 线程被唤醒的情况有以下几种：
    //1.同步队列的前继结点已取消
    //2.设置同步队列的前继结点的状态为signal失败
    //3.前继结点释放锁后唤醒当前结点
    locksupport.park(this);
    //当前线程醒来后立马检查是否被中断, 如果是则代表结点取消条件等待, 此时需要将结点移出条件队列
    if ((interruptmode = checkinterruptwhilewaiting(node)) != 0) {
      break;
    }
  }
  //线程醒来后就会以独占模式获取锁
  if (acquirequeued(node, savedstate) && interruptmode != throw_ie) {
    interruptmode = reinterrupt;
  }
  //这步操作主要为防止线程在signal之前中断而导致没与条件队列断绝联系
  if (node.nextwaiter != null) {
    unlinkcancelledwaiters();
  }
  //根据中断模式进行响应的中断处理
  if (interruptmode != 0) {
    reportinterruptafterwait(interruptmode);
  }
}

当线程调用await方法的时候，首先会将当前线程包装成node结点放入条件队列尾部。在addconditionwaiter方法中，如果发现条件队列尾结点已取消就会调用unlinkcancelledwaiters方法将条件队列所有的已取消结点清空。这步操作是插入结点的准备工作，那么确保了尾结点的状态也是condition之后，就会新建一个node结点将当前线程包装起来然后放入条件队列尾部。注意，这个过程只是将结点添加到同步队列尾部而没有挂起线程哦。

第二步：完全将锁释放

//完全释放锁
final int fullyrelease(node node) {
  boolean failed = true;
  try {
    //获取当前的同步状态
    int savedstate = getstate();
    //使用当前的同步状态去释放锁
    if (release(savedstate)) {
      failed = false;
      //如果释放锁成功就返回当前同步状态
      return savedstate;
    } else {
      //如果释放锁失败就抛出运行时异常
      throw new illegalmonitorstateexception();
    }
  } finally {
    //保证没有成功释放锁就将该结点设置为取消状态
    if (failed) {
      node.waitstatus = node.cancelled;
    }
  }
}

将当前线程包装成结点添加到条件队列尾部后，紧接着就调用fullyrelease方法释放锁。注意，方法名为fullyrelease也就这步操作会完全的释放锁，因为锁是可重入的，所以在进行条件等待前需要将锁全部释放了，不然的话别人就获取不了锁了。如果释放锁失败的话就会抛出一个运行时异常，如果成功释放了锁的话就返回之前的同步状态。

第三步：进行条件等待

//线程一直在while循环里进行条件等待
while (!isonsyncqueue(node)) {
  //进行条件等待的线程都在这里被挂起, 线程被唤醒的情况有以下几种：
  //1.同步队列的前继结点已取消
  //2.设置同步队列的前继结点的状态为signal失败
  //3.前继结点释放锁后唤醒当前结点
  locksupport.park(this);
  //当前线程醒来后立马检查是否被中断, 如果是则代表结点取消条件等待, 此时需要将结点移出条件队列
  if ((interruptmode = checkinterruptwhilewaiting(node)) != 0) {
    break;
  }
}

//检查条件等待时的线程中断情况
private int checkinterruptwhilewaiting(node node) {
  //中断请求在signal操作之前：throw_ie
  //中断请求在signal操作之后：reinterrupt
  //期间没有收到任何中断请求：0
  return thread.interrupted() ? (transferaftercancelledwait(node) ? throw_ie : reinterrupt) : 0;
}

//将取消条件等待的结点从条件队列转移到同步队列中
final boolean transferaftercancelledwait(node node) {
  //如果这步cas操作成功的话就表明中断发生在signal方法之前
  if (compareandsetwaitstatus(node, node.condition, 0)) {
    //状态修改成功后就将该结点放入同步队列尾部
    enq(node);
    return true;
  }
  //到这里表明cas操作失败, 说明中断发生在signal方法之后
  while (!isonsyncqueue(node)) {
    //如果sinal方法还没有将结点转移到同步队列, 就通过自旋等待一下
    thread.yield();
  }
  return false;
}

在以上两个操作完成了之后就会进入while循环，可以看到while循环里面首先调用locksupport.park(this)将线程挂起了，所以线程就会一直在这里阻塞。在调用signal方法后仅仅只是将结点从条件队列转移到同步队列中去，至于会不会唤醒线程需要看情况。如果转移结点时发现同步队列中的前继结点已取消，或者是更新前继结点的状态为signal失败，这两种情况都会立即唤醒线程，否则的话在signal方法结束时就不会去唤醒已在同步队列中的线程，而是等到它的前继结点来唤醒。当然，线程阻塞在这里除了可以调用signal方法唤醒之外，线程还可以响应中断，如果线程在这里收到中断请求就会继续往下执行。可以看到线程醒来后会马上检查是否是由于中断唤醒的还是通过signal方法唤醒的，如果是因为中断唤醒的同样会将这个结点转移到同步队列中去，只不过是通过调用transferaftercancelledwait方法来实现的。最后执行完这一步之后就会返回中断情况并跳出while循环。

第四步：结点移出条件队列后的操作

//线程醒来后就会以独占模式获取锁
if (acquirequeued(node, savedstate) && interruptmode != throw_ie) {
  interruptmode = reinterrupt;
}
//这步操作主要为防止线程在signal之前中断而导致没与条件队列断绝联系
if (node.nextwaiter != null) {
  unlinkcancelledwaiters();
}
//根据中断模式进行响应的中断处理
if (interruptmode != 0) {
  reportinterruptafterwait(interruptmode);
}

//结束条件等待后根据中断情况做出相应处理
private void reportinterruptafterwait(int interruptmode) throws interruptedexception {
  //如果中断模式是throw_ie就抛出异常
  if (interruptmode == throw_ie) {
    throw new interruptedexception();
  //如果中断模式是reinterrupt就自己挂起
  } else if (interruptmode == reinterrupt) {
    selfinterrupt();
  }
}

当线程终止了while循环也就是条件等待后，就会回到同步队列中。不管是因为调用signal方法回去的还是因为线程中断导致的，结点最终都会在同步队列中。这时就会调用acquirequeued方法执行在同步队列中获取锁的操作，这个方法我们在独占模式这一篇已经详细的讲过。也就是说，结点从条件队列出来后又是乖乖的走独占模式下获取锁的那一套，等这个结点再次获得锁之后，就会调用reportinterruptafterwait方法来根据这期间的中断情况做出相应的响应。如果中断发生在signal方法之前，interruptmode就为throw_ie，再次获得锁后就抛出异常；如果中断发生在signal方法之后，interruptmode就为reinterrupt，再次获得锁后就重新中断。

2.不响应线程中断的条件等待

//不响应线程中断的条件等待
public final void awaituninterruptibly() {
  //将当前线程添加到条件队列尾部
  node node = addconditionwaiter();
  //完全释放锁并返回当前同步状态
  int savedstate = fullyrelease(node);
  boolean interrupted = false;
  //结点一直在while循环里进行条件等待
  while (!isonsyncqueue(node)) {
    //条件队列中所有的线程都在这里被挂起
    locksupport.park(this);
    //线程醒来发现中断并不会马上去响应
    if (thread.interrupted()) {
      interrupted = true;
    }
  }
  if (acquirequeued(node, savedstate) || interrupted) {
    //在这里响应所有中断请求, 满足以下两个条件之一就会将自己挂起
    //1.线程在条件等待时收到中断请求
    //2.线程在acquirequeued方法里收到中断请求
    selfinterrupt();
  }
}

3.设置相对时间的条件等待(不进行自旋)

//设置定时条件等待(相对时间), 不进行自旋等待
public final long awaitnanos(long nanostimeout) throws interruptedexception {
  //如果线程被中断则抛出异常
  if (thread.interrupted()) {
    throw new interruptedexception();
  }
  //将当前线程添加到条件队列尾部
  node node = addconditionwaiter();
  //在进入条件等待之前先完全释放锁
  int savedstate = fullyrelease(node);
  long lasttime = system.nanotime();
  int interruptmode = 0;
  while (!isonsyncqueue(node)) {
    //判断超时时间是否用完了
    if (nanostimeout <= 0l) {
      //如果已超时就需要执行取消条件等待操作
      transferaftercancelledwait(node);
      break;
    }
    //将当前线程挂起一段时间, 线程在这期间可能被唤醒, 也可能自己醒来
    locksupport.parknanos(this, nanostimeout);
    //线程醒来后先检查中断信息
    if ((interruptmode = checkinterruptwhilewaiting(node)) != 0) {
      break;
    }
    long now = system.nanotime();
    //超时时间每次减去条件等待的时间
    nanostimeout -= now - lasttime;
    lasttime = now;
  }
  //线程醒来后就会以独占模式获取锁
  if (acquirequeued(node, savedstate) && interruptmode != throw_ie) {
    interruptmode = reinterrupt;
  }
  //由于transferaftercancelledwait方法没有把nextwaiter置空, 所有这里要再清理一遍
  if (node.nextwaiter != null) {
    unlinkcancelledwaiters();
  }
  //根据中断模式进行响应的中断处理
  if (interruptmode != 0) {
    reportinterruptafterwait(interruptmode);
  }
  //返回剩余时间
  return nanostimeout - (system.nanotime() - lasttime);
}

4.设置相对时间的条件等待(进行自旋)

//设置定时条件等待(相对时间), 进行自旋等待
public final boolean await(long time, timeunit unit) throws interruptedexception {
  if (unit == null) { throw new nullpointerexception(); }
  //获取超时时间的毫秒数
  long nanostimeout = unit.tonanos(time);
  //如果线程被中断则抛出异常
  if (thread.interrupted()) { throw new interruptedexception(); }
  //将当前线程添加条件队列尾部
  node node = addconditionwaiter();
  //在进入条件等待之前先完全释放锁
  int savedstate = fullyrelease(node);
  //获取当前时间的毫秒数
  long lasttime = system.nanotime();
  boolean timedout = false;
  int interruptmode = 0;
  while (!isonsyncqueue(node)) {
    //如果超时就需要执行取消条件等待操作
    if (nanostimeout <= 0l) {
      timedout = transferaftercancelledwait(node);
      break;
    }
    //如果超时时间大于自旋时间, 就将线程挂起一段时间
    if (nanostimeout >= spinfortimeoutthreshold) {
      locksupport.parknanos(this, nanostimeout);
    }
    //线程醒来后先检查中断信息
    if ((interruptmode = checkinterruptwhilewaiting(node)) != 0) {
      break;
    }
    long now = system.nanotime();
    //超时时间每次减去条件等待的时间
    nanostimeout -= now - lasttime;
    lasttime = now;
  }
  //线程醒来后就会以独占模式获取锁
  if (acquirequeued(node, savedstate) && interruptmode != throw_ie) {
    interruptmode = reinterrupt;
  }
  //由于transferaftercancelledwait方法没有把nextwaiter置空, 所有这里要再清理一遍
  if (node.nextwaiter != null) {
    unlinkcancelledwaiters();
  }
  //根据中断模式进行响应的中断处理
  if (interruptmode != 0) {
    reportinterruptafterwait(interruptmode);
  }
  //返回是否超时标志
  return !timedout;
}

5.设置绝对时间的条件等待

//设置定时条件等待(绝对时间)
public final boolean awaituntil(date deadline) throws interruptedexception {
  if (deadline == null) { throw new nullpointerexception(); } 
  //获取绝对时间的毫秒数
  long abstime = deadline.gettime();
  //如果线程被中断则抛出异常
  if (thread.interrupted()) { throw new interruptedexception(); }
  //将当前线程添加到条件队列尾部
  node node = addconditionwaiter();
  //在进入条件等待之前先完全释放锁
  int savedstate = fullyrelease(node);
  boolean timedout = false;
  int interruptmode = 0;
  while (!isonsyncqueue(node)) {
    //如果超时就需要执行取消条件等待操作
    if (system.currenttimemillis() > abstime) {
      timedout = transferaftercancelledwait(node);
      break;
    }
    //将线程挂起一段时间, 期间线程可能被唤醒, 也可能到了点自己醒来
    locksupport.parkuntil(this, abstime);
    //线程醒来后先检查中断信息
    if ((interruptmode = checkinterruptwhilewaiting(node)) != 0) {
      break;
    }
  }
  //线程醒来后就会以独占模式获取锁
  if (acquirequeued(node, savedstate) && interruptmode != throw_ie) {
    interruptmode = reinterrupt;
  }
  //由于transferaftercancelledwait方法没有把nextwaiter置空, 所有这里要再清理一遍
  if (node.nextwaiter != null) {
    unlinkcancelledwaiters();
  }
  //根据中断模式进行响应的中断处理
  if (interruptmode != 0) {
    reportinterruptafterwait(interruptmode);
  }
  //返回是否超时标志
  return !timedout;
}

6.唤醒条件队列中的头结点

//唤醒条件队列中的下一个结点
public final void signal() {
  //判断当前线程是否持有锁
  if (!isheldexclusively()) {
    throw new illegalmonitorstateexception();
  }
  node first = firstwaiter;
  //如果条件队列中有排队者
  if (first != null) {
    //唤醒条件队列中的头结点
    dosignal(first);
  }
}

//唤醒条件队列中的头结点
private void dosignal(node first) {
  do {
    //1.将firstwaiter引用向后移动一位
    if ( (firstwaiter = first.nextwaiter) == null) {
      lastwaiter = null;
    }
    //2.将头结点的后继结点引用置空
    first.nextwaiter = null;
    //3.将头结点转移到同步队列, 转移完成后有可能唤醒线程
    //4.如果transferforsignal操作失败就去唤醒下一个结点
  } while (!transferforsignal(first) && (first = firstwaiter) != null);
}

//将指定结点从条件队列转移到同步队列中
final boolean transferforsignal(node node) {
  //将等待状态从condition设置为0
  if (!compareandsetwaitstatus(node, node.condition, 0)) {
    //如果更新状态的操作失败就直接返回false
    //可能是transferaftercancelledwait方法先将状态改变了, 导致这步cas操作失败
    return false;
  }
  //将该结点添加到同步队列尾部
  node p = enq(node);
  int ws = p.waitstatus;
  if (ws > 0 || !compareandsetwaitstatus(p, ws, node.signal)) {
    //出现以下情况就会唤醒当前线程
    //1.前继结点是取消状态
    //2.更新前继结点的状态为signal操作失败
    locksupport.unpark(node.thread);
  }
  return true;
}

可以看到signal方法最终的核心就是去调用transferforsignal方法，在transferforsignal方法中首先会用cas操作将结点的状态从condition设置为0，然后再调用enq方法将该结点添加到同步队列尾部。我们再看到接下来的if判断语句，这个判断语句主要是用来判断什么时候会去唤醒线程，出现这两种情况就会立即唤醒线程，一种是当发现前继结点的状态是取消状态时，还有一种是更新前继结点的状态失败时。这两种情况都会马上去唤醒线程，否则的话就仅仅只是将结点从条件队列中转移到同步队列中就完了，而不会立马去唤醒结点中的线程。signalall方法也大致类似，只不过它是去循环遍历条件队列中的所有结点，并将它们转移到同步队列，转移结点的方法也还是调用transferforsignal方法。

7.唤醒条件队列的所有结点

//唤醒条件队列后面的全部结点
public final void signalall() {
  //判断当前线程是否持有锁
  if (!isheldexclusively()) {
    throw new illegalmonitorstateexception();
  }
  //获取条件队列头结点
  node first = firstwaiter;
  if (first != null) {
    //唤醒条件队列的所有结点
    dosignalall(first);
  }
}

//唤醒条件队列的所有结点
private void dosignalall(node first) {
  //先把头结点和尾结点的引用置空
  lastwaiter = firstwaiter = null;
  do {
    //先获取后继结点的引用
    node next = first.nextwaiter;
    //把即将转移的结点的后继引用置空
    first.nextwaiter = null;
    //将结点从条件队列转移到同步队列
    transferforsignal(first);
    //将引用指向下一个结点
    first = next;
  } while (first != null);
}

至此，我们整个的abstractqueuedsynchronizer源码分析就结束了，相信通过这四篇的分析，大家能更好的掌握并理解aqs。这个类确实很重要，因为它是其他很多同步类的基石，由于笔者水平和表达能力有限，如果哪些地方没有表述清楚的，或者理解不到位的，还请广大读者们能够及时指正，共同探讨学习。可在下方留言阅读中所遇到的问题，如果有需要aqs注释源码的也可联系笔者索取。

注：以上全部分析基于jdk1.7，不同版本间会有差异，读者需要注意。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。