AQS源码探究_06 Conditon条件队列(await方法线程入队与挂起signal方法)

Posted 2021-05-18 兴趣使然の草帽路飞

AQS源码探究_06 Conditon条件队列(await方法、线程入队与挂起、signal方法)

提示：读源码的时候尽量自己点开源码跟着博客注释一起看，不然容易迷路~

1、条件队列流程图

2、Condition接口

public interface Condition {
	// 线程等待，可抛出中断异常(可以响应中断)
	void await() throws InterruptedException;
	
	// 线程等待，但是不可响应中断
	void awaitUninterruptibly();
	
	// 线程超时等待
	boolean await(long time, TimeUnit unit) throws InterruptedException;
	
	// 线程等待，直到收到被唤醒、或收到中断信号、或到了指定的截止时间日期自动唤醒
	boolean awaitUntil(Date deadline) throws InterruptedException;
	
	// 线程唤醒
	void signal();
	
	// 唤醒所有线程
	void signalAll();
}

3、AQS中实现Condition接口的内部类ConditionObject

// 位于AQS中的内部类：
public class ConditionObject implements Condition, java.io.Serializable {
    private static final long serialVersionUID = 1173984872572414699L;
    // 队列的第一个等待者节点：头等待者
    private transient Node firstWaiter;
    // 队列的最后一个等待者节点：尾等待者
    private transient Node lastWaiter;
    
    /**
     * 无参构造函数：
     * 在ReentrantLock中的newCondition()方法内，借助ReentrantLock的静态内部类Sync，去调用
     * newCondition方法实例化ConditionObject
     */
    public ConditionObject() { }

    ...
}

接下来分析一下ConditionObject的内部方法：

3.1 await线程等待方法

// 线程等待，可抛出中断异常(可以响应中断)
public final void await() throws InterruptedException {
    // 判断当前线程是否是中断状态：
    if (Thread.interrupted())
        // 如果是中断状态，则直接抛出中断异常
        throw new InterruptedException();
    
    // 把当前线程包装成一个node，放入条件队列中，并返回封装当前线程的node
    Node node = addConditionWaiter();
    // 完全释放当前线程对应的锁（将state置为0）:
    // 为什么要释放锁呢？ 因为加着锁挂起的时候，除了本线程能唤醒（本线程已挂起），其他线程都没办法唤醒啊~
    int savedState = fullyRelease(node);
    // Condition队列中断状态：
    // 0: 在Condition队列挂起期间，未接收过中断信号~
    // -1: 在Condition队列挂起期间，接收到中断信号~
    // 1: 在Condition队列挂起期间，未接收到中断信号，但是迁移到阻塞队列之后，接收到过中断信号~
    int interruptMode = 0;
    
    // while循环条件：判断当前node节点是否在阻塞队列中: 
    // isOnSyncQueue返回true：表示当前线程对应的node已经迁移到了阻塞队列中了
    // isOnSyncQueue返回false：说明当前node仍还在条件队列中，需要park挂起~
    while (!isOnSyncQueue(node)) {
        // 挂起当前线程，使其处于等待状态~
        LockSupport.park(this);
        
        //什么时候会被唤醒？都有几种情况呢？
        // 1.常规路径：外部线程获取到lock之后，调用了 signal()方法 转移条件队列的头节点到 阻塞队列， 当这个节点获取到锁后，会唤醒。
        // 2.转移至阻塞队列后，发现阻塞队列中的前驱节点状态 是 取消状态，此时会唤醒当前节点
        // 3.当前节点挂起期间，被外部线程使用中断唤醒..

        // checkInterruptWhileWaiting: 即使在Condition队列挂起期间，线程发生中断了，对应的node仍然会被迁移到阻塞队列中（等待去获取锁）~
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    
    // 执行到这里，说明当前node已经迁移到阻塞队列了
    
    // acquireQueued：竞争队列的逻辑，线程节点竞争资源
    // 条件一：返回true 表示在阻塞队列中 被外部线程中断唤醒过..
    // 条件二：interruptMode != THROW_IE 成立，说明当前node在条件队列内 未发生过中断
    // 设置interruptMode = REINTERRUPT
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    // 考虑下 node.nextWaiter != null 条件什么时候成立呢？
    // 其实是node在条件队列内时 如果被外部线程 中断唤醒时，会加入到阻塞队列，但是并未设置nextWaiter = null。
    if (node.nextWaiter != null) // clean up if cancelled
        // 清理条件队列内取消状态的节点..
        unlinkCancelledWaiters();

    // 条件成立：说明挂起期间 发生过中断（1.条件队列内的挂起 2.条件队列之外的挂起）
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

// 把当前线程包装成一个node，放入条件队列中，并返回封装当前线程的node
// 注意：调用await方法的线程都是持锁状态的线程，也就是说：在await方法中调用的addConditionWaiter方法不存在并发问题~
private Node addConditionWaiter() {
    // 获取当前条件队列的尾节点的引用，保存到局部变量t中
    Node t = lastWaiter;
    // 条件1：t != null 成立：说明当前条件队列中，已经有node元素了
    // 条件2：(node在队列中时，它的状态是 CONDITION(-2))
    // t.waitStatus != Node.CONDITION 成立：说明当前node发生中断了...
    if (t != null && t.waitStatus != Node.CONDITION) {
        // 清理条件队列中，所有CONDITION(-2)取消状态的node节点
        unlinkCancelledWaiters();
        // 更新局部变量t 为最新的尾结点的引用~ 
        // 因为unlinkCancelledWaiters方法可能会更改lastWaiter的引用
        t = lastWaiter;
    }
    // 为当前线程创建node节点，设置状态为 CONDITION(-2)
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    
    // 如果t == null 条件成立，说明条件队列中没有任何元素，当前线程是进入队列的第一个元素
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    
    // 更新队尾节点引用指向node
    lastWaiter = node;
    
    // 返回包装当前线程的node
    return node;
}

// 即使在Condition队列挂起期间，线程发生中断了，对应的node仍然会被迁移到阻塞队列中（等待去获取锁）~
private int checkInterruptWhileWaiting(Node node) {
    // Thread.interrupted() 返回当前线程中断标记位，并且重置当前标记位为false
    return Thread.interrupted() ?
            // transferAfterCancelledWait 这个方法只有在线程是被中断唤醒时才会调用！
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 0;
}

final boolean transferAfterCancelledWait(Node node) {
    // 条件成立：说明当前node一定是在 条件队列内，因为signal 迁移节点到阻塞队列时，会将节点的状态修改为0
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        // 中断唤醒的node也会被加入到 阻塞队列中！！
        enq(node);
        // true：表示是在条件队列内被中断的.
        return true;
    }

    // 执行到这里有几种情况？
    // 1.当前node已经被外部线程调用 signal 方法将其迁移到 阻塞队列内了。
    // 2.当前node正在被外部线程调用 signal 方法将其迁移至 阻塞队列中 进行中状态..

    while (!isOnSyncQueue(node))
        Thread.yield();

    // false:表示当前节点被中断唤醒时 不在 条件队列了..
    return false;
}

// 清理条件队列内取消状态的节点..(纯链表操作的方法)
private void unlinkCancelledWaiters() {
    // 表示循环当前节点，从链表的第一个节点开始 向后迭代处理.
    Node t = firstWaiter;
    // 当前链表上一个正常状态的node节点
    Node trail = null;

    while (t != null) {
        // 当前节点的下一个节点.
        Node next = t.nextWaiter;
        // 条件成立：说明当前节点状态为 取消状态
        if (t.waitStatus != Node.CONDITION) {
            // 更新nextWaiter为null
            t.nextWaiter = null;
            // 条件成立：说明遍历到的节点还未碰到过正常节点..
            if (trail == null)
                // 更新firstWaiter指针为下个节点就可以
                firstWaiter = next;
            else
                // 让上一个正常节点指向 取消节点的 下一个节点..中间有问题的节点 被跳过去了..
                trail.nextWaiter = next;

            // 条件成立：当前节点为队尾节点了，更新lastWaiter 指向最后一个正常节点 就Ok了
            if (next == null)
                lastWaiter = trail;
        }
        else// 条件不成立执行到else，说明当前节点是正常节点
            trail = t;

        t = next;
    }
}

private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    // 条件成立：说明在条件队列内发生过中断，此时await方法抛出中断异常
    if (interruptMode == THROW_IE)
        throw new InterruptedException();

    // 条件成立：说明在条件队列外发生的中断，此时设置当前线程的中断标记位 为true
    // 中断处理 交给 你的业务处理。 如果你不处理，那什么事 也不会发生了...
    else if (interruptMode == REINTERRUPT)
        selfInterrupt();
}

3.2 fullyRelease完全释放当前线程锁的方法

// 位于AQS中：
// 完全释放当前线程对应的锁（将state置为0）:
final int fullyRelease(Node node) {
    // 完全释放锁是否成功，当failed失败时，说明当前线程是未持有锁调用await方法的线程(错误写法...)
    // 假设失败，在finally代码块中，会将刚刚加入到条件队列的，当前线程对应的node节点状态修改为取消状态
    // 后继线程就会将取消状态的节点给清理出去~ 
    boolean failed = true;
    try {
        // 获取当前线程所持有的state值。
        int savedState = getState();
        
        // 绝大部分情况下：release这里会返回true
        if (release(savedState)) {
            // 失败标记设置为false
            failed = false;
            // 返回当前线程释放的state值
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

3.3 isOnSyncQueue判断当前node节点是否在阻塞队列中

// 位于AQS中：
// 判断当前node节点是否在阻塞队列中
final boolean isOnSyncQueue(Node node) {
    // 条件一：node.waitStatus == Node.CONDITION 条件成立：说明当前node一定是在条件队列中，
    // 因为signal方法迁移节点到阻塞队列前，会将node的状态设置为0
    // 条件二：前置条件node.waitStatus != Node.CONDITION ===>然后再细分以下几种情况：
    // 1.node.waitStatus == 0 (表示当前节点已经被singal)
    // 2.node.waitStatus == 1 (表示当前线程未持有锁就调用了await方法，最终会将node的状态修改为取消状态...)
    // node.waitStatus == 0 为什么还要判断node.prev == null呢？
    // 因为：signal方法是先修改状态，再迁移
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    
    // 执行到这里，会是哪种情况？
    // node.waitStatus != CONDITION 且 node.prev != null  ===> 可以排除掉 node.waitStatus == 1 取消状态..
    // 为什么可以排除取消状态？ 因为signal方法是不会把 取消状态的node迁移走的
    // 设置prev引用的逻辑 是 迁移 阻塞队列 逻辑的设置的（enq()）
    // 入队的逻辑：1.设置node.prev = tail;   2. cas当前node为 阻塞队列的 tail 尾节点 成功才算是真正进入到 阻塞队列！ 3.pred.next = node;
    // 可以推算出，就算prev不是null，也不能说明当前node 已经成功入队到 阻塞队列了。
    
    // 条件成立：说明当前节点已经成功入队到阻塞队列，且当前节点后面已经有其它node了...
    if (node.next != null)
        return true;
    
    /**
     * 执行到这里，说明当前节点的状态为：node.prev != null 且 node.waitStatus == 0
     * findNodeFromTail 从阻塞队列的尾巴开始向前遍历查找node，如果查找到 返回true,查找不到返回false
     * 当前node有可能正在signal过程中，正在迁移中...还未完成...
     */
    return findNodeFromTail(node);
}

3.4 signal线程唤醒方法

// 位于AQS中：线程唤醒方法
public final void signal() {
    // 判断调用signal方法的线程是否是独占锁持有线程，如果不是，直接抛出异常..
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();

    // 获取条件队列的一个node
    Node first = firstWaiter;
    // 第一个节点不为null，则将第一个节点 进行迁移到阻塞队列的逻辑..
    if (first != null)
        doSignal(first);
}

// 位于AQS中：将第一个节点 进行迁移到阻塞队列 
private void doSignal(Node first) {
    do {
        // firstWaiter = first.nextWaiter因为当前first马上要出条件队列了，
        // 所以更新firstWaiter为 当前节点的下一个节点..
        // 如果当前节点的下一个节点 是 null，说明条件队列只有当前一个节点了...
        // 当前出队后，整个队列就空了..所以需要更新lastWaiter = null
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;

        // 当前first节点 出 条件队列。断开和下一个节点的关系
        first.nextWaiter = null;

        // transferForSignal(first) 返回boolean类型
        // 返回true 表示当前first节点迁移到阻塞队列成功  返回false 表示迁移失败...
        // while循环 ：(first = firstWaiter) != null  
        // 当前first迁移失败，则将first更新为 first.next 继续尝试迁移..
        // 直至迁移某个节点成功，或者 条件队列为null为止。
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

3.4 transferForSignal当前节点迁移到阻塞队列

// 位于AQS：当前节点迁移到阻塞队列：
// 返回true 表示当前first节点迁移到阻塞队列成功  返回false 表示迁移失败...
final boolean transferForSignal(Node node) {
    // cas修改当前节点的状态，修改为0，因为当前节点马上要迁移到 阻塞队列了
    // 成功：当前节点在条件队列中状态正常。
    // 失败： 1.取消状态 （线程await时 未持有锁，最终线程对应的node会设置为 取消状态）
    //       2.node对应的线程 挂起期间，被其它线程使用 中断信号 唤醒过...
    // （就会主队进入到 阻塞队列，这时也会修改状态为0）
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    // enq最终会将当前 node 入队到 阻塞队列，p 是当前节点在阻塞队列的 前驱节点.
    Node p = enq(node);

    // ws 前驱节点的状态..
    int ws = p.waitStatus;
    // 条件一：ws > 0 成立：说明前驱节点的状态在阻塞队列中是 取消状态,唤醒当前节点。
    // 条件二：前置条件(ws <= 0)，
    // compareAndSetWaitStatus(p, ws, Node.SIGNAL) 返回true 表示设置前驱节点状态为 SIGNAl状态成功
    // compareAndSetWaitStatus(p, ws, Node.SIGNAL) 返回false  ===> 什么时候会false?
    // 当前驱node对应的线程 是 lockInterrupt入队的node时，是会响应中断的，外部线程给前驱线程中断信号之后，前驱node会将
    // 状态修改为 取消状态，并且执行 出队逻辑..
    // 前驱节点状态 只要不是 0 或者 -1 那么，就唤醒当前线程。
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        // 唤醒当前node对应的线程...回头再说。
        LockSupport.unpark(node.thread);

    return true;
}