AQS佇列同步器

摘要： 佇列同步器（AbstractQueuedSynchronizer）是構建鎖和其他同步元件的基礎框架。看一下原始碼的介紹： Provides a framework for implementing blocking locks and related synchronizers (sema...

佇列同步器（AbstractQueuedSynchronizer）是構建鎖和其他同步元件的基礎框架。看一下原始碼的介紹：

Provides a framework for implementing blocking locks and related synchronizers (semaphores, events, etc) that rely on first-in-first-out (FIFO) wait queues.This class is designed to be a useful basis for most kinds of synchronizers that rely on a single atomic int value to represent state.

它使用FIFO佇列來完成資源獲取執行緒的排隊工作，使用一個int變數來表示同步狀態。

佇列同步器的介面

狀態的訪問和修改

Subclasses must define the protected methods that change this state, and which define what that state means in terms of this object being acquired or released. Given these, the other methods in this class carry out all queuing and blocking mechanics. Subclasses can maintain other state fields, but only the atomically updated int value manipulated using methods getState, setState and compareAndSetState is tracked with respect to synchronization.

佇列同步器（以下簡稱AQS）使用如下三個方法來訪問和修改同步狀態：

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可重寫的方法

To use this class as the basis of a synchronizer, redefine the following methods, as applicable, by inspecting and/or modifying the synchronization state using getState, setState and/or compareAndSetState:

• tryAcquire
• tryRelease
• tryAcquireShared
• tryReleaseShared
• isHeldExclusively

AQS可重寫的5種方法預設都丟擲UnsupportedOperationException，分別為獨佔式獲取同步狀態tryAcquire、獨佔式釋放同步狀態tryRelease、共享式獲取同步狀態tryAcquireShared、共享式釋放同步狀態tryReleaseShared、當前同步器是否在獨佔模式下被執行緒佔用isHeldExclusively。

模板方法

實現自定義同步元件時將會呼叫同步器提供的模板方法。模板方法可分為三類：獨佔式獲取和釋放同步準備狀態、共享式獲取和釋放同步狀態、查詢同步佇列中的等待執行緒狀態。這些方法如下：

方法名稱	描述
void acquire(int arg)	獨佔式獲取同步狀態，如果當前執行緒獲取同步狀態成功，則由該方法返回，否則將會進入同步佇列等待。該方法將會呼叫重寫的tryAcquire(int arg)方法
void acquireInterruptibly(int arg)	與acquire(int arg)相同，但該方法響應中斷。如果該執行緒被中斷，則該方法會丟擲中斷異常並返回。
boolean tryAcquireNanos(int arg,long nanos)	在acquireInterruptibly(int arg)基礎上增加了超時限制和返回值。
void acquireShared(int arg)	共享式獲取同步狀態，如果當前執行緒未獲取到同步狀態，將會進入同步佇列等待。
void acquireSharedInterruptibly(int arg)	與acquireShared(int arg)相同，但該方法響應中斷。
boolean tryAcquireSharedNanos(int arg,long nanos)	在acquireSharedInterruptibly(int arg)基礎上增加了超時限制和返回值。
boolean release(int arg)	獨佔式的釋放同步狀態。
boolean releaseShared(int arg)	共享式的釋放同步狀態
Collection getQueuedThreads()	獲取等待在同步佇列上的執行緒集合

實現分析

同步佇列

當前執行緒獲取同步狀態失敗時，同步器會將當前執行緒以及等待狀態等資訊構造成為一個節點(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;


volatile int waitStatus;

volatile Node prev;

volatile Node next;

volatile Thread thread;

Node nextWaiter;

final boolean isShared() {
return nextWaiter == SHARED;
}

final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}

Node() {// Used to establish initial head or SHARED marker
}

Node(Thread thread, Node mode) {// Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}

Node(Thread thread, int waitStatus) { // Used by Condition
this.waitStatus = waitStatus;
this.thread = thread;
}
}

複製程式碼

等待狀態狀態共有5種，分別是0、CANCELLED、SIGNAL、CONDITION、PROPAGATE。CANCELLED表示同步佇列中等待的執行緒等待超時或者被中斷，節點進入該狀態將不再變化。SIGNAL：後繼節點的執行緒處於等待狀態，而當前節點的執行緒如果釋放了同步狀態或者被取消將會通知後繼節點。CONDITION：節點在等待佇列中，節點執行緒等待在Condition上，當其他執行緒對Condition呼叫了signal()方法後，該節點將會從等待佇列中轉移到同步佇列中。 回到上面的獨佔式狀態獲取方法acquire(int arg):

public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
複製程式碼

上述程式碼主要完成了同步狀態獲取、節點構造、加入同步佇列以及在同步佇列中自旋等待的相關工作。節點構造和加入佇列原始碼如下：

private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// 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;
}
複製程式碼

private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
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)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
複製程式碼

當前執行緒獲取同步狀態並執行了相應的邏輯之後，就需要釋放同步狀態，使得後續節點能夠繼續獲取同步狀態release(int arg)程式碼如下：

public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
複製程式碼

private void unparkSuccessor(Node node) {

int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);

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);
}
複製程式碼