1、AQS是AbstractQueuedSynchronizer的简称。提供用于实现阻塞锁和同步器框架（信号量、事件等），依靠先入先出（FIFO）等待队列。AQS为一系列同步器依赖于一个单独的原子变量（state）的同步器提供了一个非常有用的基础。AQS对于state的操作都是基于CAS操作，保证了state的原子性和可见性。
state使用
AQS 提供了三种操作state的方法
- getState()
- setState(int)
- compareAndSetState(int, int)
具体源码如下
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);
}
AQS资源
AQS定义了两种获取资源的方式，独占（exclusive）和共享（Shared）两种方式，使用AQS值需要子类实现了以下方法的一部分或全部，如果没有实现而调用会报UnsupportedOperationException异常
- tryAcquire(int) 尝试获取独占资源
- tryRelease(int) 尝试释放独占资源
- tryAcquireShared(int) 尝试获取共享资源
- tryReleaseShared(int) 尝试释放共享资源
- isHeldExclusively() 该线程是否正在独占资源。只有用到condition才需要去实现它。
AQS的其他重要方法
- acquire(int arg) 在独占模式中获得，忽略中断。
- 调用自定义同步器的tryAcquire()尝试直接去获取资源，如果成功则直接返回；
- 没成功，则addWaiter()将该线程加入等待队列的尾部，并标记为独占模式；
- acquireQueued()使线程在等待队列中休息，有机会时（轮到自己，会被unpark()）会去尝试获取资源。获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。
- 如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。
public final void acquire(int arg) {
if (!tryAcquire(arg) && // 如果没有拿到锁
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
//提供中断当前线程的方法
selfInterrupt();
}
/**将当前线程加入到等待队列的队尾，并返回代表当前线程的节点
* Creates and enqueues node for current thread and given mode.
*/
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);//EXCLUSIVE（独占）和SHARED（共享）
// 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)) {
pred.next = node;
return node;
}
}
enq(node);// 按正常逻辑入队列
return node;
}
/**将node加入队尾
* Inserts node into queue, initializing if necessary. See picture above.
* @param node the node to insert
* @return node's predecessor
*/
private Node enq(final Node node) {
for (;;) { // CAS + 循环 = 自旋
Node t = tail;
if (t == null) { // Must initialize 队列为空
if (compareAndSetHead(new Node()))// 创建一个空的标志结点作为head结点
tail = head; // tail 和 head都是同一个节点
} else {
node.prev = t;
if (compareAndSetTail(t, node)) { // 放入tail尾部
t.next = node;
return t;
}
}
}
}
/**
* 线程阻塞等待获取锁
*/
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)) { // 如果前一个节点是head，就尝试抢锁，并且尝试抢锁成功
setHead(node); // 更换head，即当前Node出队列
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&// 检查状态，是否需要挂起线程
parkAndCheckInterrupt()) // 挂起
interrupted = true; // 如果出现中断，则修改标记
}
} finally {
if (failed)
cancelAcquire(node);
}
}
- release(int) 释放独占模式。
- 一般来说，独占都是可以释放成功，但是得判断该线程释放完全释放
public final boolean release(int arg) {
if (tryRelease(arg)) {//尝试释放
Node h = head; // 从头开始找
if (h != null && h.waitStatus != 0)
unparkSuccessor(h); // 唤醒下一个线程
return true;
}
return false;
}
/** 唤醒等待者
* 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);
}
- acquireShared(int) 在共享模式中获取，忽略中断。
- 只要没有独占资源，该方法都应该成功，需要注意唤醒线程时需要判断队列中的下个线程是否时共享资源，如果时，也要同时唤醒。
/** 线程获取共享资源的入口
*/
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0) // 判断量够不够
doAcquireShared(arg); // 没拿到资源，需要等待
}
/**等待..
* Acquires in shared uninterruptible mode.
* @param arg the acquire argument
*/
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) { // 如果前置为head
int r = tryAcquireShared(arg); // 尝试获取资源,返回资源剩余的数量
if (r >= 0) { // 拿到资源
setHeadAndPropagate(node, r); // 修改head节点
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
- releaseShared(int)释放共享资源
- 释放共享资源都能成功
/** 线程释放共享资源
*/
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {//尝试
doReleaseShared();//执行
return true;
}
return false;
}
/** 共享模式下 - 唤醒当前head节点的后续节点
*/
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) { // 判定是否还有后续节点
int ws = h.waitStatus;
if (ws == Node.SIGNAL) { // 如果状态为SIGNAL，代表需要通知后续节点
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))// 修改状态为0，通知一次
continue; // loop to recheck cases 修改失败，代表已经通知，继续处理
unparkSuccessor(h); // 唤醒
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) // 通知过后，修改节点状态为PROPAGATE
continue; // loop on failed CAS
}
if (h == head) // loop if head changed 知道其他的节点，把这个head挤下来，它才跳出循环
break;
}
}