juc下Condition类解析

Posted 2020-11-30 xxyyy

在使用Lock之前，我们使用的最多的同步方式应该是synchronized关键字来实现同步方式了。配合Object的wait()、notify()系列方法可以实现等待/通知模式。

Condition接口也提供了类似Object的监视器方法，与Lock配合可以实现等待/通知模式，但是这两者在使用方式以及功能特性上还是有差别的。Object和Condition接口的一些对比。摘自《Java并发编程的艺术》

 

一、我们首先来看一下整个Condition的运行机制及其java中Condition接口有哪些方法

 

当我们调用lock.newCondition()方法时，就会创建出一个等待队列（一个lock可以对应多个等待队列）。等我们调用condition的await()方法时，就会把当前的Node节点从阻塞队列中移到等待队列的末尾。

当使用singal()方法时，我们从对应等待队列的队头取出一个Node并且将其放入到阻塞队列中，如果是singalAll()方法,则把对应等待队列中所有的Node节点都移到阻塞队列中。

 

java中Condition接口中具体方法如下

 

二、AQS中ConditionObject的await方法分析

await()方法有多种变种，包括不响应中断的await，带有中断时间的await等等，但是基本思想都与await()方法相似,本文只分析await()方法。await()方法的分析如下

public final void await() throws InterruptedException {
    //If current thread is interrupted, throw InterruptedException.
    if (Thread.interrupted())
        throw new InterruptedException();

    //将调用await()方法的线程封装为一个node节点加入到条件队列中，并且返回该节点的一个引用
    Node node = addConditionWaiter();
    //完全释放掉当前线程对应的锁(将state的值置为0)
    //为什么要释放锁呢？  加着锁 挂起后，谁还能救你呢？
    int savedState = fullyRelease(node);
    //0 在condition队列挂起期间未接收过过中断信号
    //-1 在condition队列挂起期间接收到中断信号了
    //1 在condition队列挂起期间未接收到中断信号，但是迁移到“阻塞队列”之后 接收过中断信号。
    int interruptMode = 0;

    //isOnSyncQueue
    //true:表示当前线程对应的node节点已经迁移到阻塞队列了
    //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);
}

 　　

/**
 *  判断在等待过程中该节点是否被中断过
 */
private int checkInterruptWhileWaiting(Node node) {
    //Thread.interrupted() 返回当前线程中断标记位，并且重置当前标记位 为 false 。
    return Thread.interrupted() ?
            //transferAfterCancelledWait 这个方法只有在线程是被中断唤醒时 才会调用！
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;
}

/**
 * Transfers node, if necessary, to sync queue after a cancelled wait.
 * Returns true if thread was cancelled before being signalled.
 *
 * @param node the node
 * @return true if cancelled before the node was signalled
 */

final boolean transferAfterCancelledWait(Node node) {
    //条件成立：说明当前node一定是在 条件队列内，因为signal 迁移节点到阻塞队列时，会将节点的状态修改为0
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        //中断唤醒的node也会被加入到 阻塞队列中！！
        enq(node);
        //true：表示是在条件队列内被中断的.
        return true;
    }
    /*
     * If we lost out to a signal(), then we can‘t proceed
     * until it finishes its enq().  Cancelling during an
     * incomplete transfer is both rare and transient, so just
     * spin.
     */
    //执行到这里有几种情况？
    //1.当前node已经被外部线程调用 signal 方法将其迁移到 阻塞队列内了。
    //2.当前node正在被外部线程调用 signal 方法将其迁移至 阻塞队列中 进行中状态..

    while (!isOnSyncQueue(node))
        Thread.yield();
    //false:表示当前节点被中断唤醒时 不在 条件队列了..
    return false;
}

/**
 * 将处于取消状态的节点从等待队列中移除
 */
private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    Node trail = null;
    //从头开始遍历
    while (t != null) {
        Node next = t.nextWaiter;
        //条件成立：说明当前节点状态为 取消状态
        if (t.waitStatus != Node.CONDITION) {
            t.nextWaiter = null;
            if (trail == null)
                firstWaiter = next;
            else
                trail.nextWaiter = next;
            if (next == null)
                lastWaiter = trail;
        }
        else
            trail = t;
        t = next;
    }
}

 /**
 * Throws InterruptedException, reinterrupts current thread, or
 * does nothing, depending on mode.
 */
private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    if (interruptMode == THROW_IE)
        throw new InterruptedException();
    else if (interruptMode == REINTERRUPT)
        selfInterrupt();
}

 

 三、AQS中ConditionObject的addConditionWaiter方法分析

/**
 * 将当前线程封装为一个Node加入等待队列中
 * @return 返回封装的Node节点
 *
 * 流程
 * 1、判断当前等待队列的尾节点否为取消状态（t != null && t.waitStatus != Node.CONDITION）
 *    将处于取消状态的节点从等待队列中移除
 * 2、将当前的线程封装为一个Node，并将其加入到等待队列中
 */
private Node addConditionWaiter() {
    //获取当前条件队列的尾节点的引用 保存到局部变量 t中
    Node t = lastWaiter;


    // If lastWaiter is cancelled, clean out.
    //条件一：t != null 成立：说明当前条件队列中，已经有node元素了..
    //条件二：node 在 条件队列中时，它的状态是 CONDITION（-2）
    //     t.waitStatus != Node.CONDITION 成立：说明当前node已经被取消了..
    if (t != null && t.waitStatus != Node.CONDITION) {
        //将处于取消状态的节点从等待队列中移除
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    //将当前的线程封装为一个Node，并将其加入到等待队列中
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

 

 四、AQS中ConditionObject的fullyRelease方法分析

/**
 * 用当前状态值调用释放,返回state的值
 */
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值
            //为什么要返回savedState？
            //因为在当你被迁移到“阻塞队列”后，再次被唤醒，且当前node在阻塞队列中是head.next 而且
            //当前lock状态是state == 0 的情况下，当前node可以获取到锁，此时需要将state 设置为savedState.
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

public final boolean release(int arg) {
    //head什么情况下会被创建出来？
    //当持锁线程未释放线程时，且持锁期间 有其它线程想要获取锁时，其它线程发现获取不了锁，而且队列是空队列，此时后续线程会为当前持锁中的
    //线程 构建出来一个head节点，然后后续线程  会追加到 head 节点后面。
    if (tryRelease(arg)) {
        Node h = head;
        //条件一:成立，说明队列中的head节点已经初始化过了，ReentrantLock 在使用期间 发生过 多线程竞争了...
        //条件二：条件成立，说明当前head后面一定插入过node节点。
        if (h != null && h.waitStatus != 0)
            //唤醒后继节点..
            unparkSuccessor(h);
        return true;
    }
    return false;}

 五、AQS中ConditionObject的isOnSyncQueue方法分析

final boolean isOnSyncQueue(Node node) {
    //条件一：node.waitStatus == Node.CONDITION 条件成立，说明当前node一定是在
    //条件队列，因为signal方法迁移节点到 阻塞队列前，会将node的状态设置为 0
    //条件二：前置条件：node.waitStatus != Node.CONDITION   ===>
    // 1.node.waitStatus == 0 (表示当前节点已经被signal了)
    // 2.node.waitStatus == 1 （当前线程是未持有锁调用await方法..最终会将node的状态修改为 取消状态..）
    //node.waitStatus == 0 为什么还要判断 node.prev == null?
    //因为signal方法 是先修改状态，再迁移。
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    //条件满足，说明其一定在等待队列中
    if (node.next != null) // If has successor, it must be on queue
        return true;
    /*
     * node.prev can be non-null, but not yet on queue because
     * the CAS to place it on queue can fail. So we have to
     * traverse from tail to make sure it actually made it.  It
     * will always be near the tail in calls to this method, and
     * unless the CAS failed (which is unlikely), it will be
     * there, so we hardly ever traverse much.
     */
    /**
     * 执行到这里，说明当前节点的状态为：node.prev != null 且 node.waitStatus == 0
     * findNodeFromTail 从阻塞队列的尾巴开始向前遍历查找node，如果查找到 返回true,查找不到返回false
     * 当前node有可能正在signal过程中，正在迁移中...还未完成...
     */
    return findNodeFromTail(node);
}

private boolean findNodeFromTail(Node node) {
    Node t = tail;
    for (;;) {
        if (t == node)
            return true;
        if (t == null)
            return false;
        t = t.prev;
    }
}

 

六、AQS中singal的逻辑分析（singalAll逻辑相同）

public final void signal() {
    //判断调用signal方法的线程是否是独占锁持有线程，如果不是，直接抛出异常..
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    //获取条件队列的一个node
    Node first = firstWaiter;
    //第一个节点不为null，则将第一个节点 进行迁移到 阻塞队列的逻辑..
    if (first != null)
        doSignal(first);
}

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

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

    /*
     * Splice onto queue and try to set waitStatus of predecessor to
     * indicate that thread is (probably) waiting. If cancelled or
     * attempt to set waitStatus fails, wake up to resync (in which
     * case the waitStatus can be transiently and harmlessly wrong).
     */
    //enq最终会将当前 node 入队到 阻塞队列，p 是当前节点在阻塞队列的 前驱节点.
    Node p = enq(node);
    //条件一：ws > 0 成立：说明前驱节点的状态在阻塞队列中是 取消状态,唤醒当前节点。
    //条件二：前置条件(ws <= 0)，
    //compareAndSetWaitStatus(p, ws, Node.SIGNAL) 返回true 表示设置前驱节点状态为 SIGNAl状态成功
    //compareAndSetWaitStatus(p, ws, Node.SIGNAL) 返回false  ===> 什么时候会false?
    //当前驱node对应的线程 是 lockInterrupt入队的node时，是会响应中断的，外部线程给前驱线程中断信号之后，前驱node会将
    //状态修改为 取消状态，并且执行 出队逻辑..
    //前驱节点状态 只要不是 0 或者 -1 那么，就唤醒当前线程。
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}