Java并发包源码学习系列：AQS共享式与独占式获取与释放资源的区别
2016 年 6 月 22 日
目录
- Java并发包源码学习系列：AQS共享模式获取与释放资源
-
- void acquire(int arg)
- boolean acquireQueued(Node, int)
-
- boolean release(int arg)
- void unparkSuccessor(Node node)
-
- void acquireShared(int arg)
- void doAcquireShared(int arg)
- void setHeadAndPropagate(Node node, int propagate)
-
- boolean releaseShared(int arg)
- 独占式和共享式的区别总结
-
Java并发包源码学习系列：AQS共享模式获取与释放资源
往期回顾：
上一篇文章介绍了AQS内置队列节点的出队入队操作，以及独占式获取共享资源与释放资源的详细流程，为了结构完整，本篇继续以AQS的角度介绍另外一种：共享模式获取与释放资源的细节，本篇暂不分析具体子类如ReentrantLock、ReentrantReadWriteLock的实现，之后会陆续补充。
独占式获取资源
友情提示：本篇文章着重介绍共享模式获取和释放资源的特点，许多代码实现上面和共享式和独占式其实逻辑差不多，为了清晰对比，这边会将独占式的部分核心代码粘贴过来，注意理解共享式和独占式存在差异的地方。详细解析可戳： Java并发包源码学习系列：CLH同步队列及同步资源获取与释放
void acquire(int arg)
public final void acquire(int arg) {
if (!tryAcquire(arg) && // tryAcquire由子类实现，表示获取锁，如果成功，这个方法直接返回了
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 如果获取失败，执行
selfInterrupt();
}
boolean acquireQueued(Node, int)
// 这个方法如果返回true，代码将进入selfInterrupt()
final boolean acquireQueued(final Node node, int arg) {
// 注意默认为true
boolean failed = true;
try {
// 是否中断
boolean interrupted = false;
// 自旋，即死循环
for (;;) {
// 得到node的前驱节点
final Node p = node.predecessor();
// 我们知道head是虚拟的头节点，p==head表示如果node为阻塞队列的第一个真实节点
// 就执行tryAcquire逻辑，这里tryAcquire也需要由子类实现
if (p == head && tryAcquire(arg)) {
// tryAcquire获取成功走到这，执行setHead出队操作
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 走到这有两种情况 1.node不是第一个节点 2.tryAcquire争夺锁失败了
// 这里就判断 如果当前线程争锁失败，是否需要挂起当前这个线程
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
// 死循环退出，只有tryAcquire获取锁失败的时候failed才为true
if (failed)
cancelAcquire(node);
}
}
独占式释放资源
boolean release(int arg)
public final boolean release(int arg) {
if (tryRelease(arg)) { // 子类实现tryRelease方法
// 获得当前head
Node h = head;
// head不为null并且head的等待状态不为0
if (h != null && h.waitStatus != 0)
// 唤醒下一个可以被唤醒的线程，不一定是next哦
unparkSuccessor(h);
return true;
}
return false;
}
void unparkSuccessor(Node 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;
// 如果node的waitStatus<0为signal，CAS修改为0
// 将 head 节点的 ws 改成 0，清除信号。表示，他已经释放过了。不能重复释放。
if (ws 0) {
s = null;
// 从队尾往前找，找到没有没取消的所有节点排在最前面的【直到t为null或t==node才退出循环嘛】
for (Node t = tail; t != null && t != node; t = t.prev)
// 如果>0表示节点被取消了，就一直向前找呗，找到之后不会return，还会一直向前
if (t.waitStatus <= 0)
s = t;
}
// 如果后继节点存在且没有被取消，会走到这，直接唤醒后继节点即可
if (s != null)
LockSupport.unpark(s.thread);
}
共享式获取资源
void acquireShared(int arg)
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0) //子类实现
doAcquireShared(arg);
}
-
tryAcquireShared(int)是AQS提供给子类实现的钩子方法，子类可以自定义实现 共享式获取资源 的方式，获取状态失败返回小于0，返回零值表示被独占方式获取，返回正值表示共享方式获取。 - 如果获取失败，则进入
doAcquireShared(arg);的逻辑。
void doAcquireShared(int arg)
注意这里和独占式获取资源 acquireQueued 的区别。
private void doAcquireShared(int arg) {
// 包装成共享模式的节点，入队
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
// 自旋
for (;;) {
final Node p = node.predecessor();
if (p == head) {
// 尝试获取同步状态，子类实现
int r = tryAcquireShared(arg);
if (r >= 0) {
// 设置新的首节点，并根据条件，唤醒下一个节点
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
我们可以看到有几个存在差异的地方：
setHeadAndPropagate(node, r)
接下来我们着重看下setHeadAndPropagate方法。
void setHeadAndPropagate(Node node, int propagate)
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
// 节点出队，设置node为新的head
setHead(node);
/*
* Try to signal next queued node if:
* Propagation was indicated by caller,
* or was recorded (as h.waitStatus either before
* or after setHead) by a previous operation
* (note: this uses sign-check of waitStatus because
* PROPAGATE status may transition to SIGNAL.)
* and
* The next node is waiting in shared mode,
* or we don't know, because it appears null
*
* The conservatism in both of these checks may cause
* unnecessary wake-ups, but only when there are multiple
* racing acquires/releases, so most need signals now or soon
* anyway.
*/
// 这个方法进来的时候propagate>=0
// propagate>0表示同步状态还可以被后面的节点获取
// h指向原先的head节点，之后h = head，h表示新的head节点
// h.waitStatus 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
// 获取下一个节点
Node s = node.next;
// 没有下一个节点或下一个节点为共享式获取状态
if (s == null || s.isShared())
// 唤醒后续的共享式获取同步状态的节点
doReleaseShared();
}
}
- 先记录一下原来的头节点，然后设置node为新的头节点。
- 原先的头节点或新的头节点等待状态是propagate或signal，可以继续向下唤醒。
- 如果判断下个节点为shared节点，调用共享式释放资源方法唤醒后续节点。
共享式释放资源
boolean releaseShared(int arg)
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) { // 子类实现
doReleaseShared();
return true;
}
return false;
}
doReleaseShared()
可以发现共享模式下，无论是获取资源还是释放资源都调用了doReleaseShared方法，可见该方法是共享模式释放资源唤醒节点的核心方法，主要功能是 唤醒下一个线程或者设置传播状态 。
后继线程被唤醒后，会尝试获取共享锁，如果成功之后，则又会调用setHeadAndPropagate,将唤醒传播下去。这个方法的作用是保障在acquire和release存在竞争的情况下，保证队列中处于等待状态的节点能够有办法被唤醒。
private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases. This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
// 自旋
for (;;) {
Node h = head;
// 队列已经初始化且至少有一个节点
if (h != null && h != tail) {
int ws = h.waitStatus;
// 无论是独占还是共享，只有节点的ws为signal的时候，才会在释放的时候，唤醒后面的节点
if (ws == Node.SIGNAL) {
// cas将ws设置为0，设置失败，将会继续从循环开始
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
// 唤醒后继节点，unparkSuccessor这个方法上面已经解析过
unparkSuccessor(h);
}
// 如果ws为0，则更新状态为propagate，
// 之后setHeadAndPropagate读到ws<0的时候，会继续唤醒后面节点
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
// 如果head在这个过程中被更改了，会继续自旋
if (h == head) // loop if head changed
break;
}
}
该方法在 head 节点存在后继节点的情况下，做了两件事情：
-
如果 head 节点等待状态为 SIGNAL，则将 head 节点状态设为 0，并唤醒后继未取消节点。
-
如果 head 节点等待状态为 0，则将 head 节点状态设为 PROPAGATE，保证唤醒能够正常传播下去。
设置PROPAGATE的作用：PROPAGATE状态用在[setHeadAndPropagate](#void setHeadAndPropagate(Node node, int propagate)) ,当头节点状态被设为 PROPAGATE 后，后继节点成为新的头结点后。若 propagate > 0 条件不成立，则根据条件 h.waitStatus < 0 成立与否，来决定是否唤醒后继节点，即向后传播唤醒动作。
引入PROPAGATE是为了解决什么问题？
AbstractQueuedSynchronizer源码解读 ，强烈建议阅读这篇博客。
独占式和共享式的区别总结
共享式获取与独占式获取最大的区别就是同一时刻能否有多个线程同时获取到同步状态。
- 共享式访问资源时，同一时刻其他共享式的访问会被允许。
- 独占式访问资源时，同一时刻其他访问均被阻塞。
AQS都提供了子类实现的钩子方法，独占式的代表方法有：tryAcquire和tryRelease以及isHeldExclusively方法，共享式的代表方法有：tryAcquireShared和tryReleaseShared方法。
AQS中获取操作和释放操作的标准形式：
boolean acquire() throws InterruptedException{
while( 当前状态不允许获取操作 ){
if( 需要阻塞获取请求){
如果当前线程不在队列中，则将其插入队列
阻塞当前线程
}else{
返回失败
}
}
可能更新同步器的状态
如果线程位于队列中，则将其移除队列
返回成功
}
void release(){
更新同步器的状态
if( 新的状态允许某个被阻塞的线程获取成功 ){
解除队列中一个或多个线程的阻塞状态
}
}
图源：《并发编程的艺术》下图是独占式同步状态获取的流程
当某个线程争夺同步资源失败之后，他们都会将线程包装为节点，并加入CLH同步队列的队尾，并保持自旋，一个是 addWaiter(Node.EXCLUSIVE) ，一个是 addWaiter(Node.EXCLUSIVE) 。
同步队列中的线程在自旋时会判断其前驱节点是否为首节点，如果是首节点 node.predecessor() == head ，他们都会尝试获取同步状态，只不过：
- 独占式获取状态成功后，只会出队一个节点。
- 共享式获取状态成功后，除了出队一个节点，还会唤醒后面的节点。
线程执行完逻辑之后，他们都会释放同步状态，释放之后将会 unparkSuccessor(h) 唤醒其后可被唤醒的某个后继节点。
参考阅读
- 【死磕Java并发】—–J.U.C之AQS：同步状态的获取与释放
- Java并发之AQS详解
- AbstractQueuedSynchronizer源码解读
- Java技术之AQS详解
- 《并发编程的艺术》方腾飞
- 《Java并发编程实战》Doug Lea
About The Author
