技术博客

Java AQS 实现——排他模式

发表于 分类于 ， 阅读次数：
本文字数： 12k 阅读时长 ≈ 11 分钟

引言

本文着重介绍 AQS 的排他模式的实现方式。所有关于 Java 并发的文章均收录于<Java并发系列文章>。

排他模式

获取资源

入队操作介绍完之后，我们来看一下什么情况下需要执行入队操作，我们先从排他模式说起。下面的 acquire 是 AQS 提供的一个以排他模式获取资源的函数，我们可以看到它的执行流程是：

  1. 先尝试获取资源 tryAcquire，tryAcquire 是一个抽象函数，看完前面的锁的分类部分大家应该对它比较熟悉，因为通过 AQS 实现的各类锁实际上就是通过对 tryAcquire 这类抽象函数的覆写来达到各种锁的效果的。
  2. 如果尝试加锁失败，也就是说当前该资源已经被加锁了，就通过 addWaiter 将当前线程添加到同步队列中，注意参数是 Node.EXCLUSIVE 意味着排它锁。
  3. 加入到同步队列后，开始执行 acquireQueued 函数，猜一下应该能猜到这里面一定是进行睡眠等待的逻辑
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/**
 * Acquires in exclusive mode, ignoring interrupts.  Implemented
 * by invoking at least once {@link #tryAcquire},
 * returning on success.  Otherwise the thread is queued, possibly
 * repeatedly blocking and unblocking, invoking {@link
 * #tryAcquire} until success.  This method can be used
 * to implement method {@link Lock#lock}.
 *
 * @param arg the acquire argument.  This value is conveyed to
 *        {@link #tryAcquire} but is otherwise uninterpreted and
 *        can represent anything you like.
 */
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

下面的就是 acquireQueued 的代码。

  1. 这里面有一个循环，它会不断地获取当前节点的前序节点，如果前序节点是 head 节点（也就是那个虚拟节点，还记得吗，虚拟节点不保存有效数据，只用作指针），这里 head 节点充当一个标志位的效果，如果一个节点的前序节点是 head，那么该节点就排在队列的第一位。
  2. 如果 p == head ，我们就需要再调用一次 tryAcquire 尝试获取锁
  3. 如果第二步成功的话，当前线程已经获得到锁了，这时候要将 head 指针进行修改，可以看到 setHead 并没有使用到 CAS 指令，因为能执行到 setHead 函数的线程相当于已经获得到同步资源了，不存在竞争
  4. 在 setHead 函数中，对当前 Node 的 thread 和 prev 进行了清除，因为这时候该 Node 已经扮演了虚拟节点的角色，有必要把虚拟节点中用不到的属性进行清除
  5. 如果 CAS 执行失败（这里有很多种可能，比如被中断了，或者是刚入队，又或者是虚假唤醒（后面介绍）），则检查是否需要继续进入睡眠，一般来说如果前序节点的状态成功改为 SIGNAL 之后(但是改成 SIGNAL 之后还会再尝试获取一次锁，失败之后才会睡眠，这是防止死等的重中之重)，就可以进入等待了，SIGNAL 表明当前节点肩负着唤醒下一个节点的责任，除此之外，在检查是否需要睡眠时，如果发现前序节点的请求已经被取消，则删除该节点。
  6. 如果发现自己确实需要睡眠，则会通过 park 函数进入等待状态，因为使用的是 park 所以不需要担心，别的线程先执行唤醒之后，当前线程再进入等待的情况，因为在这种情况下 park 函数会直接返回不会进行等待
  7. 最后从等待状态中恢复过来之后，检查是否是因为中断而唤醒的，是的话，就记录一下，在返回的时候以返回值的形式告诉调用者。可见 acquire 在遇到中断时，不会抛出 InterruptedException 异常，而是循环重试。如果想要达到被中断时立即抛出异常的效果，可以使用 acquireInterruptibly, 它的实现逻辑和 acquire 基本相同，主要的区别就是 park 被中断时，会抛出 InterruptedException
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
85
86
87
88
/**
 * Acquires in exclusive uninterruptible mode for thread already in
 * queue. Used by condition wait methods as well as acquire.
 */
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)
            // 永远执行不到，因为抛出的异常都被 parkAndCheckInterrupt 中的 Thread.interrupted() 吞掉了
            cancelAcquire(node);
    }
}

/**
 * Sets head of queue to be node, thus dequeuing. Called only by
 * acquire methods.  Also nulls out unused fields for sake of GC
 * and to suppress unnecessary signals and traversals.
 */
private void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}

/**
 * Checks and updates status for a node that failed to acquire.
 * Returns true if thread should block. This is the main signal
 * control in all acquire loops.  Requires that pred == node.prev.
 *
 * @param pred node's predecessor holding status
 * @param node the node
 * @return {@code true} if thread should block
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        /*
         * 前序节点的状态已经是 SIGNAL 了，可以进入等待状态
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        return true;
    if (ws > 0) {
        /*
         * 前序节点被取消，移除被取消的节点
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
         * 修改前序节点为 SIGNAL，这时候当前线程还没进入等待状态，我们需要重新判断一下现在自己是不是第一个，是的话就不用等待了
         * 改成 SIGNAL 之后还会再尝试获取一次锁，失败之后才会睡眠，这是防止死等的重中之重，考虑如下情况 ThreadB-tryAcquire->ThreadA-Release->ThreadA-CheckStatus(!=SIGNAL)->ThreadB-ChangeStatus2SIGNAL
         * waitStatus must be 0 or PROPAGATE.  Indicate that we
         * need a signal, but don't park yet.  Caller will need to
         * retry to make sure it cannot acquire before parking.
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

/**
 * Convenience method to park and then check if interrupted
 *
 * @return {@code true} if interrupted
 */
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

作为延伸，我们这里带大家回顾一下 ReentrantLock 中公平锁部分对 tryAcquire 的实现。它对大家理解 AQS 如何在 CAS 修改同步队列的情况下（先修改前序指针->CAS 修改尾结点->修复后续指针），以哪种方式访问队列中的数据能够避开数据不同步的风险。简单地说，通过前序指针(prev)访问队列中的数据肯定是安全的。

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
/**
 * Fair version of tryAcquire.  Don't grant access unless
 * recursive call or no waiters or is first.
 */
protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 在尝试获取公平锁时，先会判断队列中是否存在前序节点。
        if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

public final boolean hasQueuedPredecessors() {
    // The correctness of this depends on head being initialized
    // before tail and on head.next being accurate if the current
    // thread is first in queue.
    Node t = tail; // Read fields in reverse initialization order
    Node h = head;
    Node s;
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}

从上述代码中可以看到，在尝试获取公平锁时，先会判断队列中是否存在前序节点。之所以这么做是因为，只有发生互斥等待时，才会出现入队等待的情况，如果全是共享模式使用资源的话，队列会一直是空的，大家别急我们接下来就介绍共享模式的实现。这里我们先着重看一下 hasQueuedPredecessors 的实现，它是怎么判断有前序节点的呢：

  1. 首先如果 h != t 是说，队列的不为空，因为 head == tail 时，队列中只存在一个虚拟节点不存在实际的等待线程
  2. 在此基础上，我们还要判断一下当前持有锁的线程是不是自己，如果 head.next == null 说明有其他线程刚执行完入队的setTail工作（因为 h != t），但是前序指针还没修复，这种情况下 head.next == null，说明有别的线程已经持有锁了
  3. 另外一种可能就是 head.next != Thread.currentThread() 这时候队首持有锁的线程不是当前线程，所以存在前序节点

既然前面提到了 acquireInterruptibly 这里我们就简单地说一下，就像前面所说它和 acquire 基本相同，确实如此，从下面的代码中可以看到，它就是遇到中断时将 InterruptedException 外抛。

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
/**
 * Acquires in exclusive mode, aborting if interrupted.
 * Implemented by first checking interrupt status, then invoking
 * at least once {@link #tryAcquire}, returning on
 * success.  Otherwise the thread is queued, possibly repeatedly
 * blocking and unblocking, invoking {@link #tryAcquire}
 * until success or the thread is interrupted.  This method can be
 * used to implement method {@link Lock#lockInterruptibly}.
 *
 */
public final void acquireInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (!tryAcquire(arg))
        doAcquireInterruptibly(arg);
}

private void doAcquireInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                // 唯一的区别在这里，这里直接抛出异常，而 acquire 中只是记录一下flag: interrupted = true;
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            // 如果抛出 InterruptedException 异常，则会执行 cancelAcquire
            cancelAcquire(node);
    }
}

当抛出异常时，就会执行到最后面的 cancelAcquire 函数，该函数中负责将前序节点中状态为 CANCELLED 的节点清除，然后再把自己的状态变为 CANCELLED。紧接着是一些清除工作：

  • 如果当前节点是尾结点，则通过 CAS 修改尾结点即可。如果 CAS 失败了也不要紧，因为当前线程状态已经是 CANCELLED 了所以其他线程会把自己清除
  • 如果当前节点不是第一个节点，即 pred != head ，我们要保证前序节点的状态是 SIGNAL，并且前序节点的线程不是当前线程，因为自己不是尾结点，所以自己当前的状态很可能就是 SIGNAL，所以这里我们无论如何要确保前序节点的状态能够修改为 SIGNAL，如果做到了，就可以大胆地通过 CAS 将自己从队列中移除
  • 否则，说明自己可能是头结点，或者前序节点都取消了，也有可能前序节点的线程就是当前线程，那么就只能由自己来唤醒后续的线程了
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
/**
 * Cancels an ongoing attempt to acquire.
 *
 * @param node the node
 */
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 is the apparent node to unsplice. CASes below will
    // fail if not, in which case, we lost race vs another cancel
    // or signal, so no further action is necessary.
    Node predNext = pred.next;

    // Can use unconditional write instead of CAS here.
    // After this atomic step, other Nodes can skip past us.
    // Before, we are free of interference from other threads.
    node.waitStatus = Node.CANCELLED;

    // If we are the tail, remove ourselves.
    if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else {
        // If successor needs signal, try to set pred's next-link
        // so it will get one. Otherwise wake it up to propagate.
        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
    }
}

唤醒的过程也很简单：

  1. 如果当前节点状态不是 CANCELLED 就清除状态
  2. 然后先看一下 next 指针指向的节点是否需要被唤醒，next == null 代表了可能发生的前序指针和后续指针不同步，s.waitStatus > 0 表示后继节点已被取消，这时候我们就需要找到下一个需要被唤醒的节点。
  3. 我们需要从尾结点出发，逐个向前找，因为前序指针肯定是安全的
  4. 如果找到了需要被唤醒的线程，就执行 unpark 唤醒它
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
/**
 * Wakes up node's successor, if one exists.
 *
 * @param node the node
 */
private void unparkSuccessor(Node node) {
    /*
     * If status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling.  It is OK if this
     * fails or if status is changed by waiting thread.
     */
    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.
     */
    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)
        LockSupport.unpark(s.thread);
}

释放资源

介绍完资源的获取，我们再来看看资源的释放流程。

  1. 首先，调用 tryRelease 函数，它也是一个抽象函数，在上层实现中一般会进行必要的检查，比如检查持有锁的线程是否是当前线程等，就比如如下 ReentrantLock 中对 tryRelease 的实现。
  2. 释放成功后，判断当前同步队列是否为空，不为空并且当前线程承担唤醒职责时（waitStatus < 0）,因为当前线程能够成功执行 tryRelease，所以当前线程的 waitStatus 不会是 CANCELLED，剩下的状态 SIGNAL 是需要承担唤醒职责的，CONDITION 和 PROPAGATE 我们后面介绍。
  3. 具体的唤醒函数 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
27
28
29
30
31
32
/**
 * Releases in exclusive mode.  Implemented by unblocking one or
 * more threads if {@link #tryRelease} returns true.
 * This method can be used to implement method {@link Lock#unlock}.
 *
 * @param arg the release argument.  This value is conveyed to
 *        {@link #tryRelease} but is otherwise uninterpreted and
 *        can represent anything you like.
 * @return the value returned from {@link #tryRelease}
 */
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

AQS 框架使用

看到这里，不知道大家对 AQS 的使用流程有没有体会，我的理解是在使用它时，一般我们需要覆写 tryAcquire 和 tryRelease 这类函数，它们是直接修改互斥资源 state 的函数，AQS 通过将具体的状态修改职责移交到子类中，能让子类实现各种类型的锁，就比如前一章我们介绍的那些。

而子类向外暴露 lock 和 unlock 函数时，又直接使用 AQS 中的 acquire 和 releas 函数，因为这些函数中封装了尝试加锁过程和加锁失败入队等待过程。

1
2
public void lock()        { acquire(1); }
public void unlock()      { release(1); }

总结一下就是 AQS 对变化报开放的态度，你可以通过它完成各种同步策略，同时对与那些样板代码，都已经被它封装在了自己内部，并使用 final 关键字修饰，例如 acquire 和 release。

参考内容

[1] linux 2.6 互斥锁的实现-源码分析
[2] 深入解析条件变量(condition variables)
[3] Linux下Condition Vairable和Mutext合用的小细节
[4] 从ReentrantLock的实现看AQS的原理及应用
[5] 不可不说的Java“锁”事
[6] 从源码层面解析yield、sleep、wait、park
[7] LockSupport中的park与unpark原理
[8] Thread.sleep、Object.wait、LockSupport.park 区别
[9] 从AQS到futex-二-JVM的Thread和Parker
[10] Java的LockSupport.park()实现分析
[11] JVM源码分析之Object.wait/notify实现
[12] Java线程源码解析之interrupt
[13] Thread.interrupt()相关源码分析
[14] Java CAS 原理剖析
[15] 源码解析 Java 的 compareAndSwapObject 到底比较的是什么
[16] 《Java并发编程的艺术》
[17] 《实战 Java 高并发程序设计》
[18] volatile关键字深入学习
[19] 为什么Netty的FastThreadLocal速度快
[20] 线程池ThreadPoolExecutor实现原理
[21] 深入理解Java线程池：ThreadPoolExecutor
[22] ConcurrentHashMap 详解一
[23] ConcurrentHashMap 详解二
[24] JUC中Atomic class之lazySet的一点疑惑
[25] The JSR-133 Cookbook for Compiler Writers
[26] 就是要你懂Java中volatile关键字实现原理

贝克街的流浪猫 wechat
您的打赏将鼓励我继续分享!
  • 本文作者： 贝克街的流浪猫
  • 本文链接： https://www.beikejiedeliulangmao.top/java/concurrent/aqs-exclusive-mode/
  • 版权声明： 本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议，转载请注明出处！
  • 创作声明： 本文基于上述所有参考内容进行创作，其中可能涉及复制、修改或者转换，图片均来自网络，如有侵权请联系我，我会第一时间进行删除。