【JUC源码】JUC核心：AQS（四）条件队列源码分析

AQS 系列：

• 【JUC源码】JUC核心：AQS（一）底层结构分析
• 【JUC源码】JUC核心：AQS（二）同步队列源码分析（独占锁）
• 【JUC源码】JUC核心：AQS（三）同步队列源码分析（共享锁）
• 【JUC源码】JUC核心：AQS（四）条件队列源码分析
• 【JUC源码】JUC核心：关于AQS的几个问题

条件队列：

• 作用

  • 实现类似 synchronized 的 wait 与 signal，实现在使用锁时对线程管理
  • 而且由于实现了 Condition，对线程的管理可以更加细化

• 命名：条件队列中将 node 叫做 waiter

• 策略：

  • 要加入阻塞队列的前提是，当前线程已经拿到了锁，并处于运行状态
  • 加入阻塞队列前要释放锁，即唤醒同步队列的队二结点拿锁运行
  • 条件队列中的所有结点都是在阻塞状态
  • 唤醒操作实际是将一个node从条件队列，移动到同步队列尾，让它去返回同步队列去休眠，并不是随机就唤醒（unpark）一个线程。

• 两个状态

  • CONDITION（-2）：条件队列中所有正常节点
  • 初始化（0）：要移到同步队列的节点

1.休眠

await（核心方法)

将线程的 node 放入条件队列，但该方法实质上整个休眠-唤醒的整体逻辑

1. 为当前线程新建一个 node，并加入条件队列队尾（state=CONDITION）
2. 将 AQS 的状态置为 0，释放当前线程的锁，然后再唤醒一个新线程（此时当前线程还未被阻塞）
3. 将当前线程通过 park() 阻塞，等待被唤醒
4. 待 signal 方法被调用后，就会有结点被移动到同步队列队尾（一般是条件队列队首结点），修改该 node 的状态为初始状态0，修改该节点前一个结点状态为 SIGNAL
5. 被移动到同步队列后，如果阻塞在条件队列的线程被唤醒了，由于此时状态已经不是 CONDITION 了，所以就可以推退出循环
6. 但是有一个问题，条件队列的这个 node 的线程虽然被唤醒了，也退出循环等待了。但是，此刻他还没有拿到锁，所以这里需要手动调用一次 acquireQueued 尝试一次拿锁（注：独占锁的 acquireQueued 是在 acquire 方法调的）

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 加入到条件队列的队尾
    Node node = addConditionWaiter();
    
    // 加入条件队列后，会释放锁，并唤醒队二去tryLock，然后删掉自己（head）
    // 自己马上就要阻塞了，必须释放之前 lock 的资源，不然自己不被唤醒的话，别的线程永远得不到该共享资源了
    int savedState = fullyRelease(node);
    int interruptMode = 0;
   
    // 当一个node刚进入条件队列，它会被阻塞再这里
    // 当某个线程被唤醒，即某个node被移动到同步队列，并且在同步队列中被唤醒了就会就会退出当前循环
    while (!isOnSyncQueue(node)) {
        
        // 阻塞在条件队列上
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
	
	// 线程虽然已经被唤醒，但是还没有锁，所以手动 acquireQueued() 尝试拿锁
	// 如果失败，该线程又会进入休眠
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) 
        // 如果状态不是CONDITION，就会自动删除
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

addConditionWaiter()

将node加入到条件队列尾，返回新添加的 waiter

• 如果尾节点状态不是 CONDITION 状态，删除条件队列中所有状态不是 CONDITION 的节点
• 如果队列为空，新增节点作为队列头节点，否则追加到尾节点上

private Node addConditionWaiter() {
    Node t = lastWaiter;
 
    // 如果尾部的 waiter 不是 CONDITION 状态了，删除所有不是Condition的节点
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    // 新建node，这个node与同步对列的队首不同，但都存着同一个thread
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    // 队列是空的，直接放到队列头
    if (t == null)
        firstWaiter = node;
    // 队列不为空，直接到队列尾部
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

fullRelease()

调用release，释放当前要加入条件对列的node的锁

• release后就会有线程在acquireQueued方法中醒来
• 醒来拿到锁后就会将head（保存当前wait线程的node，注：与条件队列的node不是同一个）

final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            // 一般为1，重入为n
            int savedState = getState();
            // 调用release
            if (release(savedState)) {
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            // 若失败，就将node置为CANCELLED，并删除
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
}

isOnSyncQueue()

确认node不在同步队列上，再阻塞，如果 node 在同步队列上，是不能够上锁的。

• node 刚被加入到条件队列中，立马就被其他线程 signal 转移到同步队列中去了
• 线程之前在条件队列中沉睡，被唤醒后加入到同步队列中去

final boolean isOnSyncQueue(Node node) {
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        if (node.next != null) // If has successor, it must be on queue
            return true;
    
        return findNodeFromTail(node);
}

2.唤醒

signal()

唤醒阻塞在条件队列中的一个节点

public final void signal() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 从头节点开始唤醒
    Node first = firstWaiter;
    if (first != null)
        // doSignal 方法会把条件队列中的节点转移到同步队列中去
        doSignal(first);
}

doSignal()

寻找被唤醒节点，从队首开始尝试转移到同步队列

private void doSignal(Node first) {
    do {
        // firstWaiter（head）依次后移，若nextWaiter为空，说明到队尾了
        // 将firstWaiter置为条件队列中的第二个节点
        // 若第二个节点是null，即当前条件队列中只有一个节点，就将lastWaiter也置为null
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        // 将第一个节点（first）的下一个节点置为null，作用是从条件队列中删除first
        first.nextWaiter = null;
     // 通过while保证一定能转移成功，若失败，就后移一位
    } while (!transferForSignal(first) && (first = firstWaiter) != null);
}   

transferForSignal（核心方法）

将当前节点移动加到同步队列队尾

1. 将node的state改为0（初始化），若失败return false
2. 调用enq将node加到同步队列队尾
   1. 由于node（引用），所以相当于直接将此node从条件队列删除->接到同步队列队尾
   2. 返回的是node在同步队列的前一个结点pre
3. 将pre设置为SIGNAL，标识后面有待唤醒的节点
4. 如果设置失败，直接唤醒当前线程

final boolean transferForSignal(Node node) {
  
    // 将 node 的状态从 CONDITION 修改成初始化，失败返回 false
    // 等下次while循环，再尝试下一个结点
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    // 将node尾插到同步队列（从条件队列删除），返回的 p（pre） 是 node 在同步队列中的前一个节点
    Node p = enq(node);
    int ws = p.waitStatus;
    // 将前一个结点的状态修改成 SIGNAL，如果成功直接返回
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        // 如果前一个结点被取消，或者状态不能修改成SIGNAL，直接唤醒（不一定能拿到锁）
        LockSupport.unpark(node.thread);
    return true;
}

signalAll()

唤醒所有结点

public final void signalAll() {
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        // 拿到头节点
        Node first = firstWaiter;
        if (first != null)
            // 从头节点开始唤醒条件队列中所有的节点
            doSignalAll(first);
}

doSignalAll()

把条件队列所有节点依次转移到同步队列去，即循环调用transferForSignal

private void doSignalAll(Node first) {
    lastWaiter = firstWaiter = null;
    do {
        // 拿出条件队列队列头节点的下一个节点
        Node next = first.nextWaiter;
        // 把头节点从条件队列中删除
        first.nextWaiter = null;
        // 头节点转移到同步队列中去
        transferForSignal(first);
        // 开始循环头节点的下一个节点
        first = next;
    } while (first != null);
}