1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
private transient volatile Node head;

private transient volatile Node tail;

private volatile int state;

static final class Node {
    volatile int waitStatus;

    volatile Node prev;

    volatile Node next;
}

private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long stateOffset;
private static final long headOffset;
private static final long tailOffset;
private static final long waitStatusOffset;
private static final long nextOffset;

static {
    try {
        stateOffset = unsafe.objectFieldOffset
            (AbstractQueuedSynchronizer.class.getDeclaredField("state"));
        headOffset = unsafe.objectFieldOffset
            (AbstractQueuedSynchronizer.class.getDeclaredField("head"));
        tailOffset = unsafe.objectFieldOffset
            (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
        waitStatusOffset = unsafe.objectFieldOffset
            (Node.class.getDeclaredField("waitStatus"));
        nextOffset = unsafe.objectFieldOffset
            (Node.class.getDeclaredField("next"));

    } catch (Exception ex) { throw new Error(ex); }
}

1
2
while (!condition)
    wait();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
    implements java.io.Serializable {

    // head节点prev和thread均为null，是无效节点
    private transient volatile Node head;

    // tail节点next为null，thread不为null，是有效节点
    private transient volatile Node tail;

    // 状态，子类自定义。如ReentrantLock是重载锁的个数，ReentrantReadWriteLock则读和写通过mask共用state。
    private volatile int state;

    // 独占锁获取锁API
    public final void acquire(int arg);

    // 独占锁释放锁API
    public final boolean release(int arg);

    // 共享锁获取锁API
    public final void acquireShared(int arg);

    // 共享锁释放锁API
    public final boolean releaseShared(int arg);

    // 模板模式，子类实现
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

    // 模板模式，子类实现
    protected boolean tryRelease(int arg) {
        throw new UnsupportedOperationException();
    }

    static final class Node {

        static final Node SHARED = new Node();

        static final Node EXCLUSIVE = null;

        // 节点状态被取消
        static final int CANCELLED =  1;
        // 后继节点状态需要被唤醒
        static final int SIGNAL    = -1;
        // 节点状态在ConditionObject的队列
        static final int CONDITION = -2;
        // 节点状态在共享锁传播状态，老版本没有这个状态，是为了解决releaseShared并发问题而引入
        static final int PROPAGATE = -3;
        // 节点状态
        volatile int waitStatus;
        // AQS上一节点，ConditionObject无作用null
        volatile Node prev;
        // AQS下一节点，ConditionObject无作用null
        volatile Node next;
        // 每个节点对应一个线程
        volatile Thread thread;
        // AQS用来区分独占锁和共享锁，ConditionObject是单向链表的下一个node
        Node nextWaiter;
    }

    public class ConditionObject implements Condition, java.io.Serializable {
        // 单向链表head
        private transient Node firstWaiter;
        // 单向链表tail
        private transient Node lastWaiter;

        public final void await();

        public final void signal();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
// AQS获取共享锁入口
public final void acquireShared(int arg) {
    // 子类实现tryAcquireShared获取锁的逻辑，失败则进入AQS双向链表
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}
// AQS释放共享锁入口
public final boolean releaseShared(int arg) {
    // 子类实现tryReleaseShared释放锁的逻辑，成功则unpark AQS双向链表
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

private void doAcquireShared(int arg) {
    // 新增共享状态节点
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            // 前驱节点是否是head
            if (p == head) {
                // 尝试获取锁
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 断开旧head，把当前节点变成head并传播unpark
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            // 节点park前置条件是前驱节点waitStatus是SIGNAL
            // 前驱节点waitStatus是0则变为SIGNAL，下一个循环准备park
            // 前驱节点waitStatus是SIGNAL则park
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head;
        // 临界竞争点
        setHead(node);
        // propagate有可能为0，需要依赖waitStatus=PROPAGATE=-3传播unpark
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }
}

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            // 正常流程
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                unparkSuccessor(h);
            }
            // 并发调用doReleaseShared的异常流程
            else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;
        }
        if (h == head)
            break;
    }
}