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

Posted 兴趣使然の草帽路飞

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了AQS源码探究_06 Conditon条件队列(await方法线程入队与挂起signal方法)相关的知识,希望对你有一定的参考价值。

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;
}

以上是关于AQS源码探究_06 Conditon条件队列(await方法线程入队与挂起signal方法)的主要内容,如果未能解决你的问题,请参考以下文章

AQS源码探究_05 Conditon条件队列(手写一个入门的BrokingQueue)

AQS源码探究_05 Conditon条件队列(手写一个入门的BrokingQueue)

AQS源码探究_02 AQS简介及属性分析

AQS源码探究_02 AQS简介及属性分析

AQS源码探究_08 CyclicBarrier源码分析

AQS源码探究_08 CyclicBarrier源码分析