AQS源码笔记

吹梦到西洲
多线程与并发

发布于：Jul 10, 2020
次浏览

AQS源码笔记

AQS全称是 AbstractQueuedSynchronizer，顾名思义，是一个用来构建锁和同步器的框架，它底层用了 CAS来保证操作的原子性，同时利用 FIFO队列实现线程间的锁竞争，将基础的同步相关抽象细节放在 AQS，这也是 ReentrantLock、CountDownLatch以及其他众多同步工具实现同步的底层实现机制。

AQS是抽象类，并不能直接实例化，当需要使用AQS的时候需要继承AQS抽象类并且重写指定的方法，这些重写方法包括线程获取资源和释放资源的方式(如ReentractLock通过分别重写线程获取和释放资源的方式实现了公平锁和非公平锁)，同时子类还需要负责共享变量state的维护，如当state = 0时表示该锁没有被占，大于 $0$ 时候代表该锁被一个或多个线程占领(重入锁)，而队列的维护(获取资源失败入队、线程唤醒、线程的状态等)不需要我们考虑，AQS已经帮我们实现好了。AQS的这种设计模式采用的正是模板方法模式。

继承了AQS的子类的主要任务包括：

  • 通过CAS操作维护共享变量state
  • 重写资源的获取方式tryAquire()
  • 重写资源的释放方式tryRelease()

AQS作为J.U.C的工具类，面向的是需要定制锁的创造者，也就是我们可以基于AQS去按需创造自己需要的锁。而如ReentrantLock这样的锁面向的则是锁的使用者。

内部类与成员变量

内部类

这是等待队列($Wait \ Queue$)的结点类。这个等待队列是CLH锁队列的一个变种。CLH锁通常用来为自旋锁服务。Doug Lea老先生在源码里的注释：

1
2
3
4
5
* <pre>
*      +------+  prev +-----+       +-----+
* head |      | <---- |     | <---- |     |  tail
*      +------+       +-----+       +-----+
* </pre>

如果要入队的话，从tail进入，出队的话，从head出，注意这入队出队的操作要保证原子性。可以看到，这个队列是由tail指向head，而不是head指向tail。

建议阅读 Doug Lea 老爷子在AQS中的大段注释。

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
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;
    /** 等待状态: 线程正在条件队列 */
    static final int CONDITION = -2;
    /** 等待状态: 指示下一个acquireShared应该无条件传播, 仅在共享模式下可用 */
    static final int PROPAGATE = -3;

    /** 当前结点的状态, SIGNAL, CONDITION, CANCELLED, PROPAGATE 代表是条件队列结点 */
    /** 为 0, 代表当前结点在 sync 队列中, 阻塞着等待排队获取锁. */
    /** 通过 CAS 修改, 如果条件允许的话也可以通过 volatile写 进行修改 */
    volatile int waitStatus;

    /** 
    * 前驱结点, 入队时赋值, 出队时为null(for GC),  
    * 在前驱结点取消后, 会进行短路操作, 
    * 找到一个未取消的结点(这种情况一定存在, 因为head结点不会被cancelled)
    * 只有在成功 acquire 的时候一个结点才能成为头结点
    * 一个 cancelled 的线程永远不会成功 acquire
    * 而且一个线程只能被自己取消, 
    */
    volatile Node prev;

    /**
    * 当前线程/结点释放时链接的后继结点
    * 入队时赋值, 在绕过前面cancelled的结点时进行调整, 在出队时清空.
    */
    volatile Node next;

    /** 这个结点锁包装的线程 构造时初始化, 使用后为 null */
    volatile Thread thread;

    /**
They are then transferred to the queue to
         * re-acquire. And because conditions can only be exclusive,
         * we save a field by using special value to indicate shared
         * mode.
         */
    /**
    * 条件队列 condition 中的后继结点, 或者是特殊值 SHARED
    * 条件队列仅在独占模式下才可访问
    */
    Node nextWaiter;

    /** 判断是否共享模式 */
    final boolean isShared() {
        return nextWaiter == SHARED;
    }

	/** 获取前驱结点 */
    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }

    Node() {}    // Used to establish initial head or SHARED marker

    Node(Thread thread, Node mode) {     // Used by addWaiter
        this.nextWaiter = mode;
        this.thread = thread;
    }

    Node(Thread thread, int waitStatus) { // Used by Condition
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

成员变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
/**
* 等待队列的头结点. 懒初始化. 除了初始化操作外, 它仅可以通过 setHead() 方法操作
* 如果头结点 head 存在的话, 那它的等待状态一定不是 CANCELLED
* 可以认为是当前持有锁的结点
*/
private transient volatile Node head;

/**
* 等待队列的尾结点, 懒初始化.  
* 仅通过 enq() 添加新结点时操作.
* 其余线程竞争锁失败后将会加入队尾, tail 始终指向队列最后一个结点. 
*/
private transient volatile Node tail;

/** 
* 最简单也最重要的变量, 代表当前锁的状态
* 0 : 未被使用
* 大于 0 : 代表被线程持有. 
* getState(), setState() 进行操作
*/
private volatile int state;

小结

从上面可以看到，AQS内部数据结构为：

  • 双向链表：同步队列。队列中的每个结点对应一个Node，AQS通过控制链表结点的插入与删除达到同步与阻塞的目的。
  • 单向链表：条件队列。

可以把同步队列和条件队列理解成储存等待状态的线程的队列，另外：

  • 条件队列中的线程并不能直接去获取资源，而要先从条件队列转到同步队列中排队获取
  • 同步队列的唤醒结果是线程去尝试获取锁
  • 条件队列的唤醒结果是把线程从条件队列移到同步队列中
  • 一个线程要么是在同步队列中，要么是在条件队列中，不可能同时存在这两个队列里面。

线程状态转换图

AQS提供共享模式与独占模式两种模式去获取资源。

当一个线程以共享模式或独占模式去获取资源的时候，如果获取失败则将该线程封装成Node结点(同时将该结点标识为共享模式或独占模式)加入到同步队列的尾部，AQS实时维护着这个同步队列，这个队列以FIFO(先进先出)来管理结点的排队，即资源的转移(获取再释放)的顺序是从头结点开始到尾结点。

独占获取

独占模式即一个线程获取到资源后，其他线程不能再对资源进行任何操作，只能阻塞，在等待队列中等待被唤醒获得资源。

acquire()

1
2
3
4
5
public final void acquire(int arg) {
  if (!tryAcquire(arg) &&
      acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    selfInterrupt();
}

在此过程中：

  1. 通过tryAcquire(arg)尝试获取锁，这个方法需要实现类去自定义获取锁的逻辑，获取成功则持有锁，获取失败则执行加入等待队列的逻辑。
  2. 获取锁失败后，执行addWaiter(Node.EXCLUSIVE)将 当前线程封装成一个Node结点，加入等待队列尾部。
  3. 执行acquireQueued(xx, xx)。该方法用来判断当前结点的前驱结点是否为头结点，尝试获取锁，如果获取成功，则当前结点会成为新的头结点。这也是获取锁的核心逻辑。

tryAcquire()

1
2
3
4
5
6
7
/**
* @param arg 传递给和获取锁的方法的参数, 或者是保存在条件等待队列里的结点的值.
* 需要继承了 AQS 的子类去实现具体的实现逻辑
*/
protected boolean tryAcquire(int arg) {
    throw new UnsupportedOperationException();
}

addWaiter()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
/**
* 为当前按线程及其模式(共享或独占)创建入队结点
* @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
*/
private Node addWaiter(Node mode) {
    // 1. 创建基于当前线程的, EXCLUSIVE 类型的结点
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    // 2. 判断队尾是否为空, 如果不为空则将结点加入队尾.
    if (pred != null) {
        node.prev = pred;
        // 3. 采用CAS插入
        // 即使并发情况, 也只有一条线程能操作成功, 其余的进行 enq 方法
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}

在此过程中：

  1. 创建基于当前线程的, EXCLUSIVE类型的结点。
  2. CAS将结点插入队尾。

enq()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
/**
* 入队操作, 向队尾插入结点. 如果有必要还要进行初始化
* @param node 要插入的结点
* @return 插入结点的前驱
*/
private Node enq(final Node node) {
    for (;;) { // 自选操作
        Node t = tail;
        if (t == null) { // Must initialize
            // CAS设置头, 初始化操作.
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // CAS 插入队尾, 成功才返回; 也就是说失败就自旋直至修改成功为止. 
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

在此过程中：

  1. 自旋机制，这是AQS里很重要的一个机制。
  2. 如果队尾结点为空，则初始化队列，将头结点设置为空结点，头结点表示正持有锁的线程。
  3. 如果队尾tail不为空，则采取CAS操作，将结点插入到队尾，失败则自旋至成功为止。

对比addWaiter()与enq()两个方法可以发现，先在addWaiter()中使用一次CAS进行尝试插入，如果成功皆大欢喜，失败的话则到enq()中进行完整的入队操作，在这个过程中存在自旋。

完整的入队流程：

  • 队列为空：先初始化，将头结点设为空结点，表示当前没有线程持有锁。
  • 队列不为空：将线程结点插入队尾。

acquireQueued()

如果前面的过程都执行成功，说明当前结点已经成功入队。

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
/**
* 对于已经入队的线程, 独占模式, 不可中断. 
*/
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        // 不可中断
        boolean interrupted = false;
        for (;;) {
            // 自旋
            final Node p = node.predecessor(); // 入队结点的前驱
            if (p == head && tryAcquire(arg)) {
                // 如果入队结点的前驱石头结点并且尝试获取锁成功的话
                // 把当前结点设为头结点, 代表它当前已经获取到锁了.
                setHead(node);
                p.next = null; // help GC
                failed = false; // 成功
                return interrupted;
            }
            // 获取锁失败, 进入挂起挂起, 可中断.
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed) // 如果获取失败, 则取消获取锁. 
            cancelAcquire(node);
    }
}

在此过程中：

  1. 判断当前结点的前驱结点是否是头结点head，如果是，则尝试获取锁。
  2. 获取锁失败，挂起。

如果当前结点的前驱是头结点head，这个时候head可能已经释放了锁，所以需要tryAcquire(arg)

shouldParkAfterFailedAcquire()

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
/**
* 检查并更新获获取锁失败的结点的状态
* 如果线程应该阻塞, 则返回 true
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        // 如果 pred 结点为 SIGNAL 状态, 返回true，说明当前结点需要挂起
        return true;
    if (ws > 0) {
        // 前驱被取消, 结点状态为 CANCELLED, 跳过前驱并重试. 
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /**
        * waitStatus 必须为 0 或者 PROPAGATE
        * 这表示我们需要 SIGNAL, 但是暂时不要 park
        * 调用者需要进行充实来确保它在park前真的获取不到锁了.
        */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

在此过程中：

  1. 判断pred的状态，如果为SIGNAL，则直接返回true进行阻塞。
  2. 删除队列中状态为CANCELLED的结点。
  3. 如果pred状态为 $0$ 或者 PROPAGATE，则将其设置为SIGNAL，再从acquireQueued方法自旋操作从新循环。

这里需要注意的时候，结点的初始值为 $0$，因此如果获取锁失败，会尝试将结点设置为 $SIGNAL$。

parkAndCheckInterrupt()

1
2
3
4
private final boolean parkAndCheckInterrupt() {
  LockSupport.park(this);
  return Thread.interrupted();
}

LockSupport是用来创建锁和其他同步类的基本线程阻塞原语。LockSupport提供 park() 和 unpark() 方法实现阻塞线程和解除线程阻塞。release释放锁方法逻辑会调用LockSupport.unpark() 方法来唤醒后继结点。

cancelAcquire()

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
/**
* 取消正在尝试获取锁的操作
*/
private void cancelAcquire(Node node) {
    // Ignore if node doesn't exist
    if (node == null)
        return;

    node.thread = null;

    // Skip cancelled predecessors
    Node pred = node.prev;
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;

    // predNext 是要取消拼接的结点. 如果不是的话, 下面的CAS将会失败.
    Node predNext = pred.next;

    // 可以利用非CAS操作. 在这个原子步骤之后, 其他的结点可以跳过CANCELLED的结点
    // 自此, 当前结点就可以被其他线程干扰了.
    node.waitStatus = Node.CANCELLED;

    // If we are the tail, remove ourselves.
    if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else {
        // 如果后继结点需要被通知, 那就通知它, 否则利用传播的方式唤醒需要被唤醒的结点。
        int ws;
        if (pred != head &&
            ((ws = pred.waitStatus) == Node.SIGNAL ||
             (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
            pred.thread != null) {
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);
        } else {
            unparkSuccessor(node);
        }

        node.next = node; // help GC
    }
}

获取锁的逻辑

img

独占释放

释放锁

1
2
3
4
5
6
7
8
9
10
11
12
/**
* 如果 tryRelease(arg) 返回 true 的话, 可以被一个或多个未被阻塞的线程所实现.
*/
public final boolean release(int arg) {
  if (tryRelease(arg)) { // 尝试释放成功的话, 检查该结点是否有后继结点, 有的话则去唤醒
    Node h = head; // head: 当前持有锁的结点
    if (h != null && h.waitStatus != 0) // 头结点状态不为 0, 代表有后继结点.
      unparkSuccessor(h); // 
    return true;
  }
  return false;
}

在此过程中：

  1. tryRelease尝试释放锁，这个方法需要继承了AQS的子类去实现自己的逻辑。
  2. 释放成功则执行唤醒后继结点的逻辑。

unparkSuccessor()

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
/**
* 如果后继结点存在的话则进行唤醒
*/
private void unparkSuccessor(Node node) {
    // 如果ws < 0 尝试去处理等待通知的结点.
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
    * Thread to unpark is held in successor, which is normally
    * just the next node.  But if cancelled or apparently null,
    * traverse backwards from tail to find the actual
    * non-cancelled successor.
    */
    // 从后向前寻找且 non-cancelled 的后继结点
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null) // 如果结点状态为 non-cancelled, 直接将后继结点唤醒.
        LockSupport.unpark(s.thread);
}

共享获取

共享模式下，线程无论是获取资源还是释放资源，都可能会唤醒后继结点。

共享模式资源的获取和独占模式资源的获取流程差不多，就是在获取资源成功后，会唤醒为共享模式的后继结点，然后被唤醒的后继结点也去获取资源。

image

acquireShared()

1
2
3
4
5
6
public final void acquireShared(int arg) {
    // 尝试获取共享锁, 小于 0 表示获取失败
    if (tryAcquireShared(arg) < 0)
        // 执行获取锁失败的逻辑
        doAcquireShared(arg);
}

tryAcquireShared()

仍然，获取锁的具体逻辑是由继承了AQS的子类去实现的。

1
2
3
protected int tryAcquireShared(int arg) {
    throw new UnsupportedOperationException();
}

doAcquireShared()

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
/**
* 共享模式, 不可中断模式下的获取.
*/
private void doAcquireShared(int arg) {
    // 添加共享锁类型结点到队列中
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) 
            // 获取线程的前驱
            final Node p = node.predecessor();
            if (p == head) {
                // 如果前驱节点为头结点，则该线程尝试获取资源。
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 获取成功 对后继 SHARED 节点持续唤醒
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt(); // Thread.currentThread().interrupt();
                    failed = false;
                    return;
                }
            }
        	// 和独占模式一样
            // 调用 shouldParkAfterFailedAcquire, 将该节点的前驱节点
            // 的状态设置为 SIGNAL，告诉前驱节点我要去“睡觉”了，当资源排
            // 到你的时候，你就通知我一下让我醒来，即节点做进入等待状态的准备。
            // 当节点做好了进入等待状态的准备，则调用 parkAndCheckInterrupt
            // 函数，让该节点进入到等待状态。  
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

setHeadAndPropagate()

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
/**
     * Sets head of queue, and checks if successor may be waiting
     * in shared mode, if so propagating if either propagate > 0 or
     * PROPAGATE status was set.
     *
     * @param node the node
     * @param propagate the return value from a tryAcquireShared
     */
private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    setHead(node);
    /*
         * Try to signal next queued node if:
         *   Propagation was indicated by caller,
         *     or was recorded (as h.waitStatus either before
         *     or after setHead) by a previous operation
         *     (note: this uses sign-check of waitStatus because
         *      PROPAGATE status may transition to SIGNAL.)
         * and
         *   The next node is waiting in shared mode,
         *     or we don't know, because it appears null
         *
         * The conservatism in both of these checks may cause
         * unnecessary wake-ups, but only when there are multiple
         * racing acquires/releases, so most need signals now or soon
         * anyway.
         */
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        // 如果节点为共享节点，则调用doReleaseShared函数唤醒后继节点。
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

共享释放

共享模式下资源释放流程和独占模式下资源释放的流程差不多，就是在释放后唤醒后继为共享模式的节点，且唤醒的动作是传播下去的，直到后继节点出现不是共享模式的，这个唤醒的过程和共享模式的获取资源的唤醒过程一样。

releaseShared()

1
2
3
4
5
6
7
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

tryReleaseShared()

1
2
3
4
5
6
/**
* 仍然, 被子类实现.
*/
protected boolean tryReleaseShared(int arg) {
    throw new UnsupportedOperationException();
}

doReleaseShared()

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
/**
* Release action for shared mode -- signals successor and ensures
* propagation. (Note: For exclusive mode, release just amounts
* to calling unparkSuccessor of head if it needs signal.)
*/
private void doReleaseShared() {
    /*
    * Ensure that a release propagates, even if there are other
    * in-progress acquires/releases.  This proceeds in the usual
    * way of trying to unparkSuccessor of head if it needs
    * signal. But if it does not, status is set to PROPAGATE to
    * ensure that upon release, propagation continues.
    * Additionally, we must loop in case a new node is added
    * while we are doing this. Also, unlike other uses of
    * unparkSuccessor, we need to know if CAS to reset status
    * fails, if so rechecking.
    */
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                // 如果节点标识后继节点需要唤醒，则调用 unparkSuccessor 方法进行唤醒。
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

共享模式下的逻辑

共享模式下的逻辑

条件队列

条件队列又称等待队列，条件队列的实现是通过ConditionObject的内部类完成的，一开始介绍了同步队列条件队列的区别。

可以把同步队和条件队列理解成储存等待状态的线程的队列，条件队列中的线程并不能直接去获取资源，而要先从条件队列转到同步队列中排队获取，一个线程要么是在同步队列中，要么是在条件队列中，不可能同时存在这两个队列里面。

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
/* 
 * 使当前线程进入等待状态，直到以下4种情况任意一个发生：
 * 1.另一个线程调用该对象的signal()，当前线程恰好是被选中的唤醒线程
 * 2.另一个线程调用该对象的signalAll()
 * 3.另一个线程interrupt当前线程（此时会抛出InterruptedException）
 * 4.虚假唤醒（源自操作系统，发生概率低）
 * ConditionObject要求调用时该线程已经拿到了其外部AQS类的排它锁（acquire成功）
 */
void await() throws InterruptedException;
/* 
 * 与await()相同，但是不会被interrupt唤醒
 */
void awaitUninterruptibly();
/* 
 * 与await()相同，增加了超时时间，超过超时时间也会停止等待
 * 三个方法功能相似，其返回值代表剩余的超时时间，或是否超时
 */
long awaitNanos(long nanosTimeout) throws InterruptedException;
boolean await(long time, TimeUnit unit) throws InterruptedException;
boolean awaitUntil(Date deadline) throws InterruptedException;
/* 
 * 唤醒一个正在等待该条件变量对象的线程
 * ConditionObject会选择等待时间最长的线程来唤醒
 * ConditionObject要求调用时该线程已经拿到了其外部AQS类的排它锁（acquire成功）
 */
void signal();
/* 
 * 唤醒所有正在等待该条件变量对象的线程
 * ConditionObject要求调用时该线程已经拿到了其外部AQS类的排它锁（acquire成功）
 */
void signalAll();

可以看到，其作用与Object原生的wait()/notify()/notifyAll()很相似，但是增加了更多的功能。下面以awaitUninterruptibly()、signal()为例，阐述一下其内部实现。

ConditionObject的内部流程图

同步队列和条件队列的关系

  • 线程执行condition.await()方法，将节点从同步队列转移到条件队列中。
  • 线程执行condition.signal()方法，将节点从条件队列中转移到同步队列。

参考

博客内容遵循 署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0) 协议

本文永久链接是：http://raymond-zhao.top/2020/07/10/2020-07-10-Java-AQS/

更新于：Jun 20, 2021

AQS

评论