Java并发编程第六篇（AQS设计理念与源码解析）

一，Java并发系列: 重中之重AQS

前面几篇文章中, 我们介绍了乐观锁, 而乐观锁的本质即是CAS, 操作系统提供了支持CAS修改内存值的原子指令, 所以乐观锁得以实现。从软件工程的角度去看, 虽然底层已经通过CAS实现了乐观锁, Java的底层已经在Unsafe这个类中封装了compareAndSwap方法, 支持了对CAS原语的调用, 为了使上层更加易用, 需要经过进一步的抽象和封装。抽象这个词虽然简单, 但私以为要做出高内聚低耦合的抽象绝对是难点。在Java中最著名的并发包就是JUC, 其中的组件和日常Java开发息息相关。

在JUC中, 我认为最核心的组件便是AQS, 可以这么理解, AQS是对CAS的一种封装和丰富, AQS引入了独占锁、共享锁等性质。基于AQS, JUC中提供了更多适用于各种预设场景的组件, 当然你也可以基于AQS开发符合自身业务场景的组件。所以, AQS作为承下启上的重点, 我们需要仔细来看。

1.1 尝试设计

首先, 我们可以尝试思考一下: 目前Java底层提供了CAS原语调用, 如果让你来设计一个中间框架, 它需要是通用的, 并且能够对被竞争的资源进行同步管理, 你会怎么做?

这里你可以停下来想一想自己的方案。当然, 我们目前的能力很难做到完全可用, 但至少可以思考一下设计思路, 再来看看大师是怎么做的。如果是我, 我会从这几点这去思考:

• 既然我要做一个框架, 首先它需要具有通用性, 因为上层业务逻辑是千变万化的, 所以这个框架在实现底层必要的同步机制的同时, 要保证对外接口的简单性和纯粹性。
• 既然CAS能够原子地对一个值进行写操作, 那么我可以将这个值(称为status)作为竞争资源的标记位。在多个线程想要去修改共享资源时, 先来读status, 如果status显示目前共享资源空闲可以被获取, 那么就赋予该线程写status的权限, 当该线程原子地修改status成功后, 代表当前线程占用了共享资源, 并将status置为不可用, 拒绝其他线程修改status, 也就是拒绝其他线程获取共享资源。
• 拒绝其他线程调用该怎么设计呢? 这里应该有两种业务场景, 有的业务线程它可能只想快速去尝试一下获取共享资源, 如果获取不到也没关系, 它会进行其他处理; 有的业务线程它可能一定要获取共享资源才能进行下一步处理, 如果当前时刻没有获取到, 它愿意等待。针对第一种场景, 直接返回共享资源的当前状态就可以了, 那么有的同学可能也会说, 第二种场景也能直接返回, 让业务线程自旋获取, 直到成功为止。这样说有一定的道理, 但是我认为存在两个弊端:
  • 第一, 让业务线程去做无保护的自旋操作会不断占用CPU时间片, 长时间自旋可能导致CPU使用率暴涨, 在CPU密集型业务场景下会降低系统的性能甚至导致不可用。但如果让上层业务去做保护机制, 无疑增加了业务开发的复杂度, 也增强了耦合。
  • 第二, 实现框架的目的是为了简化上层的操作, 封装内部复杂度, 第一点中我们也说到了需要保持对外接口的简单纯粹, 如果还需要上层进行额外的处理, 这并不是一个好的设计。

所以当业务线程它可能一定要获取共享资源才能进行下一步处理时(称为调用lock()), 我们不能直接返回。那么如果有大量的线程调用lock()时, 该如何对它们进行管理? 大致猜一猜, 可以设计一个队列来将这些线程进行排队。队列头部的线程自旋地访问status, 其他线程挂起, 这样就避免了大量线程的自旋内耗。当头部线程成功占用了共享资源, 那么它再唤醒后续一个被挂起的线程, 让它开始自旋地访问status。

我的大致思路讲完了, 事实上我说的内容和JUC中的经典同步框架AQS设计思路差不多。AQS全称是AbstractQueuedSynchronizer。顾名思义就是一个抽象的(可被继承复用), 内部存在排队(竞争资源的线程排队)的同步器(对共享资源和线程进行同步管理)。

开篇也提到了, AQS作为承下启下的重点, JUC中大量的组件以及一些开源中间件都依赖了AQS, 理解了AQS的大致思路, 我们对它有了一个粗糙的印象。想要进一步知其全貌, 剩下的就是复杂的实现细节。细节是魔鬼, 你应该会很好奇大师是怎么做的, 接下来我们就一起去AQS的源码里一探究竟。

二，AQS源码

说到看源码, 这是一件很多人都会感到恐惧的事情。我想根据自己的感悟聊三点。

• 第一: 如果0基础, 不建议读源码, 即使读了, 可能也是收效甚微, 迷茫而无所得。
• 第二, 读源码不难, 关键在于耐心, 读书百遍其义自现。此外不一定需要通读源码, 只要精读核心部分就足够了。
• 第三, 读源码的目的不是钻牛角尖, 而是为了理解细节和原理, 从细节之处学习高手的思想。

2.1 属性

我们首先来看AQS的成员属性。

private volatile int state;

state就是之前我们所说的, 用于判断共享资源是否正在被占用的标记位, volatile保证了线程之间的可见性。可见性简单来说, 就是当一个线程修改了state的值, 其他线程下一次读取都能读到最新值。state的类型是int, 可能有的同学有疑问, 为什么不是boolean? 用boolean来表示资源被占用与否, 语意上不是更明确吗?

这里就要谈到线程获取锁的两种模式, 独占和共享。简单介绍一下, 当一个线程以独占模式获取锁时, 其他任何线程都必须等待; 而当一个线程以共享模式获取锁时, 其他也想以共享模式获取锁的线程也能够一起访问共享资源, 但其他想以独占模式获取锁的线程需要等待。这就说明了, 共享模式下, 可能有多个线程正在共享资源, 所以state需要表示线程占用数量, 因此是int值。

private transient volatile Node head;
private transient volatile Node tail;

我们之前提到, AQS中存在一个队列用于对等待线程进行管理, 这个队列通过一个FIFO的双向链表来实现, 至于为什么选用这种数据结构, 在后面我们对方法的解析中, 能够体会到它的好处。head和tail变量表示这个队列的头尾。

2.2 内部类

队列中每个节点的类型是内部类Node, 我们来看看。由于篇幅原因, 这里去掉了源码中大段的注释, 建议有兴趣的同学可以自己去源码中翻看。

static final class Node {/** Marker to indicate a node is waiting in shared mode */static final Node SHARED = new Node();/** Marker to indicate a node is waiting in exclusive mode */static final Node EXCLUSIVE = null;/** waitStatus value to indicate thread has cancelled */static final int CANCELLED = 1;/** waitStatus value to indicate successor's thread needs unparking */static final int SIGNAL = -1;/** waitStatus value to indicate thread is waiting on condition */static final int CONDITION = -2;/** waitStatus value to indicate that a releaseShared should be* propagated to other nodes. This is set (for head node) in* doReleaseShared to ensure propagation continues, even if other* operations have since intervened.*/static final int PROPAGATE = -3;volatile int waitStatus;volatile Node prev;volatile Node next;volatile Thread thread;Node nextWaiter;final boolean isShared() {return nextWaiter == SHARED;}final Node predecessor() throws NullPointerException {Node p = prev;if (p == null)throw new NullPointerException();elsereturn p;}Node() {    // Used to establish initial head or SHARED marker}Node(Thread thread, Node mode) {     // Used by addWaiterthis.nextWaiter = mode;this.thread = thread;}Node(Thread thread, int waitStatus) { // Used by Conditionthis.waitStatus = waitStatus;this.thread = thread;}
}

队列里的节点有两种模式, 独占和共享, 上面简单介绍过了差别, 虽然这两者在表现的意义上不同, 但在底层的处理逻辑上没什么太大的差别, 所以本期内容我们只讲独占模式。

Node中主要存储了线程对象(thread)、节点在队列里的等待状态(waitStatus)、前后指针(prev、next)等信息。这里需要重点关注的是waitStatus这个属性, 它是一个枚举值, AQS工作时必然伴随着Node的waitStatus值的变化, 如果理解了waitStatus变化的时机, 那对理解AQS整个工作原理有很大的帮助。

waitStatus主要包含五个状态:

• 0, 节点初始化默认值或节点已经释放锁
• CANCELLED (1), 表示当前节点获取锁的请求已经被取消了
• SIGNAL (-1), 表示当前节点的后续节点需要被被唤醒
• CONDITION (-2), 表示当前节点正在等待某一个Condition对象, 和条件模式相关, 本期暂不介绍
• PROPAGATE (-3), 传递共享模式下锁释放状态, 和共享模式相关, 本期暂不介绍

Node中的方法也很简洁, predecessor就是获取前置Node。

到这里, 属性和内部类AQS的属性, 就这些内容, 非常简单。后面我们要重点关注的则是如何利用state和FIFO的队列来管理多线程的同步状态, 这些操作被封装成了方法。在对方法的解读上, 我们可以像剥洋葱一样, 自上而下, 层层深入。

2.3 方法

一开始我们提到了两种使用场景:

• 尝试获取锁, 不管有没有获取到, 立即返回。
• 必须获取锁, 如果当前时刻锁被占用, 则进行等待。

我们还没有看代码之前, 冥冥中猜测AQS最上层应该拥有这两个方法, 果然源码中tryAcquire和acquire正对应了这两个操作。

// try acquire
protected boolean tryAcquire(int arg) {throw new UnsupportedOperationException();
}// acquire
public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();
}

tryAcquire是一个被 protected 修饰的方法, 参数是一个int值, 代表对int state的增加操作, 返回值是boolean, 代表是否成功获得锁。

该方法只有一行实现 throw new UnsupportedOperationException(), 意图很明显, AQS规定继承类必须override tryAcquire方法, 否则就直接抛出UnsupportedOperationException。那么为什么这里一定需要上层自己实现?因为尝试获取锁这个操作中可能包含某些业务自定义的逻辑, 比如是否“可重入”等。

若上层调用tryAcquire返回true, 线程获得锁, 此时可以对相应的共享资源进行操作, 使用完之后再进行释放。如果调用tryAcquire返回false, 且上层逻辑上不想等待锁, 那么可以自己进行相应的处理; 如果上层逻辑选择等待锁, 那么可以直接调用acquire方法, acquire方法内部封装了复杂的排队处理逻辑, 非常易用。

接下来我们来看更加核心和复杂的acquire方法。

acquire被final修饰, 表示不允许子类擅自override, 似乎是在宣示: 等待并获取锁, 我非常可靠, 直接用就行, 其他您就甭操心了。

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

if判断条件包含了两部分:

• !tryAcquire(arg)
• acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

tryAcquire我们已经讲过了, 这里的意思是, 如果tryAcquire获取锁成功, 那么!tryAcquire为false, 说明已经获取锁, 根本不用参与排队, 也就是不用再执行后续判断条件。根据判断条件的短路规则, 直接返回。

假如tryAcquire返回false, 说明需要排队, 那么就进而执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg), acquireQueued 方法其中嵌套 addWaiter 方法。

前面说我们像剥洋葱一样来读源码, 那么先来品一品addWaiter。

/*** Creates and enqueues node for current thread and given mode.** @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared* @return the new node*/
private Node addWaiter(Node mode) {Node node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failureNode pred = tail;if (pred != null) {node.prev = pred;if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}enq(node);return node;
}

顾名思义, 这个方法的作用就是将当前线程封装成一个Node, 然后加入等待队列, 返回值即为该Node。逻辑也非常简单, 首先新建一个Node对象, 之前也说过这个队列是先入先出的, 接下来顺理成章地想到, 我们需要将其插入队尾。但是下面我们需要考虑多线程场景, 即假设存在多个线程正在同时调用addWaiter方法。

新建pred节点引用, 指向当前的尾节点, 如果尾节点不为空, 那么下面将进行三步操作:

1. 将当前节点的pre指针指向pred节点(尾节点)
2. 尝试通过CAS操作将当前节点置为尾节点
   • a. 如果返回false, 说明pred节点已经不是尾节点, 在上面的执行过程中, 尾节点已经被其他线程修改, 那么退出判断, 调用enq方法, 准备重新进入队列。
   • b. 如果返回true, 说明CAS操作之前, pred节点依然是尾节点, CAS操作使当前node顺利成为尾节点。若当前node顺利成为尾节点, 那么pred节点和当前node之间的相对位置已经确定, 此时将pred节点的next指针指向当前node, 是不会存在线程安全问题的。

由于在多线程环境下执行, 这里存在三个初学者容易迷糊的细节, 也是该方法中的重点。

1. 某线程执行到第13行时, pred引用指向的对象可能已经不再是尾节点, 所以CAS失败;
2. 如果CAS成功, 诚然CAS操作是具有原子性的, 但是14, 15两行在执行时并不具备原子性, 只不过此时pred节点和当前节点的相对位置已经确定, 其他线程只是正在插入新的尾节点, 并不会影响到这里的操作, 所以是线程安全的。
3. 需要记住的是, 当前后两个节点建立连接的时候, 首先是后节点的pre指向前节点, 当后节点成功成为尾节点后, 前节点的next才会指向后节点。

如果理解了这些, 我们再来看第18行。如过程序运行到这一行, 说明出现了两种情况之一:

• 队列为空
• 快速插入失败, 想要进行完整流程的插入, 这里所说的快速插入, 指的就是11-17行的逻辑, 当并发线程较少的情况下, 快速插入成功率很高, 程序不用进入完整流程插入, 效率会更高。

既然程序来到了第18行, 那么我们就来看看完整流程的插入是什么样子的。

private Node enq(final Node node) {for (;;) {Node t = tail;if (t == null) { // Must initializeif (compareAndSetHead(new Node()))tail = head;} else {node.prev = t;if (compareAndSetTail(t, node)) {t.next = node;return t;}}}
}

这个方法里的逻辑, 有一种似曾相识的感觉, 其实就是在最外层加上了一层死循环, 如果队列未初始化 (tail == null), 那么就尝试初始化, 如果尾插节点失败, 那么就不断重试, 直到插入成功为止。

一旦addWaiter成功之后, 不能就这么不管了, 我最初的猜测是: 既然存在一个FIFO队列, 那么可能会使用了“生产-消费”模式, 有一个消费者不断从这个队列的头部获取节点, 出队节点中封装的线程拥有拿锁的权限。

但是实际上AQS并没有这么做, 而是在各个线程中维护了当前Node的waitStatus, 根据根据不同的状态, 程序来做出不同的操作。通过调用 acquireQueued 方法, 开始对Node的waitStatus进行跟踪维护。

至于为什么没有使用我猜想中的那种方案, 大家可以说说自己的想法。

我们继续来看 acquireQueued 源码。

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 GCfailed = false;return interrupted;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}
}

首先, acquireQueued方法内定义了一个局部变量failed, 初始值为true, 意思是默认失败。还有一个变量interrupted, 初始值为false, 意思是等待锁的过程中当前线程没有被中断。再来看看在整个方法中, 哪里用到了这两个变量?

1. 第11行, return之前, failed值会改为false, 代表执行成功, 并且返回interrupted值。
2. 第15行, 如果满足判断条件, interrupted将会被改为true, 最终在第11行被返回出去。
3. 第18行, finally块中, 通过判断failed值来进行一个名为cancelAcquire的操作, 即取消当前线程获取锁的行为。

那么我们基本可以将acquireQueued分为三部分。

• 7-11行。当前置节点为head, 说明当前节点有权限去尝试拿锁, 这是一种约定。如果tryAcquire返回true, 代表拿到了锁, 那么顺理成章, 函数返回。如果不满足第7行的条件, 那么进入下一阶段。
• 13-15行。if中包含两个方法, 看名字(详细方法体后续再看)是首先判断当前线程是否需要挂起等待?如果需要, 那么就挂起, 并且判断外部是否调用线程中断;如果不需要, 那么继续尝试拿锁。
• 18-19行。如果try块中抛出非预期异常, 那么取消当前线程获取锁的行为。

这里呢, 有三点需要着重关注一下。

1. 一个约定: head节点代表当前正在持有锁的节点。若当前节点的前置节点是head, 那么该节点就开始自旋地获取锁。一旦head节点释放, 当前节点就能第一时间获取到。
2. shouldParkAfterFailedAcquire 和 parkAndCheckInterrupt方法体细节。
3. interrupted变量最终被返回出去后, 上层acquire方法判断该值, 来选择是否调用当前线程中断。这里属于一种延迟中断机制。

我们下面着重看一下第二点中提到的两个方法。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus;if (ws == Node.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 {/** 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;
}

若当前节点没有拿锁的权限或拿锁失败, 那么将会进入shouldParkAfterFailedAcquire 判断是否需要挂起(park), 方法的参数是pred Node和当前Node的引用。

回到方法中, 若pred的waitSatus为SIGNAL, 说明前置节点也在等待拿锁, 并且之后将会唤醒当前节点, 所以当前线程可以挂起休息, 返回true。

如果ws大于0, 说明pred的waitSatus是CANCELLED, 所以可以将其从队列中删除。这里通过从后向前搜索, 将pred指向搜索过程中第一个waitSatus为非CANCELLED的节点。相当于链式地删除被CANCELLED的节点。然后返回false, 代表当前节点不需要挂起, 因为pred指向了新的Node, 需要重试外层的逻辑。

除此之外, pred的ws还有两种可能, 0或PROPAGATE, 有人可能会问, 为什么不可能是CONDITION?因为waitStatus只有在其他条件模式下, 才会被修改为CONDITION, 这里不会出现, 并且只有在共享模式下, 才可能出现waitStatus为PROPAGATE, 暂时也不用管。那么在独占模式下, ws在这里只会出现0的情况。0代表pred处于初始化默认状态, 所以通过CAS将当前pred的waitStatus修改为SIGNAL, 然后返回false, 重试外层逻辑。

这个方法开始涉及到对Node的waitSatus的修改, 相对比较关键。

如果shouldParkAfterFailedAcquire返回false, 那么再进行一轮重试; 如果返回true, 代表当前节点需要被挂起, 则执行parkAndCheckInterrupt方法。

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

这个方法只有两行, 对当前线程进行挂起的操作。这里 LockSupport.park(this) 本质是通过UNSAFE下的native方法调用操作系统原语来将当前线程挂起。

此时当前Node中的线程将阻塞在此处, 直到持有锁的线程调用release方法, release方法会唤醒后续节点。

那这边的 return Thread.interrupted() 又是什么意思呢? 这是因为在线程挂起期间, 该线程可能会被调用中断方法, 线程在park期间, 无法响应中断, 所以只有当线程被唤醒, 执行到第3行, 才会去检查park期间是否被调用过中断, 如果有的话, 则将该值传递出去, 通过外层来响应中断。

通过对acquireQueued这个方法的分析, 我们可以这么说, 如果当前线程所在的节点处于头节点的后一个, 那么它将会不断去尝试拿锁, 直到获取成功。否则进行判断, 是否需要挂起。这样就能保证head之后的一个节点在自旋CAS获取锁, 其他线程都已经被挂起或正在被挂起。这样就能最大限度地避免无用的自旋消耗CPU。

但事情还没有结束, 既然大量线程被挂起, 那么就会有被唤醒的时机。上面也提到, 当持有锁的线程释放了锁, 那么将会尝试唤醒后续节点。我们一起来看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 boolean tryRelease(int arg) {throw new UnsupportedOperationException();
}

和tryAcquire一样, tryRelease也是AQS开放给上层自由实现的抽象方法。

在release中, 假如尝试释放锁成功, 下一步就要唤醒等待队列里的其他节点, 这里主要来看 unparkSuccessor这个方法。参数是head 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);
}

获取head的waitStatus, 如果不为0, 那么将其置为0, 表示锁已释放。接下来获取后续节点如果后续节点为null或者处于CANCELED状态, 那么从后往前搜索, 找到除了head外最靠前且非CANCELED状态的Node, 对其进行唤醒, 让它起来尝试拿锁。

这时, 拿锁、挂起、释放、唤醒都能够有条不紊, 且高效地进行。

关于20-22行, 可能有的同学有一个疑问, 为什么不直接从头开始搜索, 而是要花这么大力气从后往前搜索? 这个问题很好, 其实是和addWaiter方法中, 前后两个节点建立连接的顺序有关。我们看:

1. 后节点的pre指向前节点
2. 前节点的next才会指向后节点

这两步操作在多线程环境下并不是原子的, 也就是说, 如果唤醒是从前往后搜索, 那么可能前节点的next还未建立好, 那么搜索将可能会中断。