jdk concurrent condition

Condition

结构体

public class ConditionObject implements Condition, java.io.Serializable {
    private transient Node firstWaiter;
    private transient Node lastWaiter;
}

await

public class ConditionObject implements Condition, java.io.Serializable {
    public final void await() throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        // 尝试将自己放到队尾，并返回节点
        Node node = addConditionWaiter();
        // 这个写法和 synchronized 中的 await 方法很像
        // 就是释放所有请求中的信号量
        int savedState = fullyRelease(node);
        int interruptMode = 0;
        while (!isOnSyncQueue(node)) {
            // 只要不在队列中就陷入等待，被唤醒之后等待进入队列
            LockSupport.park(this);
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                // 被中断了就 break
                break;
        }
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }

    private Node addConditionWaiter() {
        // 调用 await 的前提是 lock 成功，也就是 tryAcquire 成功
        if (!isHeldExclusively()) {
            // 判定自己是否获取了锁，没有就会抛异常
            throw new IllegalMonitorStateException();
        }
        Node t = lastWaiter;
        // If lastWaiter is cancelled, clean out.
        if (t != null && t.waitStatus != Node.CONDITION) {
            // 移除所有状态不是 CONDITION 的节点
            unlinkCancelledWaiters();
            t = lastWaiter;
        }

        Node node = new Node(Node.CONDITION);
        // 最后一个节点都是 null，说明等待队列是空
        if (t == null)
            // 自己作为 first 节点
            firstWaiter = node;
        else
            // 自己是最后一个节点
            t.nextWaiter = node;
        lastWaiter = node;
        return node;
    }

    private void unlinkCancelledWaiters() {
        Node t = firstWaiter;
        Node trail = null;
        // 从 first 开始遍历，移除所有状态不是 CONDITION 的节点
        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;
        }
    }

    final int fullyRelease(Node node) {
        try {
            // 每次 acquire 都会让 state 增加
            int savedState = getState();
            // 这个写法就是一次性释放所有信号量
            if (release(savedState))
                return savedState;
            throw new IllegalMonitorStateException();
        } catch (Throwable t) {
            node.waitStatus = Node.CANCELLED;
            throw t;
        }
    }

    final boolean isOnSyncQueue(Node node) {
        // 需要注意的是，node 此时是 last 节点
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            // 自己的前一个节点为空，说明自己就是第一个等待者，所以自然不在队列中
            return false;
        if (node.next != null) // If has successor, it must be on queue
            // 只要自己的后继者存在，说明自己也在队列中
            return true;
        // 由于存在多线程竞争，所以以从后往前的遍历方式作为基准
        return findNodeFromTail(node);
    }

    private boolean findNodeFromTail(Node node) {
        // 从 tail 往前遍历，直到触达 node 节点，或者为空
        for (Node p = tail; ; ) {
            // 从后往前遍历竟然找到了自己，说明自己在队列中
            if (p == node)
                return true;
            // 到头了,但没有找到 node 说明自己肯定不在队列中
            if (p == null)
                return false;
            p = p.prev;
        }
    }
}