JUC 中的 AQS 底层详细超详解

作者：华为云开发者联盟

2022-10-25
中国香港
本文字数：8328 字
阅读完需：约 27 分钟

本文分享自华为云社区《JUC中的AQS底层详细超详解，剖析AQS设计中所需要考虑的各种问题！》，作者： breakDawn 。

java 中 AQS 究竟是做什么的？

当你使用 java 实现一个线程同步的对象时，一定会包含一个问题：

你该如何保证多个线程访问该对象时，正确地进行阻塞等待，正确地被唤醒？

关于这个问题，java 的设计者认为应该是一套通用的机制

因此将一套线程阻塞等待以及被唤醒时锁分配的机制称之为 AQS

全称 AbstractQuenedSynchronizer

中文名即抽象的队列式同步器。

基于 AQS，实现了例如 ReentenLock 之类的经典 JUC 类。

AQS 简要步骤

线程访问资源，如果资源足够，则把线程封装成一个 Node，设置为活跃线程进入 CLH 队列，并扣去资源
资源不足，则变成等待线程 Node，也进入 CLH 队列
CLH 是一个双向链式队列, head 节点是实际占用锁的线程，后面的节点则都是等待线程所对应对应的节点

AQS 的资源 state

state 定义

AQS 中的资源是一个 int 值，而且是 volatile 的，并提供了 3 个方法给子类使用：

private volatile int state;protected final int getState() {    return state;}protected final void setState(int newState) {    state = newState;}
// cas方法compareAndSetState(int oldState, int newState);

复制代码

如果 state 上限只有 1，那么就是独占模式 Exclusive，例如 ReentrantLock

如果 state 上限大于 1，那就是共享模式 Share，例如 Semaphore、CountDownLatch、ReadWriteLock，CyclicBarrier

已经有 CAS 方法了，为什么资源 state 还要定义成 volatile 的？

对外暴露的 getter/setter 方法，是走不了 CAS 的。而且 setter/getter 没有被 synchronized 修饰。所以必须要 volatile，保证可见性

这样基于 AQS 的实现可以直接通过 getter/setter 操作 state 变量，并且保证可见性，也避免重排序带来的影响。比如 CountDownLatch，ReentrantReadWriteLock，Semaphore 都有体现（各种 getState、setState）

对资源的操作什么时候用 CAS，什么使用 setState？

volatile 的 state 成员有一个问题，就是如果是复合操作的话不能保证复合操作的原子性

因此涉及 state 增减的情况，采用 CAS

如果是 state 设置成某个固定值，则使用 setState

AQS 的 CLH 队列

为什么需要一个 CLH 队列

这个队列的目的是为了公平锁的实现

即为了保证先到先得，要求每个线程封装后的 Node 按顺序拼接起来。

CLH 本质？是一个 Queue 容器吗

不是的，本质上是一个链表式的队列

因此核心在于链表节点 Node 的定义

除了比较容易想到的 prev 和 next 指针外

还包含了该节点内的线程

以及 waitStatus 等待状态

4 种等待状态如下：

CANCELLED（1）：因为超时或者中断，节点会被设置为取消状态，被取消的节点时不会参与到竞争中的，他会一直保持取消状态不会转变为其他状态；
SIGNAL（-1）：后继节点的线程处于等待状态，而当前节点的线程如果释放了同步状态或者被取消，将会通知后继节点，使后继节点的线程得以运行
CONDITION（-2）：点在等待队列中，节点线程等待在 Condition 上，当其他线程对 Condition 调用了 signal()后，改节点将会从等待队列中转移到同步队列中，加入到同步状态的获取中
PROPAGATE（-3）：表示下一次共享式同步状态获取将会无条件地传播下去
INIT（ 0）:

入队是怎么保证安全的？

入队过程可能引发冲突

因此会用 CAS 保障入队安全。

    private Node enq(final Node node) {        //多次尝试，直到成功为止        for (;;) {            Node t = tail;            //tail不存在，设置为首节点            if (t == null) {                if (compareAndSetHead(new Node()))                    tail = head;            } else {                //设置为尾节点                node.prev = t;                if (compareAndSetTail(t, node)) {                    t.next = node;                    return t;                }            }        }    }

复制代码

出队过程会发生什么？

一旦有节点出队，说明有线程释放资源了，队头的等待线程可以开始尝试获取了。

于是首节点的线程释放同步状态后，将会唤醒它的后继节点（next）

而后继节点将会在获取同步状态成功时将自己设置为首节点

**注意在这个过程是不需要使用 CAS 来保证的，因为只有一个线程能够成功获取到同步状态 **

AQS 详细资源获取流程

1. tryAcquire 尝试获取资源

AQS 使用的设计模式是模板方法模式。

具体代码如下：

public final void acquire(int arg) {    if (!tryAcquire(arg) &&        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))        // 发现中断过，则触发中断异常        selfInterrupt();}

复制代码

即 AQS 抽象基类 AbstractQueuedSynchronizer 给外部调用时，都是调的 acquire(int arg)方法。这个方法的内容是写死的。而 acquire 中，需要调用 tryAcquire(arg)，这个方法是需要子类实现的，作用是判断资源是否足够获取 arg 个

（下面部分代码注释选自： (2条消息) AQS子类的tryAcquire和tryRelease的实现_Mutou_ren的博客-CSDN博客_aqs tryacquire ）

ReentrantLock 中的 tryAcquire 实现

这里暂时只谈论一种容易理解的 tryAcuire 实现，其他附加特性的 tryAcquire 先不提。

里面主要就做这几件事：

获取当前锁的资源数
资源数为 0，说明可以抢，确认是前置节点是头节点，进行 CAS 试图争抢，抢成功就返回 true，并设置当前线程
没抢成功，返回 false
如果是重入的，则直接 set 设置增加后的状态值，状态值此时不一定为 0 和 1 了

protected final boolean tryAcquire(int acquires){ final Thread current = Thread.currentThread();            int c = getState();            // state==0代表当前没有锁，可以进行获取            if (c == 0) {            	// 非公平才有的判断，会判断是否还有前驱节点，直接自己为头节点了或者同步队列空了才会继续后面的锁的获取操作                if (!hasQueuedPredecessors()                      //CAS设置state为acquires，成功后标记exclusiveOwnerThread为当前线程                    && 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");                // 因为是重入操作，可以直接进行state的增加，所以不需要CAS                setState(nextc);                return true;            }            return false;}

复制代码

2.addWaiter 添加到等待队列

当获取资源失败，会进行 addWaiter(Node.EXCLUSIVE)， arg)。

目的是创建一个等待节点 Node，并添加到等待队列

    private Node addWaiter(Node mode) {        Node node = new Node(Thread.currentThread(), mode);        // Try the fast path of enq; backup to full enq on failure        Node pred = tail;        if (pred != null) {            node.prev = pred;            // 通过CAS竞争队尾            if (compareAndSetTail(pred, node)) {                pred.next = node;                return node;            }        }        // 竞争队尾失败，于是进行CAS频繁循环竞争队尾        enq(node);        return node;    }
    private Node enq(final Node node) {        for (;;) {            Node t = tail;            if (t == null) { // Must initialize                if (compareAndSetHead(new Node()))                    tail = head;            } else {                node.prev = t;                if (compareAndSetTail(t, node)) {                    t.next = node;                    return t;                }            }        }    }

复制代码

3. acquireQueued 循环阻塞-竞争

，并在 "处于头节点时尝试获取资源->睡眠->唤醒“中循环。

当已经跑完任务的线程释放资源时，会唤醒之前阻塞的线程。

当被唤醒后，就会检查自己是不是头节点，如果不是，且认为可以阻塞，那就继续睡觉去了

（下面代码注释部分选自 AQS（acquireQueued(Node, int) 3）–队列同步器 - 小窝蜗 - 博客园 (cnblogs.com) ）

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);                // 说明前继节点已经释放掉资源了，将其next置空，好让虚拟机提前回收掉前继节点                p.next = null; // help GC                // 获取资源成功，修改标记位                failed = false;                // 返回中断标记                return interrupted;            }            // 若前继节点不是头结点，或者获取资源失败，            // 则需要判断是否需要阻塞该节点持有的线程            // 若可以阻塞，则继续执行parkAndCheckInterrupt()函数，            // 将该线程阻塞直至被唤醒            // 唤醒后会检查是否已经被中断，若返回true，则将interrupted标志置于true            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())                interrupted = true;        }    } finally {        // 最终获取资源失败，则当前节点放弃获取资源        if (failed)            cancelAcquire(node);
    } }

复制代码

4.shouldParkAfterFailedAcquire 检查是否可以阻塞

该方法不会直接阻塞线程，因为一旦线程挂起，后续就只能通过唤醒机制，中间还发生了内核态用户态切换，消耗很大。

因此会先不断确认前继节点的实际状态，在只能阻塞的情况下才会去阻塞。

并且会过滤掉 cancel 的线程节点

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {    // 获取前继节点的等待状态    int ws = pred.waitStatus;    // 如果等待状态为Node.SIGNAL(-1)，则直接返回true即可以阻塞    // 因为这说明前继节点完成资源的释放或者中断后，会主动唤醒后继节点的（这也即是signal信号的含义），因此方法外面不用再反复CAS了，直接阻塞吧    if (ws == Node.SIGNAL) return true;
    // 如果前继节点的等待值大于0即CANCELLED（1）,说明前继节点的线程发生过cancel动作    // 那就继续往前遍历，直到当前节点的前继节点的状态不为cancel    if (ws > 0) {        do {            node.prev = pred = pred.prev;        } while (pred.waitStatus > 0);        pred.next = node;    } else {        // 前继节点的等待状态不为SIGNAL(-1)，也不为Cancel(1)        // 那么只能是PROPAGATE(-3)或者CONDITION(-2)或者INITIAL（0）        // 直接设置成SIGNAL，下一次还没CAS成功，就直接睡觉了        // 因此在前面所有节点没辩护的情况下， 最多一次之后就会返回true让外面阻塞       compareAndSetWaitStatus(pred, ws, Node.SIGNAL);    }    return false;}

复制代码

5.parkAndCheckInterrupt() 阻塞线程

使用 LockSupport.park 来阻塞当前这个对象所在的线程

private final boolean parkAndCheckInterrupt() {             LockSupport.park(this);    		 // 确认是否是中断导致的park结束，并清除中断标记             return Thread.interrupted();}
public static void park(Object blocker) {    Thread t = Thread.currentThread();    setBlocker(t, blocker);    UNSAFE.park(false, 0L);    setBlocker(t, null);}

复制代码

lockSupport.park()和普通的 wait|notify 都有啥区别？

面向的主体不一样。LockSuport 主要是针对 Thread 进进行阻塞处理，可以指定阻塞队列的目标对象，每次可以指定具体的线程唤醒。Object.wait()是以对象为纬度，阻塞当前的线程和唤醒单个(随机)或者所有线程。
实现机制不同。虽然 LockSuport 可以指定 monitor 的 object 对象，但和 object.wait()，两者的阻塞队列并不交叉。可以看下测试例子。object.notifyAll()不能唤醒 LockSupport 的阻塞 Thread.

如果还要深挖底层实现原理，可以详细见该链接简而言之，是用 mutex 和 condition 保护了一个_counter 的变量，当 park 时，这个变量置为了 0，当 unpark 时，这个变量置为 1。底层用的 C 语言的 pthread_mutex_unlock、pthread_cond_wait 、pthread_cond_signal ，但是针对了 mutex 和_cond 两个变量进行加锁。

6.总体流程图

代码中频繁出现的 interruptd 中断标记是做什么用的？

对线程调用 t1.interrupt();时

会导致 LockSupport.park() 阻塞的线程重新被唤醒

即有两种唤醒情况：被前置节点唤醒，或者被外部中断唤醒

这时候要根据调用的 acuire 类型决定是否在中断发生时结束锁的获取。

上面介绍的是不可中断锁。

在 parkAndCheckInterrupt 中，当 park 结束阻塞时时,使用的是 Thread.interrupted() 而不是 .isInterrupted() 来返回中断状态

因为前者会返回线程当前的中断标记状态同时清除中断标志位（置为 false）

外层 CAS 循环时，就不会让线程受中断标记影响，只是记录一下是否发生过中断

当获取锁成功后，如果发现有过线程中断，则会触发中断异常，

之后便由获取锁的调用者自己决定是否要处理线程中断。像下面这样：

reentrantLock.lock();try {    System.out.println("t1");    TimeUnit.SECONDS.sleep(30);} catch (InterruptedException e) {    e.printStackTrace();} finally {    reentrantLock.unlock();}

复制代码

那么另一种情况就是可中断锁了。

ReentranLock 有一个 lockInterruptibly()方法就是这种情况

线程被唤醒时，如果发现自己被中断过，就会直接抛异常而不是继续获取锁

因此如果你的线程对中断很敏感，那么就是用可中断锁，及时响应。

如果不敏感，也要注意处理中断异常。

AQS 的详细资源释放流程

首先 AQS 提供的模板方法为 release 方法。

核心逻辑就是对资源进行尝试性释放

如果成功，就唤醒等待队列中的第一个头节点

    public final boolean release(int arg) {        // 是否释放成功，tryRelease是子类要实现的方法        if (tryRelease(arg)) {            Node h = head;            // 判断头节点是否正在阻塞中，是的话唤醒            if (h != null && h.waitStatus != 0)                // 唤醒头节点                unparkSuccessor(h);            return true;        }        return false;    }

复制代码

看一下 ReteenLock 中的 tryRelease 实现

就是减一下资源值。

当资源值清零，则说明可以解除了对当前点的占用

    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;           // 设置当前占用线程为null            setExclusiveOwnerThread(null);        }        // 不需要CAS，因为只有持有锁的人才能做释放，不担心竞争        setState(c);        return free;    }

复制代码

AQS 如何实现公平和非公平？以 ReteenLock 为例，它内部 tryAcquire 有两种同步器的实现

非公平同步器 NonfairSync

公平同步器 FairSync

公平同步器和非公平同步器都是 ReentrantLock 中定义的一个 static 内部类

ReentrantLock 根据配置的不同，使用这 2 个同步器做资源的获取和同步操作

他们二者的提供的 lock 操作，本质上就是 AQS 的 acquire(1)

static final class FairSync extends Sync {    private static final long serialVersionUID = -3000897897090466540L;
    final void lock() {        acquire(1);    }

复制代码

二者在公平和非公平的实现区别上，就是唤醒线程后，只有等待队列的队头节点才会尝试竞争。

而非公平锁是只要唤醒了就可以尝试竞争。

因此核心区别在于 hasQueuedPredecessors 方法！image.png

公平和非公平锁的优点和缺点饥饿问题非公平锁可能引发“饥饿”，即一个线程反复抢占获取，而其他线程一直拿不到。

而公平锁不存在饥饿，只要排上队了就一定能拿到

性能问题

非公平锁的平均性能比公平锁要高，因为非公平锁中所有人都可以 CAS 抢占，如果同步块的时间非常短，那么可能所有人都不需要阻塞，减少 CPU 唤醒线程的开销，整体的吞吐效率会高点，CPU 也不必取唤醒所有线程，会减少唤起线程的数量。

性能测试中公平锁的耗时是非公平锁的 94.3 倍，总切换次数是 133 倍

Lock 类是默认公平还是非公平？默认是非公平的，原因就是上文考虑的性能差距过大问题，因此公平锁只能用于特定对性能要求不高且饥饿发生概率不大的场景中。

独占模式和共享模式的 AQS 区别名字上，共享模式都会带一个 shard

返回值上，独占模式相关 acuire 方法放回的是 boolean 类型，而共享模式返回的是 int 值

核心概念上，区别在于同一时刻能否有多个线程可以获取到其同步状态

释放时，共享模式需要用 CAS 进行释放，而独占模式的 release 方法则不需要，直接 setState 即可。

共享模式应用：信号量、读写锁

共享模式信号量 Semaphore 的 Sync 同步器先实现了一个静态内部类 Sync

和上面的 RLock 类一个区别在于需要 state 初始化值，不一定为 1

    Sync(int permits) {        setState(permits);    }

复制代码

再继承实现了 FairSync 和 NoFairSync

使用 CAS 实现值的增加或者减少

公平/非公平的区别同样是 hasQueuedPredecessors 的判断

        protected int tryAcquireShared(int acquires) {            for (;;) {                // 队头判断，公平锁核心                if (hasQueuedPredecessors())                    return -1;                int available = getState();                int remaining = available - acquires;                // 信号量不足，直接返回负数                if (remaining < 0 ||                    // 能抢成功，返回修改后的值，抢失败则for循环继续                    compareAndSetState(available, remaining))                    return remaining;            }        }

复制代码

AQS 如何处理重入

通过 current == getExclusiveOwnerThread()来判断并进行非 CAS 的 setState 操作

if (current == getExclusiveOwnerThread()) {    int nextc = c + acquires;    // 出现负数，说明溢出了    if (nextc < 0) //         throw new Error("Maximum lock count exceeded");    // 因为是重入操作，可以直接进行state的增加，所以不需要CAS    setState(nextc);    return true;}

复制代码

注意处理重入问题时，如果是独占锁，是可以直接 setState 而不需要 CAS 的，因为不会竞争式地重入！

ReentrantLock 释放时，也会处理重入，关键点就是对 getState() - release 后的处理，是否返回 true 或者 false

protected final boolean tryRelease(int releases) {    int c = getState() - releases;    if (Thread.currentThread() != getExclusiveOwnerThread())        throw new IllegalMonitorStateException();    boolean free = false;    if (c == 0) {        // 只有资源数为0才会解锁        // 才算释放成功，否则这锁还是占住了        free = true;        setExclusiveOwnerThread(null);    }    setState(c);    return free;}

复制代码

AQS 如何响应超时

AQS 提供的方法中带有 Nanos 后缀的方法就是支持超时中断的方法。

核心逻辑就是每次阻塞前，确认 nanosTimeout 是否已经超时了。

每次唤醒时，将 nanosTimeout 减去阻塞所花的时间，重新确认，并修改 lastTime

关键部分见下图

spinForTimeoutThreshold 是什么？

首先这个值是写死的 1000L 即 1000 纳秒

1000 纳秒是个非常小的数字，而小于等于 1000 纳秒的超时等待，无法做到十分的精确，那么就不要使用这么短的一个超时时间去影响超时计算的精确性，所以这时线程不做超时等待，直接做自旋就好了。

点击关注，第一时间了解华为云新鲜技术~

发布于: 刚刚阅读数: 4

原文链接:【http://xie.infoq.cn/article/2b593dc9ece47063575fb614c】。文章转载请联系作者。

华为云开发者联盟

关注

提供全面深入的云计算技术干货 2020-07-14 加入

华为云开发者社区，提供全面深入的云计算前景分析、丰富的技术干货、程序样例，分享华为云前沿资讯动态，方便开发者快速成长与发展，欢迎提问、互动，多方位了解云计算！传送门：https://bbs.huaweicloud.com/

发布

暂无评论

创作场景

JUC 中的 AQS 底层详细超详解

java 中 AQS 究竟是做什么的？

AQS 简要步骤

AQS 的资源 state

state 定义

已经有 CAS 方法了，为什么资源 state 还要定义成 volatile 的？

对资源的操作什么时候用 CAS，什么使用 setState？

AQS 的 CLH 队列

为什么需要一个 CLH 队列

CLH 本质？是一个 Queue 容器吗

入队是怎么保证安全的？

出队过程会发生什么？

AQS 详细资源获取流程

1. tryAcquire 尝试获取资源

ReentrantLock 中的 tryAcquire 实现

2.addWaiter 添加到等待队列

3. acquireQueued 循环阻塞-竞争

4.shouldParkAfterFailedAcquire 检查是否可以阻塞

5.parkAndCheckInterrupt() 阻塞线程

lockSupport.park()和普通的 wait|notify 都有啥区别？

6.总体流程图

代码中频繁出现的 interruptd 中断标记是做什么用的？

AQS 的详细资源释放流程

AQS 如何处理重入

AQS 如何响应超时

spinForTimeoutThreshold 是什么？

华为云开发者联盟

评论