Java的AQS详解3--Condition接口

Condition定义了等待/通知两种类型的方法，当前线程调用这些方法时，需要提前获取到Condition关联的锁。

Condition对象由Lock对象(调用Lock的newCondition()方法)创建出来的，换句话说，Condition是依赖Lock对象的。

首先跟一下Condition的await方法:

await

需要注意的是：调用await方法的线程肯定是持有锁的线程，即同步队列的头节点。

public final void await() throws InterruptedException {
    // 如果当前线程被中断，抛出中断异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 将当前线程添加到Condition条件队列
    Node node = addConditionWaiter();
    // 当前线程释放掉持有的锁，并将锁之前持有的状态值保存下来
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        // 若当前线程不在同步队列中，则挂起该线程
        // 被唤醒的时机有2个：
        // 1. 当前线程被中断；
        // 2. 当前线程被signal
        LockSupport.park(this);
        // 当前线程被唤醒，检查该线程是否在挂起期间被中断过，若被中断过，跳出while循环
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    // 关于中断模式：
    // 1. 当在被通知前被中断则将中断模式设置为THROW_IE;
    // 2. 当在被通知后则将中断模式设置为REINTERRUPT(因为acquireQueued不会响应中断)
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

接着跟一下各个子方法：

addConditionWaiter

addConditionWaiter用于将当前线程添加进Condition条件队列:

private Node addConditionWaiter() {
    // 条件队列尾节点
    Node t = lastWaiter;
    // 如果尾节点线程已取消，则执行unlinkCancelledWaiters进行取消节点的清除
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    // 将当前线程封装为Node节点，状态为Node.CONDITION
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    // 如果尾节点为null，证明条件队列为空，则将头节点指针指向当前节点
    if (t == null)
        firstWaiter = node;
    // 当前节点作为尾节点的后继节点
    else
        t.nextWaiter = node;
    // 将尾节点指针指向当前节点
    lastWaiter = node;
    return node;
}

fullyRelease

fullyRelease用于释放当前线程持有的锁。

final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        // 获取当前线程持有锁的状态值
        int savedState = getState();
        // 释放锁，并返回savedState
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        // 若最终失败，将当前节点的状态修改为Node.CANCELLED
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

checkInterruptWhileWaiting

checkInterruptWhileWaiting用于检查线程挂起期间的中断状态。

// 检查线程挂起期间的中断状态：
// 1. 若在被signal之前中断，则返回THROW_IE
// 2. 若在被signal之后中断，则返回REINTERRUPT
private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;
}

// 如果线程在被signal之前cancell，则返回true
final boolean transferAfterCancelledWait(Node node) {
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        enq(node);
        return true;
    }
    while (!isOnSyncQueue(node))
        Thread.yield();
    return false;
}

signal

signal方法将会唤醒在条件等待队列中等待时间最长的节点(首节点)，在唤醒节点之前，会将节点移动到同步队列中。

public final void signal() {
    // 如果当前线程不是持有该condition的锁，抛IllegalMonitorStateException异常
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    // 如果条件队列不为空，执行doSignal
    if (first != null)
        doSignal(first);
}

doSignal

private void doSignal(Node first) {
    do {
        // 头节点指针往后移动一个节点
        // 若当前头结点的后继节点为null，则尾节点置为null
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        // 将之前头节点的后继节点置为null，便于GC
        first.nextWaiter = null;
        // 头节点未成功转移到同步队列且头节点不为null，继续自旋
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

transferForSignal用于将条件等待队列节点转移到AQS的FIFO同步队列中。

final boolean transferForSignal(Node node) {
    // 如果CAS失败，说明节点在signal之前被取消了，返回false
    // 设置入队节点的等待状态为0
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;
    // 设置成功则进行入队操作
    Node p = enq(node);
    // 获取入队节点的前继节点的状态
    int ws = p.waitStatus;
    // 1. 如果前继节点取消，则直接唤醒当前节点线程
    // 2. 如果为非取消节点则将前继节点设置为SIGNAL，并不马上唤醒线程
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

查询方法

isOwnedBy

// 判断当前Condition是否是由指定的AQS创建的
final boolean isOwnedBy(AbstractQueuedSynchronizer sync) {
    return sync == AbstractQueuedSynchronizer.this;
}

hasWaiters

// 查询该Condition对象上是否有等待的线程
protected final boolean hasWaiters() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
        if (w.waitStatus == Node.CONDITION)
            return true;
    }
    return false;
}

getWaitQueueLength

// 获取条件等待队列的大约长度
// 因为没有在查询的时候加锁，并不能保证在查找的时候是否有新的节点加入
protected final int getWaitQueueLength() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    int n = 0;
    for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
        if (w.waitStatus == Node.CONDITION)
            ++n;
    }
    return n;
}

getWaitingThreads

// 获取等待在Condition上的线程集合
protected final Collection<Thread> getWaitingThreads() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    ArrayList<Thread> list = new ArrayList<Thread>();
    for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
        if (w.waitStatus == Node.CONDITION) {
            Thread t = w.thread;
            if (t != null)
                list.add(t);
        }
    }
    return list;
}
}