JAVA - AQS源码解读

JAVA - AQS源码解读

前言

AQS全称AbstractQueuedSynchronizer。是LockCountDownLatchSemaphore(信号量)的基础。这个AQS我在大约一年多以前看过一次,但是当时水平有限,根本没看懂什么意思,最后对于AQS的理解都是来自于网上的博文,可以说是吃人家嚼过的。。这次重读源码还是收获颇丰。

Node

看一下Node的结构:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
//标识等待节点处于共享模式
static final Node SHARED = new Node();

//标识等待节点处于独占模式 这里直接是个null啊,有点懵逼,不知道有什么讲究
static final Node EXCLUSIVE = null;
//由于超时或中断,节点已被取消
static final int CANCELLED = 1;
//表示下一个节点是通过park堵塞的,需要通过unpark唤醒
static final int SIGNAL = -1;
//表示线程在等待条件变量(先获取锁,加入到条件等待队列,然后释放锁,等待条件变量满足条件;只有重新获取锁之后才能返回)
static final int CONDITION = -2;
//表示后续结点会传播唤醒的操作,共享模式下起作用
static final int PROPAGATE = -3;

//等待状态:对于condition节点,初始化为CONDITION;其它情况,默认为0
volatile int waitStatus;
volatile Node prev;
volatile Node next;
//线程对象,这里我们可以看出来,一个Node其实就是一个线程的包装
volatile Thread thread;
//对于Condtion表示下一个等待条件变量的节点;其它情况下用于区分共享模式和独占模式;
Node nextWaiter;



这里对这个nextWaiter的概念还不是很了解,故去找了一些大神的博客看了一下:

  • 独占模式:每次只能有一个线程能持有资源
  • 共享模式:允许多个线程同时持有资源

这么说可能还是很抽象,借鉴大神的几个例子:

  1. CountDownLatchawait方法可以在多个线程中调用,当CountDownLatch的计数器为0后,调用await的方法都会依次返回。 也就是说多个线程可以同时等待await方法返回,因此它适合被设计成共享模式,因为它获取的是一个共享资源,资源在所有调用await方法的线程间共享;
  2. ReentrantLock提供了lockunlock方法,只允许一个线程获得锁,因此它适合被设计成独占模式,因为它获取的是一个独占资源,资源不能在调用lock方法的线程间共享;
  3. Semaphore维护了一组许可,acquire方法获取许可,如果有可用的许可,方法返回,否则block可用看到,acquire获取到也是一个共享资源,只不过资源的数量有限制,因此它适合被设计成共享模式;
  4. ReentrantReadWriteLock提供了读写锁,写操作是独占的,读操作是可以彼此共享的,因此它同时使用了独占和共享模式;

我们总结一下,说白了就是资源的可见性,资源所有线程可见,就是共享,否则独占。我们就拿出两个来看一下,ReentrantLock的设计就是独占,上面也说了,对于ReentrantLock来说,它要访问的资源就是前继节点(为什么我们下面会说到),这个只有后继节点访问就可以了。我们再看CountDownLatch,对它来说,资源就是计数器,多个线程都在等待计数器归零,所以这个就是共享的。这么说不知道对不对,如果有偏差希望指正。

acquire

这篇文章的分析,会结合一下子类的实现,我们以ReentrantLock为例。看一下Lock方法,非公平的情况下,先直接尝试获取锁,失败了就会调用acquire方法:

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

很简单,先尝试获取,失败了返回,成功了把节点一个独占节点加入队列。我们看一下tryAcquire方法的子类实现(以后子类都默认ReentrantLock)最终会调用nonfairTryAcquire

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
//尝试更新状态
if (compareAndSetState(0, acquires)) {
//更新状态成功,就把独占线程设置为当前线程
setExclusiveOwnerThread(current);
return true;
}
} else if (current == getExclusiveOwnerThread()) {
//如果独占线程已经是当前线程,把state + acquires(其实就是加一)
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
//更新状态
setState(nextc);
return true;
}
return false;
}

这个acquires就是1,代码比较简单,不多说了,简单提一下如果独占线程就是当前线程的情况,这个情况下,会把state加1,这里就是可重入的概念,我们在释放的时候再说。

addWaiter

我们看了tryAcquireacquire,再看看如果尝试获取成功之后入队列的操作,先看addWaiter

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private Node addWaiter(Node mode) {
//根据传入的模式(独占or共享)创建Node对象,我们这里是独占
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
//这里如果不为空,说明前面还有线程在等待,尝试CAS入列,如果入列失败,则调用enq采用自选的方式入列
//这里也写了
if (pred != null) {
node.prev = pred;
//通过CAS更新tail节点
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}

我们注意下,上面在pred.next = node;这里,我们是通过CAS来更新tail,CAS成功才会进来,这是tail已经是新节点了,这时node.prev已经是旧的tail了,我们为了从头开始向后遍历,再把pred.next设置为新node

这里我们看一下enq,注释也说了,这个是通过自旋的方式入队:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
private Node enq(final Node node) {
for (;;) {
Node t = tail;
//由于采用lazy initialize,当队列为空时,需要进行初始化
if (t == null) { // Must initialize
//通过CAS设置head和tail节点
if (compareAndSetHead(new Node()))
tail = head;
} else {
//将node的前继节点设置为原tail节点
node.prev = t;
//CAS更新tail节点,更新成功则将原tail节点的后节点设置为node
if (compareAndSetTail(t, node)) {
//还是为了能从队列头开始遍历队列,设置一下next
t.next = node;
return t;
}
}
}
}

我们可以看到,AQS本质上改自自旋锁。

acquireQueued

到这addWaiter结束了,我们看一下acquireQueued

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
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
//仍然通过自旋,根据前面的逻辑,此处传入的为新入列的节点
for (;;) {
final Node p = node.predecessor();
//如果node的前一节点为head节点,而head节点为空节点,说明node是等待队列里排在最前面的节点
if (p == head && tryAcquire(arg)) {
//获取资源成功,将node设置为头节点,setHead清空节点属性thread,prev
setHead(node);
p.next = null; // help GC
failed = false;
//返回是否发生中断
return interrupted;
}
//如果acquire失败,是否要park,如果是调用parkAndCheckInterrupt
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
//只有循环中出现异常,才会进入该逻辑
cancelAcquire(node);
}
}

上面也有注释,如果需要parkparkAndCheckInterrupt最终调用LockSupport.park。看一下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
28
29
30
31
32
//shouldParkAfterFailedAcquire会将前节点的状态改为Node.SIGNAL,
//接着在下一次循环中调用parkAndCheckInterrupt阻塞线程,注意,设置为SIGNAL之后并没有返回true而是false,这就是为什么是下一次循环中才会阻塞线程
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);
//更新next指针,去掉中间取消状态的节点
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.
*/
//更新pred节点的waitStatus为SIGNAL
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}

我们看到线程自旋入队的过程中,如果失败了并且需要挂起就挂起,唤醒了再尝试,失败了再挂起。判断是否需要挂起的过程中,在特定情况下还会移除队列中取消状态的节点。

cancelAcquire

看完获取再看取消获取:

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
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;

node.thread = null;

// Skip cancelled predecessors
//获取node的前向节点
Node pred = node.prev;
//如果发现前向节点状态为CANCELLED,则继续向前找,直到找到状态正常的节点,这里跟上面的park的时候一样
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;

//如果node为tail节点,则将pred更新为tail节点
if (node == tail && compareAndSetTail(node, pred)) {
//由于pred为新的尾节点,因此将其next设为null
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
}
}

先看一下如果要cancelAcquire的节点是尾节点,那么就把尾节点(入参node)的前继节点设置为尾节点。

再看一下如果不是尾节点的情况,这里比较复杂。我们看一下if里的条件:

  1. pred不是head节点:如果pred为头节点,而node又被cancel,则node.next为等待队列中的第一个节点,需要unpark唤醒。
  2. pred节点状态为SIGNAL或能更新为SIGNAL
  3. predthread变量不能为null

我们这里来看一下SIGNAL的含义,看注释含义:

该节点的后继节点是阻塞的,所以当前节点在释放或取消时必须取消其后继,为了避免冲突,获取方法必须首先指示他们需要一个信号,然后重试原子获取,然后在失败时阻止。

再看上面的三个条件,我们就比较清楚了。如果是头结点需要取消,那么就要唤醒后面的节点。如果状态无法更新成SIGNAL,同样需要唤醒后继节点,因为根据SIGNAL节点的含义我们知道,SIGNAL节点后的节点是需要唤醒的,并且后继节点是阻塞的。如果不能更新成SIGNAL,那我们需要手动唤醒后面节点,当前节点同样不会影响后继节点去获取资源。

我们看一下unparkSuccessor(node)这里,这是唤醒node的后继节点

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;

if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
//如果节点为空或者被取消了,则从队列尾部开始查找,找到离node最近的非null且状态正常的节点
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);
}

至此,获取和取消获取的全部过程已经结束了。