java1.8 AQS AbstractQueuedSynchronizer学习
Posted zjj911
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了java1.8 AQS AbstractQueuedSynchronizer学习相关的知识，希望对你有一定的参考价值。
AQS concurrent并发包中非常重要的顶层锁类，往往用的比较多的是ReentrantLock，
然而ReentrantLock的实现依赖AbstractQueuedSynchronizer
在到上层CountDownLatch、Semaphore等不是基于ReentrantLock就是AbstractQueuedSynchronizer
so AbstractQueuedSynchronizer的重要性不言而喻，提升个人在线程方面的能力AQS是必学类之一
AbstractQueuedSynchronizer如其名，抽象的队列式的同步器
主要实现类：
ReentrantLock
ReentrantReadWriteLock
Semaphore
CountDownLatch
SynchronousQueue
一、主要内容概述
AbstractQueuedSynchronizer主要功能
1.在多线程环境下，每个线程对应一个Node节点，维持一个FIFO线程等待的双向队列，队列节点就是Node节点，在Node节点中标记当前现在的各种状态，
包括是否中断、放弃等
2.实现了完善的线程队列执行方法，包括入队等待，唤醒，跳过无效线程等，因此自定义实现的同步器只需要维护好每个线程的当前状态就可以了
3.提供两种执行模式：
独占
相当于只有一个或者一份资源，一次只能一个线程执行，队列头线程获取该资源执行，比如ReentrantLock
共享
相当于有多分线程可以让多个线程同时执行，比如CountDownLatch、Semaphore
Node节点主要点
1.执行模式
static final Node SHARED = new Node();//共享
static final Node EXCLUSIVE = null;//独占
2.线程状态标识
static final int CANCELLED = 1;
无效状态，如超时或被中断等，在队列执行到该线程节点会直接跳过并踢掉该节点
static final int SIGNAL = -1;
正常状态，代表一切正常，就等着到我执行
static final int CONDITION = -2;
Condition状态，标识线程的执行需要等待一定的Condition条件，注意此时会有两个队列Condition条件等待条件队列和全局的同步执行队列，新的条件线程节点会在等待条件队列中，
当达到条件时即其他线程调用Condition的signal()方法后，在等待条件队列中的节点获得Condition条件后会转入到全局的同步执行队列等待执行。
static final int PROPAGATE = -3;
共享状态，只与共享执行模式有关，在共享执行模式中标识结点线程处于可运行状态。
3.线程状态维护
private volatile int state; 非常重要的线程标识，控制其等于上述的值来进行各线程节点的执行，注意是volatile线程共享的
AQS主要点
1.自定义同步器实现类
上面说到AQS已经自行完成了对队列的维护，实现类只需要维护Node的状态就可以了，因此主要学习的就是AQS怎么维护队列的，而实现类主要需要实现的方法有4个
独占模式：
tryAcquire(int) 尝试获取资源，成功true，失败false。
tryRelease(int) 尝试释放资源，成功true，失败false。
共享模式：
tryAcquireShared(int) 尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
tryReleaseShared(int) 尝试释放资源，释放后允许唤醒后续等待结点返回true，否则返回false。
这个4个方法注意不是抽象方法，而直接抛出异常的有实现方法，这么干主要是为了自定义同步器实现时不需要硬性规定去实现，比如只需要独占模式，那么共享模式相关的就没必要实现
2.队列维护
维护主要是acquire获取资源-执行-release释放资源
以ReentrantLock为例其主要是lock和unlock，而lock调用的就是AQS的acquire，unlock调用release
二、队列维护的主要源码
独占模式：
1.acquire 请求资源
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
1.1 tryAcquire(arg) 尝试获取资源，注意是&&，so如果获取成功就没后面什么事了，该方法需要子类重写，AQS直接是抛异常的就不看了
1.2 addWaiter(Node.EXCLUSIVE), arg) 获取失败好说就要等了，因此加入等待队列，注意是EXCLUSIVE独占模式的节点
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode); //构建指定模式的节点，这里是EXCLUSIVE独占模式
// Try the fast path of enq; backup to full enq on failure
Node pred = tail; //赋值队尾节点
if (pred != null) { //不是空队列
node.prev = pred;
if (compareAndSetTail(pred, node)) {//CAS判断pred是不是队尾，虽然上面赋值tail，防止多线程改变队尾
pred.next = node; //新节点加到队尾
return node;
}
}
enq(node);//空队列或者上面队尾改变入队失败，入队
return node;
}
private Node enq(final Node node) {
for (;;) { //死循环+CAS 多线程 看的比较多了，
Node t = tail;
if (t == null) { // Must initialize 空队列
if (compareAndSetHead(new Node())) 确定队头是null 新建新节点入队
tail = head;
} else { //不空 或者初始化后 死循环来入队node
node.prev = t;
if (compareAndSetTail(t, node)) { //入队尾
t.next = node;
return t;
}
}
}
}
1.3 acquireQueued() 入队后保持观察看是不是到自己了
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true; 是否成功获取资源 与最后的failed相关
try {
boolean interrupted = false; 是否被中断过
for (;;) { 自旋
final Node p = node.predecessor(); 获取前节点
if (p == head && tryAcquire(arg)) { 如果前节点是首节点那马上就到自己了，尝试去获取资源
setHead(node); 说明获取资源成功了，那么首节点肯定已经处理完了或者自己放弃了，把自己设置成首节点
p.next = null; // help GC 踢掉队列中首节点，辅组GC
failed = false;
return interrupted; 返回是否中断过
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt()) 这两个方法比较重要
interrupted = true; 说明线程收到过中断信息
}
} finally {
if (failed)
failed为true说明获取资源未成功，按源码自旋理论上是不可能不成功的，除非本线程出现异常，因此这里取消本节点的资源请求等
包括1.节点状态置为取消 2.队列中剔除本节点 3.唤醒后续节点
cancelAcquire(node);
}
}
1.4 shouldParkAfterFailedAcquire(p, node) 到这里说明自己不是第二节点，应该去休眠等待，该方法主要检查队列状态，判断本线程是否真的可以休眠
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) { 说明前节点是放弃状态CANCELLED
/*
* 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.
*/
说明ws=CONDITION，PROPAGATE，也是正常状态的一种，把状态变更成待执行的SIGNAL
这里没自旋说明是允许执行失败，但其外层方法acquireQueued中是有自旋了，会再次判断的
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
1.5 parkAndCheckInterrupt() 线程休眠
允许到这里说明shouldParkAfterFailedAcquire返回true表示检查状态没问题，自己可以休眠，因此park休眠
注意这里返回的是线程中断状态，因此在acquireQueued中设置interrupted = true
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
1.6 selfInterrupt() 等待中收到过中断没处理，此时进行中断，通过acquire方法说明到这肯定是没获取到资源的，而且acquireQueued知道自己这时是首节点，
第二节点是有自旋检查的，因此直接中断没什么问题
Thread.currentThread().interrupt(); 源码没什么就是直接中断线程
cancelAcquire 在1.3中有这么一步，是在入队后有异常，取消本节点的操作。这个方法也是比较厉害的
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.
1 if (node == tail && compareAndSetTail(node, pred)) { 如果本节点在队尾，设置前节点成为队尾，可能失败，设置失败说明自己也不是队尾了
compareAndSetNext(pred, predNext, null); 设置前节点的后节点为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;
2 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 {
3 unparkSuccessor(node); 前节点是首节点，唤醒后续有效节点
}
node.next = node; // help GC
}
}
这里的踢掉即出队有3中情况
1.node是tail 直接踢掉 位置1
2.node不是tail，也不是head的后续节点即第二节点 位置2
3.node是head后续第二节点 位置3
这里是有可能存在没踢掉本节点1，2中方法就运行完了的，因为有acquireQueued的自旋和状态判断所以不需要严格保证多线程时能出队，没出队在检查状态时也会完成出队操作
unparkSuccessor 唤醒后续有效节点
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); 本节点是正常待执行节点，那就是自己执行完了，唤醒后续第二节点，本节点状态置为初始化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); 唤醒找到的后续有效第二节点
}
2.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;
}
2.1 tryRelease 依旧由实现类去实现 本类直接抛异常
理论上释放应该都是会成功的，因为是独占模式，只有本线程在运行，其他线程都阻塞着呢，也就没有线程安全问题了，直接唤醒第二节点
unparkSuccessor 上文有就不分析了
共享模式：
1.acquireShared 请求资源
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
tryAcquireShared 依旧由实现类去实现，但注意这里的返回值在AQS中已经定义好了，返回值小于0表示资源不够获取失败，运行doAcquireShared进入等待队列，大于等于0则获取成功直接执行。
1.1 doAcquireShared
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) { 大于等于0 表示获取资源成功
setHeadAndPropagate(node, r); head指向本节点，如有资源并唤醒后续节点，r剩余资源数
p.next = null; // help GC
if (interrupted) 维护队列期间受到中断，则处理中断
selfInterrupt();
failed = false;
return;
}
}
检查状态，进入阻塞
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed) 异常取消本节点
cancelAcquire(node);
}
}
与独占模式对比好像区别不大，区别在
1.selfInterrupt 位置不同，但是效果是完全一样的没区别
2.setHeadAndPropagate方法操作首节点和唤醒有些不同
1.2 setHeadAndPropagate head指向本节点，如有资源并唤醒后续节点
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below 原head结点
setHead(node); head设置为本节点
/*
* 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.
*/
条件：
1.还有剩余资源
2.原头结点释放了
3.原头结点有效
4.现头结点释放了
5.现头结点有效
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next; next 就是第二节点
if (s == null || s.isShared()) 没有后续节点，或者后续节点是共享模式
doReleaseShared(); 主要是唤醒后续，在release中详解
}
}
2.releaseShared 释放资源
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
tryReleaseShared 依旧由实现类去实现
doReleaseShared唤醒
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;
if (ws == Node.SIGNAL) { 头结点正常状态
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) 更新头结点状态为默认0
continue; // loop to recheck cases
unparkSuccessor(h); 唤醒头结点后续最近的有效节点
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) 头结点是0，置为PROPAGATE 共享状态
continue; // loop on failed CAS
}
if (h == head) // loop if head changed 头结点在上面的if操作中没被其它线程改变 这本次根据头结点的唤醒完成了break
break;
}
}
三 条件ConditionObject
条件锁也是比较常用的，其主要是用来替代Object的wait()、notify()，相比来说Condition更加安全高效，能实现的操作也更多，比如等待时间，多队列等，因此推荐使用Condition。
Condition现有的实现中只在独占模式的实现类中ReentrantLock、ReentrantReadWriteLock有用到，共享模式是不支持的
注意Condition的使用必须在锁间，即lock和unlock之前
同时在本文最开始也有提到，在AQS中使用Condition是存在两个队列的，Condition会有一个条件队列
Condition主要有3个方法：await()等待、signal()唤醒单个
await实现：
//线程进入等待 注意Condition的操作是在锁中的，因此能是执行await的肯定是执行队列的独享模式的首节点，同时也就不需要用CAS
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter(); //添加新的条件节点，即当前线程进入条件队列
int savedState = fullyRelease(node);//释放当前线程节点的同步状态，并唤醒执行队列后续节点
注意addConditionWaiter，fullyRelease的执行会在条件队列队尾加新节点，在执行队列中踢掉本节点
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
当节点没变成执行队列节点，开篇CONDITION状态时有提到。同时addConditionWaiter也能看出，实际Node上的firstWaiter、nextWaiter维护条件队列，prev、next维护执行队列
LockSupport.park(this); 本线程进入等待
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
中断类型，此时该线程肯定是被唤醒的，0表示等待期间没中断过循环继续，否则就中断过，跳出循环处理中断
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE) 说明不在条件队列，也可能是有中断但不是中断抛异常退出的类型。尝试获取同步状态，返回中断状态
interruptMode = REINTERRUPT; 能进来说明中断过，那中断类型肯定就是后响应模式
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();//踢掉无效条件节点
if (interruptMode != 0) 有中断
reportInterruptAfterWait(interruptMode); 处理中断
}
//添加新的条件节点到队尾
private Node addConditionWaiter() {
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
if (t != null && t.waitStatus != Node.CONDITION) {//条件队列队尾不空，但队尾节点状态不是CONDITION状态
unlinkCancelledWaiters();//踢掉无效节点
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION); 创建条件节点
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node; 添加到尾节点
return node;
}
//踢掉条件队列中不是CONDITION状态的所有节点
private void unlinkCancelledWaiters() {
Node t = firstWaiter;
Node trail = null;
while (t != null) {
Node next = t.nextWaiter;
if (t.waitStatus != Node.CONDITION) {//当前节点不是CONDITION状态
t.nextWaiter = null;
if (trail == null)//trail为null只有t是首节点的时候，即防止首节点不是条件节点
firstWaiter = next; //修改首节点
else
trail.nextWaiter = next;//踢掉当前节点t
if (next == null)
lastWaiter = trail;
}
else
trail = t;
t = next;
}
}
其他以await开头的方法实际主体基本一致，就不写出来了
signal()实现:
public final void signal() {
if (!isHeldExclusively()) 判断当前执行线程是不是获取到锁的线程
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first); 唤醒
}
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null) 说明是最后一个条件节点了也维护了firstWaiter
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
节点状态改为0，改失败只能是出什么异常了，节点被置位取消了
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
Node p = enq(node); //节点加到执行队列，注意节点也同时在条件队列中，不是删除添加，同时在两个队列中
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) 无效或者更新执行队列前一节点状态失败
LockSupport.unpark(node.thread); 唤醒该节点线程
return true;
}
signalAll()实际就是循环执行signal没什么好说的
四 小结
AQS是一个基本的同步队列底层，使用上ReentrantLock、ReentrantReadWriteLock、Semaphore、CountDownLatch、SynchronousQueue这些完善的并发锁类会用的比较多，学习并发锁AQS是不可或缺的
学习比较多的源码后，个人感觉CAS，自旋CAS是主体，然而往往能看晕，自己实现没有大牛级别的经验很难把各个情况考虑完善，学习之路漫漫长。
学习还是很有必要的，看懂虽说自己不能同样实现，但最基本的对JDK的这些锁类有了深入了解，使用起来起码就能更得心应手
以上是关于java1.8 AQS AbstractQueuedSynchronizer学习的主要内容，如果未能解决你的问题，请参考以下文章