深扒AQS(二)之condition

概述

我们在第一节中AQS的属性那里说过AQS类中还有一个ConditionObject的内部类

public class ConditionObject implements Condition, java.io.Serializable {

        // 条件队列首节点
        private transient Node firstWaiter;
        // 条件队列尾节点
        private transient Node lastWaiter;
        
        // methods
}

可以看到它继承自Condition,Condition我们好像一直没介绍,因为我默认大家有一些并发基础的,在前边的文章中我们也用过Conition,总之,可以这么认为,每个锁都可以生成很多个不同的condition对象,每个condition都可以wait()当前线程,同样也可以signal(),这个什么时候wait()什么时候signal()就要根据我们的业务逻辑来啦。

接着,我们需要首先明确两个队列的概念,阻塞队列和条件队列。

阻塞队列就是我们上一节所说的队列,这个队列中的节点都想获取锁

条件队列则是由condition形成的条件队列,线程被await操作挂起后就会被放入条件队列,这个队列中的节点都被挂起,他们都等待被signal进入阻塞队列再次获取锁

当前线程被await后,就会被挂起放入条件队列中,看下怎么实现的吧。

await()

public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
                
            // 将当前线程构造成条件节点加入condition的条件队列尾部,node即为构造的节点
            Node node = addConditionWaiter();
            
            // 释放锁,wait是要释放当前持有锁的,返回释放锁之前状态
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            
            // isOnSyncQueue为true代表node已被转移到阻塞队列;
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            // 此时进入阻塞队列,自旋获取锁
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
}
  1. 将当前线程构造为条件节点加入条件队列尾部
  2. 释放锁
  3. 挂起当前线程并等待进入阻塞队列
  4. 线程被signal唤醒进入阻塞队列
  5. 进入阻塞队列自旋获取锁,其实到这里应该明白了,上一节中所谓的资源就是锁

接下来就按顺序分析一哈

加入条件队列

// 将当前线程构造成节点加入条件队列尾部
private Node addConditionWaiter() {
            Node t = lastWaiter;
            
            //尾节点被取消的话,就清除尾节点
            if (t != null && t.waitStatus != Node.CONDITION) {
                // 遍历链表清除状态不是CONDITION的节点
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
            // 当前线程构建CONDITION节点加入条件队列尾部
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
}

可以条件队列节点就是复用上一节中的队列也就是阻塞队列的节点

释放锁

final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            // 当前state
            int savedState = getState();
            if (release(savedState)) {
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            // 释放锁失败
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
}

  • release()我们上一节说过了,释放资源(锁)并唤醒阻塞队列中的下个节点。

  • 这个方法失败,会将节点设置为”取消”状态,并抛出异常 IllegalMonitorStateException。

  • 如果一个线程在不持有 lock 的基础上,就去调用 condition.await() 方法,它能进入条件队列,但是在上面的这个方法中,由于它不持有锁,会抛出异常并成为 CANCELLED节点

等待进入阻塞队列

int interruptMode = 0;

// 如果不在阻塞队列中就挂起线程
while (!isOnSyncQueue(node)) {

    LockSupport.park(this);
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
        break;
}

// 判断node是否处于阻塞队列
final boolean isOnSyncQueue(Node node) {

    // 阻塞队列中的节点状态不是CONDITION,prev是null说明也不在阻塞队列
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
        
    // 此时node的状态必不为CONDITION且node.prev不为null 
    //如果next不为空,肯定在阻塞队列;
    if (node.next != null) 
        return true;
        
    // 从队尾往前找node,找到的话返回true,否则返回false
    return findNodeFromTail(node);
}

// 从阻塞队列队尾往前遍历找node
private boolean findNodeFromTail(Node node) {
        Node t = tail;
        for (;;) {
            if (t == node)
                return true;
            if (t == null)
                return false;
            t = t.prev;
        }
}

唤醒线程,转移到阻塞队列

public final void signal() {

            // 调用signal前必须先持有当前独占锁,注意这个方法需要开发人员实现
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            // 唤醒条件队列中第一个节点,等待最久
            if (first != null)
                doSignal(first);
}

private void doSignal(Node first) {
                // while循环,若first节点迁移不成功,选择first后第一个节点进行转移
            do {
                // 将 firstWaiter 指向 first 节点后面的第一个,因为 first马上要被迁移到阻塞队列
                // 若将 first 移除后,后面没有节点在等待了,那么需要将 lastWaiter 置为 null
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
}

// 将节点从条件队列转移到阻塞队列
final boolean transferForSignal(Node node) {
        
        // cas失败说明其他线程完成了转移,返回继续转移下个节点;成功的话,waitStatus置为0,上一节说过,阻塞队列节点初始状态时waitstatus为0
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        
        // 将该节点自旋加入阻塞队列尾部,p是加入阻塞队列后的前驱节点
        Node p = enq(node);
        int ws = p.waitStatus;
        
        // ws<=0且cas成功的话直接返回true;否则unpark唤醒线程返回true,总之到了这一步迁移已经完成
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

唤醒后检查中断状态

int interruptMode = 0;
while (!isOnSyncQueue(node)) {
    // 线程挂起
    LockSupport.park(this);
    
    // 如果发生了中断退出while循环
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
        break;
}

说了这么久,我们还在这几行代码中转悠,哈哈,不要着急,一行一行来看。我们知道第一次执行到这个程序块儿时,会被park在这个地方。那么,使得此处可以继续执行的有哪些地方呢?

  • 上一步signal后我们的线程已经由条件队列转移到阻塞队列,我们知道进入阻塞队列后有机会被unpark获取锁
  • 上一步signal中的最后一步,转移以后的前驱节点状态是CANCELLED,或者对前驱节点的CAS操作失败了,也会unpark;
  • 假唤醒;
  • park的时候,另一个线程对这个线程中断

现在假设就发生了上边任意一种情况,我们继续往后执行,AQS中定义了几种中断模式常量intrruptMode

// 退出await的时候重新设置中断
 private static final int REINTERRUPT =  1;

// await退出的时候需要抛出异常
 private static final int THROW_IE    = -1;
 
 0 说明await期间没有发生中断
// 判断是否在线程挂起期间发生了中断,如果发生了中断,是 signal 调用之前中断的:抛出异常;还是 signal 之后发生的中断:重新设置中断
private int checkInterruptWhileWaiting(Node node) {
            return Thread.interrupted() ?
                (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
                0;
}

// 线程处于中断状态,才会调用此方法
final boolean transferAfterCancelledWait(Node node) {
        // 如果此时signal已经发生,NODE状态就不再是CONDITION,所以CAS成功的话,说明中断是在signal前发生的
        if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
            // 即使signal还没发生,此时是中断引起的unpark,依然会由于条件队列迁移到阻塞队列尾部
            enq(node);
            return true;
        }
        
        // 执行到这一步,说明上边CAS失败,说明waitstatus已不为CONDITION,即signal已经发生后才发生的中断
        // signal 方法会将节点转移到阻塞队列,但是可能还没完成,这边自旋等待其完成
        while (!isOnSyncQueue(node))
            Thread.yield();
        return false;
}

获取独占锁

到了这一步,条件队列中的节点已经转移到阻塞队列中,可以获取锁了

if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;

acquireQueued我们上一节分析过,这个方法返回的时候,代表当前线程获取了锁,而且 state == savedState了。

该方法的返回值就是代表线程是否被中断,如果返回 true,说明被中断了,而且若是 interruptMode != THROW_IE,说明在 signal 之前就发生中断了,这里将 interruptMode 设置为 REINTERRUPT,用于待会重新中断,其实只是要把这个中断标志保留,留给开发人员用。

if (node.nextWaiter != null) // clean up if cancelled
    unlinkCancelledWaiters();
// 最后处理中断
if (interruptMode != 0)
    reportInterruptAfterWait(interruptMode);

这里突然冒出来个nextWaiter,事实上若是正常signal的话,我们前边代码里很清楚,first.nextWaiter = null,正常signal的话nextWaiter是null;这种不为null的情况实际上就是因为中断引发的,我们说了中断也会使得线程由条件队列进入阻塞队列,此时并没有对后边的条件节点进行处理,正是在这里处理的。

处理中断状态

很简单,不再细说

private void reportInterruptAfterWait(int interruptMode)
            throws InterruptedException {
            if (interruptMode == THROW_IE)
                throw new InterruptedException();
            else if (interruptMode == REINTERRUPT)
                selfInterrupt();
}

static void selfInterrupt() {
        Thread.currentThread().interrupt();
}

总结

又是写了一下午,中间差点把自己绕进去,总之还是写完了,内容有点多,如果有什么纰漏,欢迎指出!

其实就是线程构造条件节点加入条件队列,然后释放锁,然后park等待unpark或者中断,这二者都会使条件节点转为阻塞队列中的节点,接着在阻塞队列中尝试获取锁,最后对中断进行处理。

个人公众号:
《深扒AQS(二)之condition》

    原文作者:幻想波普星
    原文地址: https://blog.csdn.net/qq_26364669/article/details/102863940
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞