Java AQS 实现——排他模式

引言

本文着重介绍 AQS 的排他模式的实现方式。所有关于 Java 并发的文章均收录于<Java并发系列文章>

排他模式

获取资源

入队操作介绍完之后,我们来看一下什么情况下需要执行入队操作,我们先从排他模式说起。下面的 acquire 是 AQS 提供的一个以排他模式获取资源的函数,我们可以看到它的执行流程是:

  1. 先尝试获取资源 tryAcquire,tryAcquire 是一个抽象函数,看完前面的锁的分类部分大家应该对它比较熟悉,因为通过 AQS 实现的各类锁实际上就是通过对 tryAcquire 这类抽象函数的覆写来达到各种锁的效果的。
  2. 如果尝试加锁失败,也就是说当前该资源已经被加锁了,就通过 addWaiter 将当前线程添加到同步队列中,注意参数是 Node.EXCLUSIVE 意味着排它锁。
  3. 加入到同步队列后,开始执行 acquireQueued 函数,猜一下应该能猜到这里面一定是进行睡眠等待的逻辑
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/**
* Acquires in exclusive mode, ignoring interrupts. Implemented
* by invoking at least once {@link #tryAcquire},
* returning on success. Otherwise the thread is queued, possibly
* repeatedly blocking and unblocking, invoking {@link
* #tryAcquire} until success. This method can be used
* to implement method {@link Lock#lock}.
*
* @param arg the acquire argument. This value is conveyed to
* {@link #tryAcquire} but is otherwise uninterpreted and
* can represent anything you like.
*/
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}

下面的就是 acquireQueued 的代码。

  1. 这里面有一个循环,它会不断地获取当前节点的前序节点,如果前序节点是 head 节点(也就是那个虚拟节点,还记得吗,虚拟节点不保存有效数据,只用作指针),这里 head 节点充当一个标志位的效果,如果一个节点的前序节点是 head,那么该节点就排在队列的第一位。
  2. 如果 p == head ,我们就需要再调用一次 tryAcquire 尝试获取锁
  3. 如果第二步成功的话,当前线程已经获得到锁了,这时候要将 head 指针进行修改,可以看到 setHead 并没有使用到 CAS 指令,因为能执行到 setHead 函数的线程相当于已经获得到同步资源了,不存在竞争
  4. 在 setHead 函数中,对当前 Node 的 thread 和 prev 进行了清除,因为这时候该 Node 已经扮演了虚拟节点的角色,有必要把虚拟节点中用不到的属性进行清除
  5. 如果 CAS 执行失败(这里有很多种可能,比如被中断了,或者是刚入队,又或者是虚假唤醒(后面介绍)),则检查是否需要继续进入睡眠,一般来说如果前序节点的状态成功改为 SIGNAL 之后(但是改成 SIGNAL 之后还会再尝试获取一次锁,失败之后才会睡眠,这是防止死等的重中之重),就可以进入等待了,SIGNAL 表明当前节点肩负着唤醒下一个节点的责任,除此之外,在检查是否需要睡眠时,如果发现前序节点的请求已经被取消,则删除该节点。
  6. 如果发现自己确实需要睡眠,则会通过 park 函数进入等待状态,因为使用的是 park 所以不需要担心,别的线程先执行唤醒之后,当前线程再进入等待的情况,因为在这种情况下 park 函数会直接返回不会进行等待
  7. 最后从等待状态中恢复过来之后,检查是否是因为中断而唤醒的,是的话,就记录一下,在返回的时候以返回值的形式告诉调用者。可见 acquire 在遇到中断时,不会抛出 InterruptedException 异常,而是循环重试。如果想要达到被中断时立即抛出异常的效果,可以使用 acquireInterruptibly, 它的实现逻辑和 acquire 基本相同,主要的区别就是 park 被中断时,会抛出 InterruptedException
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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
/**
* Acquires in exclusive uninterruptible mode for thread already in
* queue. Used by condition wait methods as well as acquire.
*/
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;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
// 永远执行不到,因为抛出的异常都被 parkAndCheckInterrupt 中的 Thread.interrupted() 吞掉了
cancelAcquire(node);
}
}

/**
* Sets head of queue to be node, thus dequeuing. Called only by
* acquire methods. Also nulls out unused fields for sake of GC
* and to suppress unnecessary signals and traversals.
*/
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}

/**
* Checks and updates status for a node that failed to acquire.
* Returns true if thread should block. This is the main signal
* control in all acquire loops. Requires that pred == node.prev.
*
* @param pred node's predecessor holding status
* @param node the node
* @return {@code true} if thread should block
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* 前序节点的状态已经是 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 {
/*
* 修改前序节点为 SIGNAL,这时候当前线程还没进入等待状态,我们需要重新判断一下现在自己是不是第一个,是的话就不用等待了
* 改成 SIGNAL 之后还会再尝试获取一次锁,失败之后才会睡眠,这是防止死等的重中之重,考虑如下情况 ThreadB-tryAcquire->ThreadA-Release->ThreadA-CheckStatus(!=SIGNAL)->ThreadB-ChangeStatus2SIGNAL
* 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;
}

/**
* Convenience method to park and then check if interrupted
*
* @return {@code true} if interrupted
*/
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}

作为延伸,我们这里带大家回顾一下 ReentrantLock 中公平锁部分对 tryAcquire 的实现。它对大家理解 AQS 如何在 CAS 修改同步队列的情况下(先修改前序指针->CAS 修改尾结点->修复后续指针),以哪种方式访问队列中的数据能够避开数据不同步的风险。简单地说,通过前序指针(prev)访问队列中的数据肯定是安全的。

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
/**
* Fair version of tryAcquire. Don't grant access unless
* recursive call or no waiters or is first.
*/
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
// 在尝试获取公平锁时,先会判断队列中是否存在前序节点。
if (!hasQueuedPredecessors() &&
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");
setState(nextc);
return true;
}
return false;
}

public final boolean hasQueuedPredecessors() {
// The correctness of this depends on head being initialized
// before tail and on head.next being accurate if the current
// thread is first in queue.
Node t = tail; // Read fields in reverse initialization order
Node h = head;
Node s;
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}

从上述代码中可以看到,在尝试获取公平锁时,先会判断队列中是否存在前序节点。之所以这么做是因为,只有发生互斥等待时,才会出现入队等待的情况,如果全是共享模式使用资源的话,队列会一直是空的,大家别急我们接下来就介绍共享模式的实现。这里我们先着重看一下 hasQueuedPredecessors 的实现,它是怎么判断有前序节点的呢:

  1. 首先如果 h != t 是说,队列的不为空,因为 head == tail 时,队列中只存在一个虚拟节点不存在实际的等待线程
  2. 在此基础上,我们还要判断一下当前持有锁的线程是不是自己,如果 head.next == null 说明有其他线程刚执行完入队的setTail工作(因为 h != t),但是前序指针还没修复,这种情况下 head.next == null,说明有别的线程已经持有锁了
  3. 另外一种可能就是 head.next != Thread.currentThread() 这时候队首持有锁的线程不是当前线程,所以存在前序节点

既然前面提到了 acquireInterruptibly 这里我们就简单地说一下,就像前面所说它和 acquire 基本相同,确实如此,从下面的代码中可以看到,它就是遇到中断时将 InterruptedException 外抛。

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
/**
* Acquires in exclusive mode, aborting if interrupted.
* Implemented by first checking interrupt status, then invoking
* at least once {@link #tryAcquire}, returning on
* success. Otherwise the thread is queued, possibly repeatedly
* blocking and unblocking, invoking {@link #tryAcquire}
* until success or the thread is interrupted. This method can be
* used to implement method {@link Lock#lockInterruptibly}.
*
*/
public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}

private void doAcquireInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.EXCLUSIVE);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
// 唯一的区别在这里,这里直接抛出异常,而 acquire 中只是记录一下flag: interrupted = true;
throw new InterruptedException();
}
} finally {
if (failed)
// 如果抛出 InterruptedException 异常,则会执行 cancelAcquire
cancelAcquire(node);
}
}

当抛出异常时,就会执行到最后面的 cancelAcquire 函数,该函数中负责将前序节点中状态为 CANCELLED 的节点清除,然后再把自己的状态变为 CANCELLED。紧接着是一些清除工作:

  • 如果当前节点是尾结点,则通过 CAS 修改尾结点即可。如果 CAS 失败了也不要紧,因为当前线程状态已经是 CANCELLED 了所以其他线程会把自己清除
  • 如果当前节点不是第一个节点,即 pred != head ,我们要保证前序节点的状态是 SIGNAL,并且前序节点的线程不是当前线程,因为自己不是尾结点,所以自己当前的状态很可能就是 SIGNAL,所以这里我们无论如何要确保前序节点的状态能够修改为 SIGNAL,如果做到了,就可以大胆地通过 CAS 将自己从队列中移除
  • 否则,说明自己可能是头结点,或者前序节点都取消了,也有可能前序节点的线程就是当前线程,那么就只能由自己来唤醒后续的线程了
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
/**
* Cancels an ongoing attempt to acquire.
*
* @param node the node
*/
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;

node.thread = null;

// Skip cancelled predecessors
Node pred = node.prev;
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;

// If we are the tail, remove ourselves.
if (node == tail && compareAndSetTail(node, pred)) {
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
}
}

唤醒的过程也很简单:

  1. 如果当前节点状态不是 CANCELLED 就清除状态
  2. 然后先看一下 next 指针指向的节点是否需要被唤醒,next == null 代表了可能发生的前序指针和后续指针不同步,s.waitStatus > 0 表示后继节点已被取消,这时候我们就需要找到下一个需要被唤醒的节点。
  3. 我们需要从尾结点出发,逐个向前找,因为前序指针肯定是安全的
  4. 如果找到了需要被唤醒的线程,就执行 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
27
28
29
30
31
/**
* Wakes up node's successor, if one exists.
*
* @param node the 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);
}

释放资源

介绍完资源的获取,我们再来看看资源的释放流程。

  1. 首先,调用 tryRelease 函数,它也是一个抽象函数,在上层实现中一般会进行必要的检查,比如检查持有锁的线程是否是当前线程等,就比如如下 ReentrantLock 中对 tryRelease 的实现。
  2. 释放成功后,判断当前同步队列是否为空,不为空并且当前线程承担唤醒职责时(waitStatus < 0),因为当前线程能够成功执行 tryRelease,所以当前线程的 waitStatus 不会是 CANCELLED,剩下的状态 SIGNAL 是需要承担唤醒职责的,CONDITION 和 PROPAGATE 我们后面介绍。
  3. 具体的唤醒函数 unparkSuccessor 我们前面刚介绍过,这里就不再赘述了
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
/**
* Releases in exclusive mode. Implemented by unblocking one or
* more threads if {@link #tryRelease} returns true.
* This method can be used to implement method {@link Lock#unlock}.
*
* @param arg the release argument. This value is conveyed to
* {@link #tryRelease} but is otherwise uninterpreted and
* can represent anything you like.
* @return the value returned from {@link #tryRelease}
*/
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) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}

AQS 框架使用

看到这里,不知道大家对 AQS 的使用流程有没有体会,我的理解是在使用它时,一般我们需要覆写 tryAcquire 和 tryRelease 这类函数,它们是直接修改互斥资源 state 的函数,AQS 通过将具体的状态修改职责移交到子类中,能让子类实现各种类型的锁,就比如前一章我们介绍的那些。

而子类向外暴露 lock 和 unlock 函数时,又直接使用 AQS 中的 acquire 和 releas 函数,因为这些函数中封装了尝试加锁过程和加锁失败入队等待过程。

1
2
public void lock()        { acquire(1); }
public void unlock() { release(1); }

总结一下就是 AQS 对变化报开放的态度,你可以通过它完成各种同步策略,同时对与那些样板代码,都已经被它封装在了自己内部,并使用 final 关键字修饰,例如 acquire 和 release。

参考内容

[1] linux 2.6 互斥锁的实现-源码分析
[2] 深入解析条件变量(condition variables)
[3] Linux下Condition Vairable和Mutext合用的小细节
[4] 从ReentrantLock的实现看AQS的原理及应用
[5] 不可不说的Java“锁”事
[6] 从源码层面解析yield、sleep、wait、park
[7] LockSupport中的park与unpark原理
[8] Thread.sleep、Object.wait、LockSupport.park 区别
[9] 从AQS到futex-二-JVM的Thread和Parker
[10] Java的LockSupport.park()实现分析
[11] JVM源码分析之Object.wait/notify实现
[12] Java线程源码解析之interrupt
[13] Thread.interrupt()相关源码分析
[14] Java CAS 原理剖析
[15] 源码解析 Java 的 compareAndSwapObject 到底比较的是什么
[16] 《Java并发编程的艺术》
[17] 《实战 Java 高并发程序设计》
[18] volatile关键字深入学习
[19] 为什么Netty的FastThreadLocal速度快
[20] 线程池ThreadPoolExecutor实现原理
[21] 深入理解Java线程池:ThreadPoolExecutor
[22] ConcurrentHashMap 详解一
[23] ConcurrentHashMap 详解二
[24] JUC中Atomic class之lazySet的一点疑惑
[25] The JSR-133 Cookbook for Compiler Writers
[26] 就是要你懂Java中volatile关键字实现原理

贝克街的流浪猫 wechat
您的打赏将鼓励我继续分享!
  • 本文作者: 贝克街的流浪猫
  • 本文链接: https://www.beikejiedeliulangmao.top/java/concurrent/aqs-exclusive-mode/
  • 版权声明: 本博客所有文章除特别声明外,均采用 BY-NC-SA 许可协议。转载请注明出处!
  • 创作声明: 本文基于上述所有参考内容进行创作,其中可能涉及复制、修改或者转换,图片均来自网络,如有侵权请联系我,我会第一时间进行删除。