一、AQS概述
AQS全名AbstractQueuedSynchronizer,意為抽象隊列同步器，JUC(java.util.concurrent包)下面的Lock和其他一些并發(fā)工具類都是基于它來實現(xiàn)的。AQS維護了一個volatile的state和一個CLH（FIFO）雙向隊列。
二、分析
state
state是一個由volatile修飾的int變量，它的訪問方式有三種：
- getState()
- setState(int newState)
- compareAndSetState(int expect, int update)
/**
* 由volatile修飾的state
*/
private volatile int state;
/**
* 基于內(nèi)存可見性的讀
*/
protected final int getState() {
return state;
}
/**
* 基于內(nèi)存可見性的寫
*/
protected final void setState(int newState) {
state = newState;
}
/**
* 使用CAS+volatile，基于原子性與可見性的對state進行設值
*/
protected final boolean compareAndSetState(int expect, int update) {
// 使用Unsafe類，調(diào)用JNI方法
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
資源獲取主要有兩種形式：
- 獨占式（EXCLUSIVE）
僅有一個線程能在同一時刻獲取到資源并處理，如ReentrantLock的實現(xiàn)。
- 共享式（SHARED）
多個線程可以同時獲取到資源并處理，如Semaphore/CountDownLatch等。
AQS中大部分邏輯已經(jīng)被實現(xiàn)，集成類只需要重寫state的獲取（acquire）與釋放(release)方法，因為在AQS中，這些方法默認定義的實現(xiàn)方式都是拋出不支持操作異常，所以按需實現(xiàn)即可。
其中需要繼承類重寫的方法有：
- tryAcquire(int arg)
此方法是獨占式的獲取資源方法，成功則返回true,失敗返回false。
- tryRelease(int arg)
此方法是獨占式的釋放資源方法，成功則返回true,失敗返回false。
- tryAcquireShared(int arg)
此方法是共享式的獲取資源方法，返回負數(shù)表示失敗，0表示獲取成功，但是沒有可用資源，正數(shù)表示獲取成功，且有可用資源。
- tryReleaseShared(int arg)
此方法是共享式的釋放資源方法，如果允許喚醒后續(xù)等待線程則返回true,不允許則返回false。
- isHeldExclusively()
判斷當前線程是否正在獨享資源，是則返回true,否則返回false。
CLH(FIFO)隊列
AQS中是通過內(nèi)部類Node來維護一個CLH隊列的。源碼如下：
static final class Node {
/** 標記共享式訪問 */
static final Node SHARED = new Node();
/** 標記獨占式訪問 */
static final Node EXCLUSIVE = null;
/** 字段waitStatus的值，表示當前節(jié)點已取消等待 */
static final int CANCELLED = 1;
/**字段waitStatus的值，表示當前節(jié)點取消或釋放資源后，通知下一個節(jié)點 */
static final int SIGNAL = -1;
/** 表示正在等待觸發(fā)條件 */
static final int CONDITION = -2;
/**
* 表示下一個共享獲取應無條件傳播
*/
static final int PROPAGATE = -3;
/**
* Status field, taking on only the values:
* SIGNAL: 表示當前節(jié)點取消或釋放資源后，通知下一個節(jié)點
* CANCELLED: 表示當前節(jié)點已取消等待
* CONDITION: 表示正在等待觸發(fā)條件
* PROPAGATE: 表示下一個共享獲取應無條件傳播
*/
volatile int waitStatus;
/**
* 前節(jié)點
*/
volatile Node prev;
/**
* 下一個節(jié)點
*/
volatile Node next;
/**
* 節(jié)點對應線程
*/
volatile Thread thread;
/**
* 下一個等待的節(jié)點
*/
Node nextWaiter;
/**
* 是否是共享式訪問
*/
final boolean isShared() {
return nextWaiter == SHARED;
}
/**
* 返回前節(jié)點
*/
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
Node() { // 共享式訪問的構(gòu)造函數(shù)
}
Node(Thread thread, Node mode) { // 用于被添加等待者使用
this.nextWaiter = mode;
this.thread = thread;
}
Node(Thread thread, int waitStatus) { // 用于Condition使用
this.waitStatus = waitStatus;
this.thread = thread;
}
}
獨占模式-獲取資源
使用AQS中的acquire(int arg)方法
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
該方法分為4個部分：
- tryAcquire()
需要自己實現(xiàn)的方法，如果獲取到資源使用權(quán)，則返回true,反之fasle。如果獲取到資源，返回true,!true為false,根據(jù)&&的短路性，則不會執(zhí)行后續(xù)方法，直接跳過程序。如果未獲取到資源，返回false,!false為true,則進入后續(xù)方法。
- addWaiter()
如果未獲取到資源使用權(quán)，則首先會調(diào)用此方法。上源碼：
private Node addWaiter(Node mode) {
// 封裝當前線程和獨占模式
Node node = new Node(Thread.currentThread(), mode);
// 獲取尾部節(jié)點
Node pred = tail;
if (pred != null) {
node.prev = pred;
// CAS設置尾部節(jié)點
if (compareAndSetTail(pred, node)) {
// 將為節(jié)點的下一節(jié)點指向當前node
pred.next = node;
return node;
}
}
// 如果尾結(jié)點為空或者設置尾結(jié)點失敗
enq(node);
return node;
}
private Node enq(final Node node) {
// 如果CAS設置未成功則死循環(huán)
for (;;) {
// 獲得尾結(jié)點
Node t = tail;
// 如果尾節(jié)點為空，說明CLH隊列為空，需要初始化
if (t == null) {
if (compareAndSetHead(new Node()))
tail = head;
} else {
// 設置當前節(jié)點的前驅(qū)節(jié)點
node.prev = t;
// CAS設置當前節(jié)點為尾結(jié)點
if (compareAndSetTail(t, node)) {
// 設置后驅(qū)節(jié)點
t.next = node;
return t;
}
}
}
}
- acquiredQueued()
final boolean acquireQueued(final Node node, int arg) {
// 標識資源獲取是否失敗
boolean failed = true;
try {
// 標識線程是否中斷
boolean interrupted = false;
for (;;) {
// 獲得當前節(jié)點的前驅(qū)節(jié)點
final Node p = node.predecessor();
// 如果前驅(qū)節(jié)點為頭結(jié)點，說明快到當前節(jié)點了，嘗試獲取資源
if (p == head && tryAcquire(arg)) {
// 獲取資源成功
// 設置當前節(jié)點為頭結(jié)點
setHead(node);
// 取消前驅(qū)節(jié)點（以前的頭部）的后節(jié)點，方便GC回收
p.next = null; // help GC
// 標識未失敗
failed = false;
// 返回中斷標志
return interrupted;
}
// 如果當前節(jié)點的前驅(qū)節(jié)點不是頭結(jié)點或獲取資源失敗
// 需要用shouldParkAfterFailedAcquire函數(shù)判斷是否需要阻塞該節(jié)點持有的線程
// 如果需要阻塞，則執(zhí)行parkAndCheckInterrupt方法，并設置被中斷
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
// 如果最終獲取資源失敗
if (failed)
// 當前節(jié)點取消獲取資源
cancelAcquire(node);
}
}
- selfInterrupt()
中斷當前線程
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
獨占模式-釋放資源
release() 釋放資源并喚醒后繼線程
public final boolean release(int arg) {
if (tryRelease(arg)) {
// 獲取頭結(jié)點
Node h = head;
// 頭結(jié)點不為空且等待狀態(tài)值不為0
if (h != null && h.waitStatus != 0)
// 喚醒后續(xù)等待線程
unparkSuccessor(h);
return true;
}
return false;
}
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
// 如果等待狀態(tài)值小于0
if (ws < 0)
// 使用CAS將waitStatus設置為0
compareAndSetWaitStatus(node, ws, 0);
Node s = node.next;
// 如果當前節(jié)點沒有后繼節(jié)點或者后繼節(jié)點放棄競爭資源
if (s == null || s.waitStatus > 0) {
s = null;
// 從隊列尾部循環(huán)直到當前節(jié)點，找到最近的且等待狀態(tài)值小于0的節(jié)點
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
// 如果找到的后繼節(jié)點不為空,則喚醒其持有的線程
if (s != null)
LockSupport.unpark(s.thread);
}
