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

叁歲伎倆 2022-10-31 01:43 222阅读 0赞

> AQS 系列：
> 
>  *  [【JUC源码】JUC核心：AQS（一）底层结构分析][JUC_JUC_AQS]
>  *  [【JUC源码】JUC核心：AQS（二）同步队列源码分析（独占锁）][JUC_JUC_AQS 1]
>  *  [【JUC源码】JUC核心：AQS（三）同步队列源码分析（共享锁）][JUC_JUC_AQS 2]
>  *  [【JUC源码】JUC核心：AQS（四）条件队列源码分析][JUC_JUC_AQS 3]
>  *  [【JUC源码】JUC核心：关于AQS的几个问题][JUC_JUC_AQS 4]

条件队列：

*  作用
    
     *  实现类似 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);
    }

[JUC_JUC_AQS]: https://blog.csdn.net/weixin_43935927/article/details/108693296
[JUC_JUC_AQS 1]: https://blog.csdn.net/weixin_43935927/article/details/113901484
[JUC_JUC_AQS 2]: https://blog.csdn.net/weixin_43935927/article/details/113904384
[JUC_JUC_AQS 3]: https://blog.csdn.net/weixin_43935927/article/details/113901502
[JUC_JUC_AQS 4]: https://blog.csdn.net/weixin_43935927/article/details/108699927