AQS(AbstractQueuedSynchronizer)源码解析

前言

java.util.concurrent包(之后简称JUC包)中,提供了大量的同步与并发的工具类,是多线程编程的“利器”。其中locks包下,包含了多种锁,如ReentrantLock独占锁、ReentrantReadWriteLock读写锁、Semaphore信号量(共享锁)等,而这些锁有一个共同的基础类:AbstractQueuedSynchronizer。

AQS简介

AQS是一个抽象类,不可以被实例化,它的设计之初就是为了让子类通过继承来实现多样的功能的。它内部提供了一个FIFO的等待队列,用于多个线程等待一个事件(锁)。它有一个重要的状态标志——state,该属性是一个int值,表示对象的当前状态(如0表示lock,1表示unlock)。AQS提供了三个protected final的方法来改变state的值,分别是:getState、setState(int)、compareAndSetState(int, int)。根据修饰符,它们是不可以被子类重写的,但可以在子类中进行调用,这也就意味着子类可以根据自己的逻辑来决定如何使用state值。

java的修饰符作用域如下:

/ 同一类中 同一包中 同包内的子类 不同包内的子类 全局
public + + + + +
protected + + + +
default(无修饰符) + + +
private +

表中 + 表示可以访问, 空白表示无法访问

AQS的子类应当被定义为内部类,作为内部的helper对象。事实上,这也是juc种锁的做法,如ReentrantLock,便是通过内部的Sync对象来继承AQS的。AQS中定义了一些未实现的方法(抛出UnsupportedOperationException异常)

  • tryAcquire(int) 尝试获取state
  • tryRelease(int) 尝试释放state
  • tryAcquireShared(int) 共享的方式尝试获取
  • tryReleaseShared(int) 共享的方式尝试释放
  • isHeldExclusively() 判断当前是否为独占锁

这些方法是子类需要实现的,可以选择实现其中的一部分。根据实现方式的不同,可以分为两种:独占锁和共享锁。其中JUC中锁的分类为:

  • 独占锁:ReentrantLock、ReentrantReadWriteLock.WriteLock
  • 共享锁:ReentrantReadWriteLock.ReadLock、CountDownLatch、CyclicBarrier、Semaphore

其实现方式为:

  • 独占锁实现的是tryAcquire(int)、tryRelease(int)
  • 共享锁实现的是tryAcquireShared(int)、tryReleaseShared(int)

如独占锁的实现方式是:

1
2
3
4
5
6
7
8
9
Acquire:
while (!tryAcquire(arg)) {
//将当前线程加入FIFO队列中;
//自旋或阻塞当前线程;
}
Release:
if (tryRelease(arg))
//唤醒队列中的第一个线程(head);

AQS中还提供了一个内部类ConditionObject,它实现了Condition接口,可以用于await/signal。采用CLH队列的算法,唤醒当前线程的下一个节点对应的线程,而signalAll唤醒所有线程。

总的来说,AQS提供了三个功能:

  1. 实现独占锁
  2. 实现共享锁
  3. 实现Condition模型

源码解析

Node解析

AQS内部定义了一个static final的内部类Node,用于实现等待队列CLH,满足FIFO结构,其队列结构如下所示:

1
2
3
4
+------+ prev +-----+ +-----+
head | | <---- | | <---- | | tail
| | ----> | | ----> | |
+------+ next +-----+ +-----+

队列为一个双向链表结构,保存了head、tail两个指针,分别指向链表头部、尾部。当需要添加节点时,直接在tail位置添加,而dequeue操作直接对head节点进行。Node中定义如下常量:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
/** 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 the next acquireShared should
* unconditionally propagate
*/
static final int PROPAGATE = -3;

以上常量分别用于设置如下属性的值:

1
2
3
4
5
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter;

Node类型的常量SHARED、EXCLUSIVE用于设置nextWaiter,用于表示当前节点是共享的,还是互斥的,分别用于共享锁和独占锁。int类型的常量CANCELLED、SIGNAL、CONDITION、PROPAGATE用于设置waitStatus,用于在ConditionObject中使用,可以实现await/signal模型。

Node有三个构造函数:

1
2
3
4
5
6
//不存放任何线程,用于生成哨兵节点
Node() ;
//用于锁
Node(Thread thread, Node mode) ;
//用于Condition
Node(Thread thread, int waitStatus) ;

AQS属性

AQS使用内部类Node,构造一个双向链表,用作FIFO队列;除此之外,AQS还存放一个int类型的属性state,用于表示当前的同步状态。

1
2
3
4
5
6
//链表头节点
private transient volatile Node head;
//链表尾节点
private transient volatile Node tail;
//同步状态
private volatile int state;

head节点是一个哨兵节点,不存放实际的“线程”节点(使用Node的无参构造函数)。tail指向链表的最后一个节点,当新增节点时,将新节点作为当前tail的下一个节点,通过CAS设置成功后,将新节点设为新的tail节点即可。新增节点的源码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
private Node enq(final Node node) {
for (;;) { //死循环
Node t = tail;
if (t == null) { // 空链表,head、tail都为空
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}

enq操作是一个无限循环的操作,最终总会成功,但根据代码可知,AQS应不是starvation free的,因为某个线程可能会持续的enq失败。AQS提供了形如doAcquireNanos方法,但其超时返回false操作是在addWaiter方法(内部调用enq)之后,也无法回避enq的starvation。在此顺便说一下,AQS也是无法保证fair的,也就是说先入队列的线程不一定先获取到锁。节点的CAS是通过Unsafe来实现的,在state中统一说明。

state表示AQS当前的同步状态,如0表示lock,1表示unlock状态。对state的操作,提供了三个方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
//读取当前state
protected final int getState() {
return state;
}
//直接写入,不考虑当前值
protected final void setState(int newState) {
state = newState;
}
//保证读-写的原子性
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

可以看到compareAndSetState使用的是unsafe对象的compareAndSwapInt方法,传入this指针,state属性的偏移地址,期待值expect,更新值update,可以实现CAS操作。state属性的偏移地址获取方式如下:

1
2
3
4
5
6
7
8
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long stateOffset;
static {
try {
stateOffset = unsafe.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("state"));
} catch (Exception ex) { throw new Error(ex); }
}

实际上,AQS的head、tail节点,内部类Node的waitStatus、next属性均使用unsafe对象,通过偏移地址来进行CAS操作。Unsafe是sun.misc包下的类,在Java API中没有官方文档,因为它是用于实现Java库的,Java中有一个功能类似的类,可以实现对象属性的CAS操作,可以参考我的另一篇博客AtomicXFieldUpdater,属性原子修改的外部工具类,关于Unsafe的使用,可以参考Guide to Unsafe

AQS还有一个属性static final long spinForTimeoutThreshold = 1000L;,用于表示自旋的时间,小于1000纳秒的采用自旋锁,大于1000纳秒,使用LockSupport.park方法,将线程挂起。

重要方法分析

AQS是用于实现独占锁或共享锁的,对于一个锁来说,最重要的就是lock和unlock操作了,对应到AQS中,为acquire、release方法,由于AQS需要和子类进行“合作”,因此需要子类的定义来进行联合分析,为简单起见,使用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
class Mutex implements Lock, java.io.Serializable {
// Our internal helper class
private static class Sync extends AbstractQueuedSynchronizer {
// Reports whether in locked state
protected boolean isHeldExclusively() {
return getState() == 1;
}
// Acquires the lock if state is zero
public boolean tryAcquire(int acquires) {
assert acquires == 1; // Otherwise unused
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
// Releases the lock by setting state to zero
protected boolean tryRelease(int releases) {
assert releases == 1; // Otherwise unused
if (getState() == 0) throw new IllegalMonitorStateException();
setExclusiveOwnerThread(null);
setState(0);
return true;
}
// Provides a Condition
Condition newCondition() { return new ConditionObject(); }
// Deserializes properly
private void readObject(ObjectInputStream s)
throws IOException, ClassNotFoundException {
s.defaultReadObject();
setState(0); // reset to unlocked state
}
}
// The sync object does all the hard work. We just forward to it.
private final Sync sync = new Sync();
public void lock() { sync.acquire(1); }
public boolean tryLock() { return sync.tryAcquire(1); }
public void unlock() { sync.release(1); }
public Condition newCondition() { return sync.newCondition(); }
public boolean isLocked() { return sync.isHeldExclusively(); }
public boolean hasQueuedThreads() { return sync.hasQueuedThreads(); }
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
public boolean tryLock(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}
}

可以看到,lock方法调用内部类的acquire方法,也就是AQS的acquire方法。unlock方法调用release方法。
下面对两个流程进行分析

acquire

acquire是独占锁的获取锁的方法,其源码如下:

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

acquire方法非常简单,如果tryAcquire失败(返回false),则调用acquireQueued方法,将当前线程加入到等待队列中,并中断当前线程,等待唤醒。

tryAcquire由子类实现,下面先分析acquireQueued方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
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;
}
//park当前线程
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

acquireQueued在addWaiter之后,再次尝试获取锁,与tryAcquire不同的是,返回值代表的是当前线程在等待时是否可中断,通过返回interrupted,true表示可中断,false表示不可中断。通过判断当前节点是否为队列第一个节点,来决定是否获取成功,acquireQueued方法相当于提供了一个默认方法,会被子类的tryAcquire方法屏蔽掉(若tryAcquire返回true的话)。acquireQueued使用了死循环来判断当前节点前一节点是否为head,是,则获取到锁。但这个方法真的是死循环吗?其实不是的,trick就在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
26
27
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;
}

由于在锁的使用情景内,Node的waitStatus初始必定是0,因此获取锁的时候shouldParkAfterFailedAcquire方法首次进入,前一节点的waitStatus必定为0,执行compareAndSet将其设置为Node.SIGNAL(-1),再次调用shouldParkAfterFailedAcquire时,必定返回true。因此acquireQueued方法并不是死循环,而是在调用两次shouldParkAfterFailedAcquire方法之后(第一次将waitStatus设为-1,第二次返回true),执行了parkAndCheckInterrupt方法,挂起了当前线程。parkAndCheckInterrupt内调用了LockSupport.park(this),查阅API可知,park方法挂起当前线程,并在以下三种情况下立即返回。

  • 其他线程调用unpark,唤醒了当前线程
  • 其他线程调用了interrupt,中断了当前线程
  • 方法虚假返回(for no reason)

在AQS中,常见的为调用unpark(其他线程执行release释放锁时)唤醒了当前线程。当前线程唤醒之后,继续调用acquireQueue中的for循环,判断是否可以获取锁。

tryAcquire会调用子类Mutex.Sync的实现,其代码如下:

1
2
3
4
5
6
7
8
9
// 如果state为0,则获取到锁
public boolean tryAcquire(int acquires) {
assert acquires == 1; // Otherwise unused
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}

由此可见,AQS提供了一个模板,子类需要实现其tryAcquire方法,实现具体的获取锁逻辑(通过对state的读、写),子类lock方法直接调用AQS的acquire方法即可。

release方法

Mutex的unlock方法调用了release方法,在AQS中定义,源码如下:

1
2
3
4
5
6
7
8
9
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}

还是同样的配方,release方法调用子类实现的tryRelease,返回true后,表示获取成功,之后判断头节点,由于锁的实现中,waitStatus必定为0或者-1,而当一个线程lock成功后,waitStatus必定为-1,所以执行unparkSuccessor方法,该方法首先将head节点的waitStatus设为0,之后判断head下一节点是否为空,若不为空且waitStatus不大于0(大于0表示线程被取消),则调用LockSupport.unpark唤醒下一节点对应的线程;若为空或线程被取消,从tail节点开始遍历队列,找到队列中距离head节点最近的、未被cancel(waitStatus小于0)的节点,调用LockSupport.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
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);
}

这里简单介绍一下为什么循环是从tail往前遍历,这是因为CAS操作无法对双向链表进行原子插入,在enq中,具体的插入是,先将新节点指向tail,然后CAS将tail设为新节点,因此对于pred指针的设置时原子性的,可以保证从tail向前遍历可以找到所有的节点。而next指针只是一种优化路径,方便查找,并不能保证next为null,则该节点为队列的最后一个节点。

tryRelease方法的源码如下:

1
2
3
4
5
6
7
8
// 将state设置为0,解锁
protected boolean tryRelease(int releases) {
assert releases == 1; // Otherwise unused
if (getState() == 0) throw new IllegalMonitorStateException();
setExclusiveOwnerThread(null);
setState(0);
return true;
}

由源码可知,tryRelease只需要将state设置为0即可,因为调用unlock方法的必定是之前调用lock成功的,因此当前state必定为1,为安全起见,使用getState判断是否为0,若为0,说明线程出错。state设置时,不需要调用CAS方法,只需要setState即可,保证write对于其他线程可见即可(通过volatile内存屏障保证)。

总结

AQS提供了一个框架,在其上可以构建丰富的线程同步工具类,JUC包中ReadWriteLock、CountDownLatch都是基于AQS实现的,AQS在JUC包中的地位相当重要。其类图如下:

盗图使用,详见“JUC锁”01之 框架

AQS提供了三大功能:独占锁、共享锁、ConditionObject。子类在实现中,可以实现其一部分方法。其编程思想值得借鉴,通过超类实现基本的处理流程,将其中部分抽成未实现方法,默认抛出异常,由子类实现,这种解耦方式,最大化的减少了代码的重复,且便于子类在实现中个性化自己的处理逻辑。

很久没写博客了,准备以AQS入手,深入分析一下JUC包,flag就这么立起来了,希望可以实现吧~~

参考资料