Java并發(fā)系列之AbstractQueuedSynchronizer源碼分析（條件隊(duì)列）

public interface Condition {
  
  //響應(yīng)線程中斷的條件等待
  void await() throws InterruptedException;
  
  //不響應(yīng)線程中斷的條件等待
  void awaitUninterruptibly();
  
  //設(shè)置相對(duì)時(shí)間的條件等待(不進(jìn)行自旋)
  long awaitNanos(long nanosTimeout) throws InterruptedException;
  
  //設(shè)置相對(duì)時(shí)間的條件等待(進(jìn)行自旋)
  boolean await(long time, TimeUnit unit) throws InterruptedException;
  
  //設(shè)置絕對(duì)時(shí)間的條件等待
  boolean awaitUntil(Date deadline) throws InterruptedException;
  
  //喚醒條件隊(duì)列中的頭結(jié)點(diǎn)
  void signal();
  
  //喚醒條件隊(duì)列的所有結(jié)點(diǎn)
  void signalAll();
  
}

//響應(yīng)線程中斷的條件等待
public final void await() throws InterruptedException {
  //如果線程被中斷則拋出異常
  if (Thread.interrupted()) {
    throw new InterruptedException();
  }
  //將當(dāng)前線程添加到條件隊(duì)列尾部
  Node node = addConditionWaiter();
  //在進(jìn)入條件等待之前先完全釋放鎖
  int savedState = fullyRelease(node);
  int interruptMode = 0;
  //線程一直在while循環(huán)里進(jìn)行條件等待
  while (!isOnSyncQueue(node)) {
    //進(jìn)行條件等待的線程都在這里被掛起, 線程被喚醒的情況有以下幾種：
    //1.同步隊(duì)列的前繼結(jié)點(diǎn)已取消
    //2.設(shè)置同步隊(duì)列的前繼結(jié)點(diǎn)的狀態(tài)為SIGNAL失敗
    //3.前繼結(jié)點(diǎn)釋放鎖后喚醒當(dāng)前結(jié)點(diǎn)
    LockSupport.park(this);
    //當(dāng)前線程醒來后立馬檢查是否被中斷, 如果是則代表結(jié)點(diǎn)取消條件等待, 此時(shí)需要將結(jié)點(diǎn)移出條件隊(duì)列
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) {
      break;
    }
  }
  //線程醒來后就會(huì)以獨(dú)占模式獲取鎖
  if (acquireQueued(node, savedState) && interruptMode != THROW_IE) {
    interruptMode = REINTERRUPT;
  }
  //這步操作主要為防止線程在signal之前中斷而導(dǎo)致沒與條件隊(duì)列斷絕聯(lián)系
  if (node.nextWaiter != null) {
    unlinkCancelledWaiters();
  }
  //根據(jù)中斷模式進(jìn)行響應(yīng)的中斷處理
  if (interruptMode != 0) {
    reportInterruptAfterWait(interruptMode);
  }
}

//完全釋放鎖
final int fullyRelease(Node node) {
  boolean failed = true;
  try {
    //獲取當(dāng)前的同步狀態(tài)
    int savedState = getState();
    //使用當(dāng)前的同步狀態(tài)去釋放鎖
    if (release(savedState)) {
      failed = false;
      //如果釋放鎖成功就返回當(dāng)前同步狀態(tài)
      return savedState;
    } else {
      //如果釋放鎖失敗就拋出運(yùn)行時(shí)異常
      throw new IllegalMonitorStateException();
    }
  } finally {
    //保證沒有成功釋放鎖就將該結(jié)點(diǎn)設(shè)置為取消狀態(tài)
    if (failed) {
      node.waitStatus = Node.CANCELLED;
    }
  }
}

//線程一直在while循環(huán)里進(jìn)行條件等待
while (!isOnSyncQueue(node)) {
  //進(jìn)行條件等待的線程都在這里被掛起, 線程被喚醒的情況有以下幾種：
  //1.同步隊(duì)列的前繼結(jié)點(diǎn)已取消
  //2.設(shè)置同步隊(duì)列的前繼結(jié)點(diǎn)的狀態(tài)為SIGNAL失敗
  //3.前繼結(jié)點(diǎn)釋放鎖后喚醒當(dāng)前結(jié)點(diǎn)
  LockSupport.park(this);
  //當(dāng)前線程醒來后立馬檢查是否被中斷, 如果是則代表結(jié)點(diǎn)取消條件等待, 此時(shí)需要將結(jié)點(diǎn)移出條件隊(duì)列
  if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) {
    break;
  }
}

//檢查條件等待時(shí)的線程中斷情況
private int checkInterruptWhileWaiting(Node node) {
  //中斷請(qǐng)求在signal操作之前：THROW_IE
  //中斷請(qǐng)求在signal操作之后：REINTERRUPT
  //期間沒有收到任何中斷請(qǐng)求：0
  return Thread.interrupted() ? (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 0;
}

//將取消條件等待的結(jié)點(diǎn)從條件隊(duì)列轉(zhuǎn)移到同步隊(duì)列中
final boolean transferAfterCancelledWait(Node node) {
  //如果這步CAS操作成功的話就表明中斷發(fā)生在signal方法之前
  if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
    //狀態(tài)修改成功后就將該結(jié)點(diǎn)放入同步隊(duì)列尾部
    enq(node);
    return true;
  }
  //到這里表明CAS操作失敗, 說明中斷發(fā)生在signal方法之后
  while (!isOnSyncQueue(node)) {
    //如果sinal方法還沒有將結(jié)點(diǎn)轉(zhuǎn)移到同步隊(duì)列, 就通過自旋等待一下
    Thread.yield();
  }
  return false;
}

//線程醒來后就會(huì)以獨(dú)占模式獲取鎖
if (acquireQueued(node, savedState) && interruptMode != THROW_IE) {
  interruptMode = REINTERRUPT;
}
//這步操作主要為防止線程在signal之前中斷而導(dǎo)致沒與條件隊(duì)列斷絕聯(lián)系
if (node.nextWaiter != null) {
  unlinkCancelledWaiters();
}
//根據(jù)中斷模式進(jìn)行響應(yīng)的中斷處理
if (interruptMode != 0) {
  reportInterruptAfterWait(interruptMode);
}

//結(jié)束條件等待后根據(jù)中斷情況做出相應(yīng)處理
private void reportInterruptAfterWait(int interruptMode) throws InterruptedException {
  //如果中斷模式是THROW_IE就拋出異常
  if (interruptMode == THROW_IE) {
    throw new InterruptedException();
  //如果中斷模式是REINTERRUPT就自己掛起
  } else if (interruptMode == REINTERRUPT) {
    selfInterrupt();
  }
}

//不響應(yīng)線程中斷的條件等待
public final void awaitUninterruptibly() {
  //將當(dāng)前線程添加到條件隊(duì)列尾部
  Node node = addConditionWaiter();
  //完全釋放鎖并返回當(dāng)前同步狀態(tài)
  int savedState = fullyRelease(node);
  boolean interrupted = false;
  //結(jié)點(diǎn)一直在while循環(huán)里進(jìn)行條件等待
  while (!isOnSyncQueue(node)) {
    //條件隊(duì)列中所有的線程都在這里被掛起
    LockSupport.park(this);
    //線程醒來發(fā)現(xiàn)中斷并不會(huì)馬上去響應(yīng)
    if (Thread.interrupted()) {
      interrupted = true;
    }
  }
  if (acquireQueued(node, savedState) || interrupted) {
    //在這里響應(yīng)所有中斷請(qǐng)求, 滿足以下兩個(gè)條件之一就會(huì)將自己掛起
    //1.線程在條件等待時(shí)收到中斷請(qǐng)求
    //2.線程在acquireQueued方法里收到中斷請(qǐng)求
    selfInterrupt();
  }
}

//設(shè)置定時(shí)條件等待(相對(duì)時(shí)間), 不進(jìn)行自旋等待
public final long awaitNanos(long nanosTimeout) throws InterruptedException {
  //如果線程被中斷則拋出異常
  if (Thread.interrupted()) {
    throw new InterruptedException();
  }
  //將當(dāng)前線程添加到條件隊(duì)列尾部
  Node node = addConditionWaiter();
  //在進(jìn)入條件等待之前先完全釋放鎖
  int savedState = fullyRelease(node);
  long lastTime = System.nanoTime();
  int interruptMode = 0;
  while (!isOnSyncQueue(node)) {
    //判斷超時(shí)時(shí)間是否用完了
    if (nanosTimeout <= 0L) {
      //如果已超時(shí)就需要執(zhí)行取消條件等待操作
      transferAfterCancelledWait(node);
      break;
    }
    //將當(dāng)前線程掛起一段時(shí)間, 線程在這期間可能被喚醒, 也可能自己醒來
    LockSupport.parkNanos(this, nanosTimeout);
    //線程醒來后先檢查中斷信息
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) {
      break;
    }
    long now = System.nanoTime();
    //超時(shí)時(shí)間每次減去條件等待的時(shí)間
    nanosTimeout -= now - lastTime;
    lastTime = now;
  }
  //線程醒來后就會(huì)以獨(dú)占模式獲取鎖
  if (acquireQueued(node, savedState) && interruptMode != THROW_IE) {
    interruptMode = REINTERRUPT;
  }
  //由于transferAfterCancelledWait方法沒有把nextWaiter置空, 所有這里要再清理一遍
  if (node.nextWaiter != null) {
    unlinkCancelledWaiters();
  }
  //根據(jù)中斷模式進(jìn)行響應(yīng)的中斷處理
  if (interruptMode != 0) {
    reportInterruptAfterWait(interruptMode);
  }
  //返回剩余時(shí)間
  return nanosTimeout - (System.nanoTime() - lastTime);
}

//設(shè)置定時(shí)條件等待(相對(duì)時(shí)間), 進(jìn)行自旋等待
public final boolean await(long time, TimeUnit unit) throws InterruptedException {
  if (unit == null) { throw new NullPointerException(); }
  //獲取超時(shí)時(shí)間的毫秒數(shù)
  long nanosTimeout = unit.toNanos(time);
  //如果線程被中斷則拋出異常
  if (Thread.interrupted()) { throw new InterruptedException(); }
  //將當(dāng)前線程添加條件隊(duì)列尾部
  Node node = addConditionWaiter();
  //在進(jìn)入條件等待之前先完全釋放鎖
  int savedState = fullyRelease(node);
  //獲取當(dāng)前時(shí)間的毫秒數(shù)
  long lastTime = System.nanoTime();
  boolean timedout = false;
  int interruptMode = 0;
  while (!isOnSyncQueue(node)) {
    //如果超時(shí)就需要執(zhí)行取消條件等待操作
    if (nanosTimeout <= 0L) {
      timedout = transferAfterCancelledWait(node);
      break;
    }
    //如果超時(shí)時(shí)間大于自旋時(shí)間, 就將線程掛起一段時(shí)間
    if (nanosTimeout >= spinForTimeoutThreshold) {
      LockSupport.parkNanos(this, nanosTimeout);
    }
    //線程醒來后先檢查中斷信息
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) {
      break;
    }
    long now = System.nanoTime();
    //超時(shí)時(shí)間每次減去條件等待的時(shí)間
    nanosTimeout -= now - lastTime;
    lastTime = now;
  }
  //線程醒來后就會(huì)以獨(dú)占模式獲取鎖
  if (acquireQueued(node, savedState) && interruptMode != THROW_IE) {
    interruptMode = REINTERRUPT;
  }
  //由于transferAfterCancelledWait方法沒有把nextWaiter置空, 所有這里要再清理一遍
  if (node.nextWaiter != null) {
    unlinkCancelledWaiters();
  }
  //根據(jù)中斷模式進(jìn)行響應(yīng)的中斷處理
  if (interruptMode != 0) {
    reportInterruptAfterWait(interruptMode);
  }
  //返回是否超時(shí)標(biāo)志
  return !timedout;
}

//設(shè)置定時(shí)條件等待(絕對(duì)時(shí)間)
public final boolean awaitUntil(Date deadline) throws InterruptedException {
  if (deadline == null) { throw new NullPointerException(); } 
  //獲取絕對(duì)時(shí)間的毫秒數(shù)
  long abstime = deadline.getTime();
  //如果線程被中斷則拋出異常
  if (Thread.interrupted()) { throw new InterruptedException(); }
  //將當(dāng)前線程添加到條件隊(duì)列尾部
  Node node = addConditionWaiter();
  //在進(jìn)入條件等待之前先完全釋放鎖
  int savedState = fullyRelease(node);
  boolean timedout = false;
  int interruptMode = 0;
  while (!isOnSyncQueue(node)) {
    //如果超時(shí)就需要執(zhí)行取消條件等待操作
    if (System.currentTimeMillis() > abstime) {
      timedout = transferAfterCancelledWait(node);
      break;
    }
    //將線程掛起一段時(shí)間, 期間線程可能被喚醒, 也可能到了點(diǎn)自己醒來
    LockSupport.parkUntil(this, abstime);
    //線程醒來后先檢查中斷信息
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) {
      break;
    }
  }
  //線程醒來后就會(huì)以獨(dú)占模式獲取鎖
  if (acquireQueued(node, savedState) && interruptMode != THROW_IE) {
    interruptMode = REINTERRUPT;
  }
  //由于transferAfterCancelledWait方法沒有把nextWaiter置空, 所有這里要再清理一遍
  if (node.nextWaiter != null) {
    unlinkCancelledWaiters();
  }
  //根據(jù)中斷模式進(jìn)行響應(yīng)的中斷處理
  if (interruptMode != 0) {
    reportInterruptAfterWait(interruptMode);
  }
  //返回是否超時(shí)標(biāo)志
  return !timedout;
}

//喚醒條件隊(duì)列中的下一個(gè)結(jié)點(diǎn)
public final void signal() {
  //判斷當(dāng)前線程是否持有鎖
  if (!isHeldExclusively()) {
    throw new IllegalMonitorStateException();
  }
  Node first = firstWaiter;
  //如果條件隊(duì)列中有排隊(duì)者
  if (first != null) {
    //喚醒條件隊(duì)列中的頭結(jié)點(diǎn)
    doSignal(first);
  }
}

//喚醒條件隊(duì)列中的頭結(jié)點(diǎn)
private void doSignal(Node first) {
  do {
    //1.將firstWaiter引用向后移動(dòng)一位
    if ( (firstWaiter = first.nextWaiter) == null) {
      lastWaiter = null;
    }
    //2.將頭結(jié)點(diǎn)的后繼結(jié)點(diǎn)引用置空
    first.nextWaiter = null;
    //3.將頭結(jié)點(diǎn)轉(zhuǎn)移到同步隊(duì)列, 轉(zhuǎn)移完成后有可能喚醒線程
    //4.如果transferForSignal操作失敗就去喚醒下一個(gè)結(jié)點(diǎn)
  } while (!transferForSignal(first) && (first = firstWaiter) != null);
}

//將指定結(jié)點(diǎn)從條件隊(duì)列轉(zhuǎn)移到同步隊(duì)列中
final boolean transferForSignal(Node node) {
  //將等待狀態(tài)從CONDITION設(shè)置為0
  if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
    //如果更新狀態(tài)的操作失敗就直接返回false
    //可能是transferAfterCancelledWait方法先將狀態(tài)改變了, 導(dǎo)致這步CAS操作失敗
    return false;
  }
  //將該結(jié)點(diǎn)添加到同步隊(duì)列尾部
  Node p = enq(node);
  int ws = p.waitStatus;
  if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) {
    //出現(xiàn)以下情況就會(huì)喚醒當(dāng)前線程
    //1.前繼結(jié)點(diǎn)是取消狀態(tài)
    //2.更新前繼結(jié)點(diǎn)的狀態(tài)為SIGNAL操作失敗
    LockSupport.unpark(node.thread);
  }
  return true;
}

//喚醒條件隊(duì)列后面的全部結(jié)點(diǎn)
public final void signalAll() {
  //判斷當(dāng)前線程是否持有鎖
  if (!isHeldExclusively()) {
    throw new IllegalMonitorStateException();
  }
  //獲取條件隊(duì)列頭結(jié)點(diǎn)
  Node first = firstWaiter;
  if (first != null) {
    //喚醒條件隊(duì)列的所有結(jié)點(diǎn)
    doSignalAll(first);
  }
}

//喚醒條件隊(duì)列的所有結(jié)點(diǎn)
private void doSignalAll(Node first) {
  //先把頭結(jié)點(diǎn)和尾結(jié)點(diǎn)的引用置空
  lastWaiter = firstWaiter = null;
  do {
    //先獲取后繼結(jié)點(diǎn)的引用
    Node next = first.nextWaiter;
    //把即將轉(zhuǎn)移的結(jié)點(diǎn)的后繼引用置空
    first.nextWaiter = null;
    //將結(jié)點(diǎn)從條件隊(duì)列轉(zhuǎn)移到同步隊(duì)列
    transferForSignal(first);
    //將引用指向下一個(gè)結(jié)點(diǎn)
    first = next;
  } while (first != null);
}