[JDK源码]J.U.C-AQS.ConditionObject

Posted 2021-12-30 L._l

ConditionObject内部类

条件变量 ConditionObject 实现了 Condition 接口

变量构造器定义

public class ConditionObject implements Condition, java.io.Serializable 
    //第一个等待节点
    private transient Node firstWaiter;
    //最后一个等待节点
    private transient Node lastWaiter;
    // 重复中断状态位
    private static final int REINTERRUPT =  1;
    //发生异常状态位
    private static final int THROW_IE    = -1;
    
    public ConditionObject()  

await 等待操作

当节点被添加到等待队列后，需要等待条件成立，await相关方法。

//根据外部中断
public final void await() throws InterruptedException 
    if (Thread.interrupted())//是否线程中断
        throw new InterruptedException();//抛出.,
    Node node = addConditionWaiter();//生成一个Node等待节点将其放入条件阻塞队列
    int savedState = fullyRelease(node);//fullyRelease 调用release，释放AQS竞争队列中当前节点后面的等待锁的节点
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) //节点未放入到AQS的竞争队列之前一直阻塞
        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);

//保证线程不可中断的等待
public final void awaitUninterruptibly() 
    Node node = addConditionWaiter();
    //将线程节点从AQS竞争队列中 释放
    int savedState = fullyRelease(node);
    boolean interrupted = false;
    //节点在等待队列， 让线程阻塞 往里放
    while (!isOnSyncQueue(node)) 
        LockSupport.park(this);
        if (Thread.interrupted())
            interrupted = true;
    
    if (acquireQueued(node, savedState) || interrupted)
        selfInterrupt();

下面这三个 和await差不多 多了 时间 ，时间类型

public final long awaitNanos(long nanosTimeout)//超时等待   整体流程和 aiait一样，只是park方法有等待时长
public final boolean awaitUntil(Date deadline)//park 方法换成了 parkUntil 
public final boolean await(long time, TimeUnit unit)

addConditionWaiter原理

//添加等待节点
private Node addConditionWaiter() 
    Node t = lastWaiter;
    // 判断最后一个等待节点是否位空，状态CONDITION 不是unlinkCancelledWaiters方法
    if (t != null && t.waitStatus != Node.CONDITION) 
        unlinkCancelledWaiters();
        t = lastWaiter;
    
    //创建新的节点  插入到等待队列	标准链表操作
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;

fullyRelease原理

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)// 如果失败，那么将节点状态变为CANCELLED即可
            node.waitStatus = Node.CANCELLED;
    

isOnSyncQueue原理

final boolean isOnSyncQueue(Node node) 
    //节点状态CONDITION 前节点不为空   说明不在竞争队列上
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    if (node.next != null) //若节点的next节点不为空，那么一定在AQS 竞争队列上
        return true;
    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;
        
    

reportInterruptAfterWait原理

private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException 
    //根据外部状态 执行不同操作
    if (interruptMode == THROW_IE)
        throw new InterruptedException();
    else if (interruptMode == REINTERRUPT)
        selfInterrupt();

checkInterruptWhileWaiting 原理

private int checkInterruptWhileWaiting(Node node) 
    return Thread.interrupted() ? //线程是否发生了中断   ，正常情况下返回0
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;

unlinkCancelledWaiters原理

//从条件队列中断开已取消的侍者节点。
private void unlinkCancelledWaiters() 
    Node t = firstWaiter;
    Node trail = null;
    while (t != null) 
        Node next = t.nextWaiter;
        if (t.waitStatus != Node.CONDITION) //状态CONDITION 的不要
            t.nextWaiter = null;
            if (trail == null)
                firstWaiter = next;
            else
                trail.nextWaiter = next;
            if (next == null)
                lastWaiter = trail;
        
        else
            trail = t;
        t = next;
    

就是说先将节点放进等待队列，然后释放竞争队列的，等条件达到要求、中断或者超时，线程等待完成后 重新获取锁

signal 唤醒操作

//唤醒操作  调用 doSignal方法
public final void signal() 
    if (!isHeldExclusively())//保证当前线程持有锁的状态下执行
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);

//唤醒一个等待节点 调用transferForSignal 方法
private void doSignal(Node first) 
    do //设置下一个等待节点
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
     while (!transferForSignal(first) &&
             (first = firstWaiter) != null);//如果放到 竞争队列失败 继续唤醒下一个节点

singall

//唤醒所有 调用 doSignalAll方法
public final void signalAll() 
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignalAll(first);


private void doSignalAll(Node first) 
    lastWaiter = firstWaiter = null;
    do //循环遍历 transferForSignal 方法唤醒节点
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);
        first = next;
     while (first != null);

transferForSignal原理

final boolean transferForSignal(Node node) 
	//CAS 将线程节点状态修改为0 
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;
	//enq 方法 将节点添加到AQS竞争队列
    Node p = enq(node);
    int ws = p.waitStatus;
    //如果前驱节点的状态 >0   CAS设置前驱节点状态SIGNAL 失败  unpark方法唤醒当前线程
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;

transferForSignal方法将节点转移到AQS的队列中，实际是调用了enq 方法。