一、什么是AQS
AQS：抽象队列同步器(AbstractQueuedSynchronizer)，是并发容器（java.util.concurrent）下locks包内的一个类。提供了一种实现阻塞锁和一系列依赖FIFO等待队列的同步器。
AQS作为并发包java.util.concurrent的基础类，很多并发类都继承了他。
并发包中继承AQS类关系
java8上AQS源码加注释超过了两千行，搜索引擎找到的解释很多都是密密麻麻的源码介绍，看上去就很头痛。下面我通过图文结合的方式结合其他文章详细描述下AQS。在阅读之前最好要再熟悉下设计模式中的模板方法，以及CAS，双向链表、Park的内容。有助于后续理解AQS。
模板方法模式介绍
CAS介绍
AQS双向链表
并发中的LockSupport工具类(Park、unPark)
二、AQS内部类Node属性说明
AQS有一个非常重要的静态内部类Node，如下图所示：
Node类
static final class Node {
// 共享模式和独占模式的标记
static final Node SHARED = new Node();
static final Node EXCLUSIVE = null;
// 四种状态.
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
/*
status属性用于表示状态：
1）CANCELLED 和 CONDITION 状态表示当前 Node 的状态；
2）SIGNAL 和 PROPAGATE 状态表示的是下一个 Node 的状态
1. SIGNAL：当前 Node 的后继 Node 将被阻塞。当前 Node 释放锁或
者被取消时，它要将后继 Node 唤醒；
2. CANCELLED：由于超时或者中断，该 Node 中的线程处于取消状态。
一旦 Node 处于取消状态，它不能再变成其他状态，取消状态的 Node 不会被阻塞；
3. CONDITION： Node 处于条件队列上。调用 await 时，会使用 nextWaiter 来链接，
不在 CLH 队列中。当调用 transferred 后，如果该状态被设置为0，Node 才会进入 CLH
队列；
4. PROPAGATE：releaseShared 操作应传播给其他的所有 Node。当 setHead 完成后，执行传播
操作；
5. 状态0表示不是以上的状态。Node 创建时的默认状态。
*/
volatile int waitStatus;
// 前驱节点
volatile Node prev;
// 后继节点
volatile Node next;
// 获取同步状态的线程
volatile Thread thread;
/**
Node既可以作为同步队列节点使用，也可以作为Condition的等待队列节点使用
在作为同步队列节点时，nextWaiter可能有两个值：EXCLUSIVE、SHARED标识当前节点是
独占模式还是共享模式；
在作为等待队列节点使用时，nextWaiter保存后继节点。
*/
Node nextWaiter;
内部类Node等待模式和状态介绍
三、加锁过程
3.1 线程加锁过程说明
假设现在有线程T1、T2、T3三个线程，尝试获取锁，我们把不同情况分别分析一下。jdk版本为1.8，以公平锁为例进行演示。
ReentrantLock lock = new ReentrantLock(true);
// 加锁
lock.lock();
try {
// 临界区资源........
} finally {
//释放锁
lock.unlock();
}
1.T1上来就获取到锁
整个加锁过程运行代码如下：
只有线程T1尝试加锁流程
// 1.调用lock方法.
lock.lock();
public void lock() {
sync.lock();
}
final void lock() {
acquire(1);
}
public final void acquire(int arg) {
// 执行tryAcquire方法尝试获取锁
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
protected final boolean tryAcquire(int acquires) {
// 获取当前线程
final Thread current = Thread.currentThread();
// 此时线程只有T1，没有其他线程加锁，所以c=0
int c = getState();
if (c == 0) {
// 同样的没有队列存在hasQueuedPredecessors返回false，
// !hasQueuedPredecessors()为true。 cas比较(0,accquires)期望值
// 为0，accquires=1同样为true。
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
// 将当前线程设置为排他锁模式
setExclusiveOwnerThread(current);
// 获取锁成功，流程结束。
return true;
}
}
}
2.T1持有锁时，T2尝试获取锁
新增节点
线程T2获取锁步骤
前面代码同上，省略
public final void acquire(int arg) {
// T2尝试获取锁
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
protected final boolean tryAcquire(int acquires) {
// 线程T2
final Thread current = Thread.currentThread();
// 此时线程T1持有锁，c=1
int c = getState();
if (c == 0) {
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
// current是T2，getExclusiveOwnerThread()现在还是T1
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
// 返回false，则!tryAccquire即为true，比较acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法
return false;
}
// 仔细看下addWaiter就是上面aqs双向链表图示的代码.
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// pred 赋值尾节点，此时还没有队列呢pred=null；走enq(node)方法
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
private Node enq(final Node node) {
// 自旋
for (;;) {
// 第一次循环，同样的t=null;
// 第二次循环，初始化节点后，t=tail不为空了，走else
Node t = tail;
if (t == null) {
// 走if条件cas设置头.
if (compareAndSetHead(new Node()))
// tail=head，就是初始化了一个头尾指向一个thread=null的节点。初始化出来了第一个节点。
tail = head;
} else {
// 第二次循环，设置T2的pre节点指向thread=null.
node.prev = t;
// cas比较，移动尾部指向
if (compareAndSetTail(t, node)) {
// thread=null node 指向T2。可以参考![新增节点图]再熟悉下整个过程.
t.next = node;
return t;
}
}
}
}
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// 找当前线程的前驱节点p
final Node p = node.predecessor();
// 现在队列中只有一个线程，前驱节点为thread=null的head节点
// p == head为true,tryAcquire再次尝试获取锁，如果此时T1释放
//锁了，还没有调用unlock().再次尝试tryAcquire为true
//设置T2为头节点,头结点p的next指向为null，双向链表断开
// 如果tryAcquire为false没有获取到锁，执行下面的if方法.
// 第二次重复第一次的操作.
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 第一次shouldParkAfterFailedAcquire返回false.继续自旋.
// 第二次调用if方法，shouldParkAfterFailedAcquire返回true
// 调用parkAndCheckInterrupt方法，线程等待。直到线程T1调用unlock方法后，唤醒该队列中的第一个线程。
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
// 找当前线程的前驱节点
final Node predecessor() throws NullPointerException {
//
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// 初次ws=0;
// 第二次自旋后，ws值经过第一次cas设置为了-1
int ws = pred.waitStatus;
// return true
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// cas将pred的waitStatus值设为阻塞-1
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
// 第一次返回false;
return false;
}
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
private void cancelAcquire(Node node) {
if (node == null)
return;
node.thread = null;
Node pred = node.prev;
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;
Node predNext = pred.next;
node.waitStatus = Node.CANCELLED;
if (node == tail && compareAndSetTail(node, pred)) {
compareAndSetNext(pred, predNext, null);
} else {
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
}
}
队列中有些条件没有再详细说明，先熟悉基本后再结合一些视频和书籍知识点。才可以更加清楚。否则上来就去看源码往往会一头扎进去很久都出不来。
网友评论