AQS条件变量

- 日理万妓 2022-10-16 07:23 128阅读 0赞

# ReectrantLock的条件等待功能 #

ReectrantLock 之所以具有条件等待功能，也得全归功于 AQS。

public abstract class AbstractQueuedSynchronizer{ 
        // ConditionObject 执行条件等待功能。
        public class ConditionObject implements Condition, java.io.Serializable { 
    		// 它里面维护了个线程队列。保存那些不满足条件需要休息的线程。
            
            // 队头。
    		private transient Node firstWaiter;
    		// 队尾。
            private transient Node lastWaiter;
    }

# \# 条件等待 #

ReectrantLock lock = new ReectrantLock();
    Condition condition = lock.newCondition();
    condition.await();

Condition condition = lock.newCondition();

// 创建条件变量。
    public Condition newCondition() { 
    // 调用同步器，创建条件变量。
    return sync.newCondition();
    }
    
    final ConditionObject newCondition() { 
        return new ConditionObject();
    }

condition.await();

public final void await() throws InterruptedException { 
        // 如果当前线程被打断过，直接抛异常。
        if (Thread.interrupted())
            throw new InterruptedException();
        // 将当前线程包装成 node，加入条件队列，并返回该节点。
        Node node = addConditionWaiter();
        // 释放当前线程上的所有锁。（因为当前线程要条件等待，它得把锁让出来）
        // 线程可能锁重入了，所以一次性将所有重入的锁都释放。
        // 释放锁的时候，会唤醒等待队列中第二个节点，如果它没有被取消。
        int savedState = fullyRelease(node);
        int interruptMode = 0;
        while (!isOnSyncQueue(node)) { 
            // park 当前线程。不满足条件的线程就停在这里了。-----------
            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);
    }

addConditionWaiter()

private Node addConditionWaiter() { 
        // 获取队尾指针。
        Node t = lastWaiter;
        // If lastWaiter is cancelled, clean out.
        // 清楚 条件队列中已经被取消的节点。
        if (t != null && t.waitStatus != Node.CONDITION) { 
            unlinkCancelledWaiters();
            t = lastWaiter;
        }
        // 将当前线程包装成 Node节点。Node.CONDITION = -2。
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        // 将 node 节点加入 条件队列。
        if (t == null)
            firstWaiter = node;
        else
            t.nextWaiter = node;
        lastWaiter = node;
        // 返回 node 节点。
        return node;
    }

# 唤醒条件等待 #

调用是这样的的，T1 如果处于条件等待，必须由 T2 线程将 T1 唤醒。T1此时是处于条件变量的队列中，被唤醒后，

1.  它要做的事情就是从条件变量队列中离开，
2.  加入到等待队列中去。
3.  等待被唤醒（可能会立即被唤醒），再去重新竞争锁，拿到就执行代码，拿不到就追加到等待队列中 park。

T2 条件唤醒 T1 是执行 `condition.signal()`

public final void signal() { 
        // 如果当前线程不是锁的持有者，抛异常。
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        // 拿出条件队列的头指针
        Node first = firstWaiter;
        // 头指针不为空，说明条件队列中有节点。
        if (first != null)
            // 执行唤醒
            doSignal(first);
    }

private void doSignal(Node first) { 
        do { 
            // firstWaiter 指向 first 的下一个节点，如果下一个节点为空，说明后面没了。
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            first.nextWaiter = null;
            // transferForSignal 将 条件队列 中的 node 节点移动到 等待队列 中。
            // 如果 transferForSignal 成功，则整个循环结束。
            // 失败，如果有下一个节点，那就 transferForSignal 下一个节点。
        } while (!transferForSignal(first) &&
                 (first = firstWaiter) != null);
    }

final boolean transferForSignal(Node node) { 
            // CAS 将 node 的 CONDITION 从 -2 变成 0。
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                // 失败，则返回 false。
                return false;
            
            // 将 node 追加到 等待队列中。
            // 返回 node 的前驱节点。
            Node p = enq(node);
            // 获取前驱节点的 waitStatus。
            int ws = p.waitStatus;
            // 计划是让前驱节点唤醒 node，如果 ws > 0, 前驱节点被取消了。
            // 或者是 前驱的 waitStatus 不能被标记为 -1 ，那前驱就没法唤醒 node 了。
            if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
                // 如果前驱没法唤醒 node，那就在这里将 node 唤醒。
                LockSupport.unpark(node.thread);
            // 返回 true。
            return true;
        }

那被唤醒的 T1 从哪里开始执行呢？从哪里跌倒，从那里爬起来。

public final void await() throws InterruptedException { 
        // 如果当前线程被打断过，直接抛异常。
        if (Thread.interrupted())
            throw new InterruptedException();
        // 将当前线程包装成 node，加入条件队列，并返回该节点。
        Node node = addConditionWaiter();
        // 释放当前线程上的所有锁。（因为当前线程要条件等待，它得把锁让出来）
        int savedState = fullyRelease(node);
        int interruptMode = 0;
        while (!isOnSyncQueue(node)) { 
            // park 当前线程。
            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);
    }