Java - JUC核心类AbstractQueuedSynchronizer(AQS)底层实现
一. AQS内部结构介绍
JUC是Java中一个包 java.util.concurrent 。在这个包下,基本存放了Java中一些有关并发的类,包括并发工具,并发集合,锁等。
AQS(抽象队列同步器)是JUC下的一个基础类,大多数的并发工具都是基于AQS实现的。
AQS本质并没有实现太多的业务功能,只是对外提供了三点核心内容,来帮助实现其他的并发内容。
三点核心内容:
- int state
- 比如ReentrantLock或者ReentrantReadWriteLock, 它们获取锁的方式,都是对state变量做修改实现的。
- 比如CountDownLatch基于state作为计数器,同样的Semaphore也是用state记录资源个数。
- Node对象组成的双向链表(AQS中)
- 比如ReentrantLock,有一个线程没有拿到锁资源,当线程需要等待,则需要将线程封装为Node对象,将Node添加到双向链表,将线程挂起,等待即可。
- Node对象组成的单向链表(AQS中的ConditionObject类中)
- 比如ReentrantLock,一个线程持有锁资源时,执行了await方法(类比synchronized锁执行对象的wait方法),此时这个线程需要封装为Node对象,并添加到单向链表。
二. Lock锁和AQS关系
ReentrantLock就是基于AQS实现的。ReentrantLock类中维护这个一个内部抽象类Sync,他继承了AQS类。ReentrantLock的lock和unlock方法就是调用的Sync的方法。
AQS流程(简述)
1. 当new了一个ReentrantLock时,AQS默认state值为0, head 和 tail 都为null;
2. A线程执行lock方法,获取锁资源。
3. A线程将state通过cas操作从0改为1,代表获取锁资源成功。
4. B线程要获取锁资源时,锁资源被A线程持有。
5. B线程获取锁资源失败,需要添加到双向链表中排队。
6. 挂起B线程,等待A线程释放锁资源,再唤醒挂起的B线程。
7. A线程释放锁资源,将state从1改为0,再唤醒head.next节点。
8. B线程就可以重新尝试获取锁资源。
注: 修改AQS双向链表时要保证一个私有属性变化和两个共有属性变化,只需要让tail变化保证原子性即可。不能先改tail(会破坏双向链表)

三. AQS - Lock锁的tryAcquire方法
ReentrantLock中的lock方法实际是执行的Sync的lock方法。
Sync是一个抽象类,继承了AQS
Sync有两个子类实现:
- FairSync: 公平锁
- NonFairSync: 非公平锁
Sync的lock方法实现:
- // 非公平锁
- final void lock() {
- // CAS操作,尝试将state从0改为1
- // 成功就拿到锁资源, 失败执行acquire方法
- if (compareAndSetState(0, 1))
- // 成功就设置互斥锁的为当前线程拥有
- setExclusiveOwnerThread(Thread.currentThread());
- else
- acquire(1);
- }
-
- // 公平锁
- final void lock() {
- acquire(1);
- }
如果CAS操作没有成功,需要执行acquire方法走后续
acquire方法是AQS提供的,公平和非公平都是走的这个方法
- public final void acquire(int arg) {
- // 1. tryAcquire方法: 再次尝试拿锁
- // 2. addWaiter方法: 没有获取到锁资源,去排队
- // 3. acquireQueued方法:挂起线程和后续被唤醒继续获取锁资源的逻辑
- if (!tryAcquire(arg) &&
- acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
- // 如果这个过程中出现中断,在整个过程结束后再自我中断
- selfInterrupt();
- }
在AQS中tryAcquire是没有具体实现逻辑的,AQS直接在tryAcquire方法中抛出异常
在公平锁和非公平锁中有自己的实现。
- // 非公平锁
- protected final boolean tryAcquire(int acquires) {
- return nonfairTryAcquire(acquires);
- }
-
- // 非公平锁再次尝试拿锁 (注:该方法属于Sync类中)
- final boolean nonfairTryAcquire(int acquires) {
- // 获取当前线程对象
- final Thread current = Thread.currentThread();
- // 获取state状态
- int c = getState();
- // state是不是没有线程持有锁资源,可以尝试获取锁
- if (c == 0) {
- // 再次CAS操作尝试修改state状态从0改为1
- if (compareAndSetState(0, acquires)) {
- // 成功就设置互斥锁的为当前线程拥有
- setExclusiveOwnerThread(current);
- return true;
- }
- }
- // 锁资源是否被当前线程所持有 (可重入锁)
- else if (current == getExclusiveOwnerThread()) {
- // 持有锁资源为当前, 则对state + 1
- int nextc = c + acquires;
- // 健壮性判断
- if (nextc < 0) // overflow
- // 超过最大锁重入次数会抛异常(几率很小,理论上存在)
- throw new Error("Maximum lock count exceeded");
- // 设置state状态,代表锁重入成功
- setState(nextc);
- return true;
- }
- return false;
- }
- // 公平锁
- protected final boolean tryAcquire(int acquires) {
- // 获取当前线程对象
- final Thread current = Thread.currentThread();
- // 获取state状态
- int c = getState();
- // state是不是没有线程持有锁资源
- if (c == 0) {
- // 当前锁资源没有被其他线程持有
- // hasQueuedPredecessors方法: 锁资源没有被持有,进入队列排队
- // 排队规则:
- // 1. 检查队列没有线程排队,抢锁。
- // 2. 检查队列有线程排队,查看当前线程是否排在第一位,如果是抢锁,否则入队列(注:该方法只是判断,没有真正入队列)
- if (!hasQueuedPredecessors() &&
- compareAndSetState(0, acquires)) {
- // 再次CAS操作尝试, 成功就设置互斥锁的为当前线程拥有
- setExclusiveOwnerThread(current);
- return true;
- }
- }
- // 锁资源是否被当前线程所持有 (可重入锁)
- else if (current == getExclusiveOwnerThread()) {
- // 持有锁资源为当前, 则对state + 1
- int nextc = c + acquires;
- // 健壮性判断
- if (nextc < 0)
- // 超过最大锁重入次数会抛异常(几率很小,理论上存在)
- throw new Error("Maximum lock count exceeded");
- // 设置state状态,代表锁重入成功
- setState(nextc);
- return true;
- }
- return false;
- }
四. AQS的addWaiter方法
addWaiter方法,就是将当前线程封装为Node对象,并且插入到AQS的双向链表。
- // 线程入队列排队
- private Node addWaiter(Node mode) {
- // 将当前对象封装为Node对象
- // Node.EXCLUSIVE 表示互斥 Node.SHARED 表示共享
- Node node = new Node(Thread.currentThread(), mode);
- // 获取tail节点
- Node pred = tail;
- // 判断双向链表队列有没有初始化
- if (pred != null) {
- // 将当前线程封装的Node节点prev属性指向tail尾节点
- node.prev = pred;
- // 通过CAS操作设置当前线程封装的Node节点为尾节点
- if (compareAndSetTail(pred, node)) {
- // 成功则将上一个尾节点的next属性指向当前线程封装的Node节点
- pred.next = node;
- return node;
- }
- }
- // 没有初始化head 和 tail 都等于null
- // enq方法: 插入双向链表和初始化双向链表
- enq(node);
- // 完成节点插入
- return node;
- }
-
- // 插入双向链表和初始化双向链表
- private Node enq(final Node node) {
- // 死循环
- for (;;) {
- // 获取当前tail节点
- Node t = tail;
- // 判断尾节点是否初始
- if (t == null) { // Must initialize
- // 通过CAS操作初始化初始化一个虚拟的Node节点,赋给head节点
- if (compareAndSetHead(new Node()))
- tail = head;
- } else {
- // 完成当前线程Node节点加入AQS双向链表的过程
- // 当前线程封装的Node的上一个prev属性指向tail节点
- // 流程: 1. prev(私有) ---> 2. tail(共有) ---> 3. next (共有)
- node.prev = t;
- // 通过CAS操作修改tail尾节点指向当前线程封装的Node
- if (compareAndSetTail(t, node)) {
- // 将当前线程封装的Node节点赋给上一个Node的下一个next属性
- t.next = node;
- return t;
- }
- }
- }
- }
五. AQS的acquireQueued方法
acquireQueued方法主要就是线程挂起以及重新尝试获取锁资源的地方
重新获取锁资源主要有两种情况:
- 上来就排在head.next,就回去尝试拿锁
- 唤醒之后尝试拿锁
- // 当前线程Node添加到AQS队列后续操作
- final boolean acquireQueued(final Node node, int arg) {
- // 标记,记录拿锁状态 失败
- boolean failed = true;
- try {
- // 中断状态
- boolean interrupted = false;
- // 死循环
- for (;;) {
- // 获取当前节点的上一个节点 prev
- final Node p = node.predecessor();
- // 判断当前节点是否是head,是则代表当前节点排在第一位
- // 如果是第一位,执行tryAcquire方法尝试拿锁
- if (p == head && tryAcquire(arg)) {
- // 都成功,代表拿到锁资源
- // 将当前线程Node设置为head节点,同时将Node的thread 和 prev属性设置为null
- setHead(node);
- // 将上一个head的next属性设置为null,等待GC回收
- p.next = null; // help GC
- // 拿锁状态 成功
- failed = false;
- // 返回中断状态
- return interrupted;
- }
- // 没有获取到锁 --- 尝试挂起线程
- // shouldParkAfterFailedAcquire方法: 挂起线程前的准备
- // parkAndCheckInterrupt方法: 挂起当前线程
- if (shouldParkAfterFailedAcquire(p, node) &&
- parkAndCheckInterrupt())
- // 设置中断线程状态
- interrupted = true;
- }
- } finally {
- // 取消节点
- if (failed)
- cancelAcquire(node);
- }
- }
-
- // 检查并更新无法获取锁节点的状态
- private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
- // 获取上一个节点的ws状态
- /**
- * SIGNAL(-1) 表示当前节点释放锁的时候,需要唤醒下一个节点。或者说后继节点在等待当前节点唤醒,后继节点入队时候,会将前驱节点更新给signal。
- * CANCELLED(1) 表示当前节点已取消调度。当timeout或者中断情况下,会触发变更为此状态,进入该状态后的节点不再变化。
- * CONDITION(-2) 当其他线程调用了condition的signal方法后,condition状态的节点会从等待队列转移到同步队列中,等待获取同步锁。
- * PROPAGATE(-3) 表示共享模式下,前驱节点不仅会唤醒其后继节点,同时也可能唤醒后继的后继节点。
- * 默认(0) 新节点入队时候的默认状态。
- */
- int ws = pred.waitStatus;
- // 判断上个节点ws状态是否是 -1, 是则挂起
- if (ws == Node.SIGNAL)
- return true;
- if (ws > 0) {
- /**
- * 判断上个节点是否是取消或者其他状态。
- * 向前找到不是取消状态的节点,修改ws状态。
- * 注意:那些放弃的结点,由于被自己“加塞”到它们前边,它们相当于形成一个无引用链,
- * 稍后就会被GC回收,这个操作实际是把队列中的cancelled节点剔除掉。
- */
- do {
- node.prev = pred = pred.prev;
- } while (pred.waitStatus > 0);
- pred.next = node;
- } else {
- // 如果前驱节点正常,那就把上一个节点的状态通过CAS的方式设置成-1
- compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
- }
- return false;
- }
-
- // 挂起当前线程
- private final boolean parkAndCheckInterrupt() {
- // 挂起当前线程
- LockSupport.park(this);
- // 返回中断标志
- return Thread.interrupted();
- }
六. AQS的Lock锁的release方法
- // 互斥锁模式 解锁
- 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) {
- // 锁状态 = 状态 - 1
- int c = getState() - releases;
- // 判断锁是是否是当前线程持有
- if (Thread.currentThread() != getExclusiveOwnerThread())
- // 当前线程没有持有抛出异常
- throw new IllegalMonitorStateException();
- boolean free = false;
- // 当前锁状态变为0,则清空锁归属线程
- if (c == 0) {
- free = true;
- setExclusiveOwnerThread(null);
- }
- // 设置锁状态为0
- setState(c);
- return free;
- }
-
- // 唤醒线程
- private void unparkSuccessor(Node node) {
- // 获取头节点的状态
- int ws = node.waitStatus;
- if (ws < 0)
- // 通过CAS将头节点的状态设置为初始状态
- compareAndSetWaitStatus(node, ws, 0);
- // 后继节点
- 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);
- }
以上仅供参考!!