A counting semaphore. Conceptually, a semaphore maintains a set of permits. Each {@link #acquire} blocks if necessary until a permit isavailable, and then takes it. Each {@link #release} adds a permit,potentially releasing a blocking acquirer.
一組數量的訊號，只有獲取到訊號的執行緒才允許執行。通過acquire進行獲取，如果獲取不到則需要阻塞等待直到一個訊號可用。release會釋放一個號誌。通過這種方式可以實現限流。
Semaphore的底層實現依舊依賴於AQS的共用鎖機制。
static final class Node {
/** Marker to indicate a node is waiting in shared mode */
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode */
static final Node EXCLUSIVE = null;
/** waitStatus value to indicate thread has cancelled */
static final int CANCELLED = 1;
/** waitStatus value to indicate successor's thread needs unparking */
static final int SIGNAL = -1;
/** waitStatus value to indicate thread is waiting on condition */
static final int CONDITION = -2;
static final int PROPAGATE = -3;
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter;
}
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
private transient volatile Node head;
/**
* Tail of the wait queue, lazily initialized. Modified only via
* method enq to add new wait node.
*/
private transient volatile Node tail;
/**
* The synchronization state.
*/
private volatile int state;//最重要的一個變數
}
public class ConditionObject implements Condition, java.io.Serializable {
private static final long serialVersionUID = 1173984872572414699L;
/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter;
}
public final void acquire(int arg) {
if (!tryAcquire(arg) &&//嘗試獲取鎖
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//如果獲取鎖失敗，新增到佇列中，由於ReentrantLock是獨佔鎖所以節點必須是EXCLUSIVE型別
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);//cas失敗後執行入隊操作，繼續嘗試
return node;
}
private Node enq(final Node node) {
for (;;) {
Node t = tail;//獲取尾指標
if (t == null) { //代表當前佇列沒有節點
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)) {//前驅節點等於頭節點嘗試cas搶鎖。
setHead(node);//搶鎖成功將當前節點設定為頭節點
p.next = null; // help GC 當頭結點置空
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&//當佇列中有節點在等待，判斷是否應該阻塞
parkAndCheckInterrupt())//阻塞等待，檢查中斷標識位
interrupted = true;//將中斷標識位置為true
}
} finally {
if (failed)//
cancelAcquire(node);//取消當前節點
}
}
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)//如果當前節點的前驅節點的狀態大於0,代表是取消狀態，一直找到不是取消狀態的節點
node.prev = pred = pred.prev;
Node predNext = pred.next;//將當前要取消的節點斷鏈
node.waitStatus = Node.CANCELLED;//將當前節點的等待狀態置為CANCELLED
// If we are the tail, remove ourselves.
if (node == tail && compareAndSetTail(node, pred)) {//如果當前節點是尾結點，將尾結點替換為淺語節點
compareAndSetNext(pred, predNext, null);//將當前節點的下一個節點置為空，因為當前節點是最後一個節點沒有next指標
} 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;
if (pred != head &&//前驅節點不等於頭結點
((ws = pred.waitStatus) == Node.SIGNAL ||//前驅節點的狀態不等於SIGNAL
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&//前驅節點的狀態小於0，並且cas將前驅節點的等待置為SIGNAL
pred.thread != null) {//前驅節點的執行緒補位空
Node next = node.next;//獲取當前節點的next指標
if (next != null && next.waitStatus <= 0)//如果next指標不等於空並且等待狀態小於等於0，標識節點有效
compareAndSetNext(pred, predNext, next);//將前驅節點的next指標指向下一個有效節點
} else {
unparkSuccessor(node);//喚醒後續節點 條件：1.前驅節點是頭結點 2.當前節點不是signal,在ReentransLock中基本不會出現，在讀寫鎖時就會出現
}
node.next = node; // help GC 將參照指向自身
}
}
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);//cas將當前節點置為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;//將s置為空
for (Node t = tail; t != null && t != node; t = t.prev)//從尾結點遍歷找到一個不是取消狀態的節點
if (t.waitStatus <= 0)
s = t;
}
if (s != null)//如果s不等於空
LockSupport.unpark(s.thread);//喚醒當前節點s
}
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;//獲取上一個節點的等待狀態
if (ws == Node.SIGNAL)//如果狀態為SIGNAL,代表後續節點有節點可以喚醒，可以安心阻塞去
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
if (ws > 0) {//如果當前狀態大於0，代表節點為CANCELLED狀態
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;//從尾節點開始遍歷，找到下一個狀態不是CANCELLED的節點。將取消節點斷鏈移除
} 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>0時，已經找到了一個不是取消狀態的前驅節點。
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);//將找到的不是CANCELLED節點的前驅節點，將其等待狀態置為SIGNAL
}
return false;
}
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)//如果當前節點的前驅節點的狀態大於0,代表是取消狀態，一直找到不是取消狀態的節點
node.prev = pred = pred.prev;
Node predNext = pred.next;//將當前要取消的節點斷鏈
node.waitStatus = Node.CANCELLED;//將當前節點的等待狀態置為CANCELLED
// If we are the tail, remove ourselves.
if (node == tail && compareAndSetTail(node, pred)) {//如果當前節點是尾結點，將尾結點替換為淺語節點
compareAndSetNext(pred, predNext, null);//將當前節點的下一個節點置為空，因為當前節點是最後一個節點沒有next指標
} 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;
if (pred != head &&//前驅節點不等於頭結點
((ws = pred.waitStatus) == Node.SIGNAL ||//前驅節點的狀態不等於SIGNAL
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&//前驅節點的狀態小於0，並且cas將前驅節點的等待置為SIGNAL
pred.thread != null) {//前驅節點的執行緒補位空
Node next = node.next;//獲取當前節點的next指標
if (next != null && next.waitStatus <= 0)//如果next指標不等於空並且等待狀態小於等於0，標識節點有效
compareAndSetNext(pred, predNext, next);//將前驅節點的next指標指向下一個有效節點
} else {
unparkSuccessor(node);//喚醒後續節點 條件：1.前驅節點是頭結點 2.當前節點不是signal,在ReentransLock中基本不會出現，在讀寫鎖時就會出現
}
node.next = node; // help GC 將參照指向自身
}
}
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);//cas將當前節點置為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;//將s置為空
for (Node t = tail; t != null && t != node; t = t.prev)//從尾結點遍歷找到一個不是取消狀態的節點
if (t.waitStatus <= 0)
s = t;
}
if (s != null)//如果s不等於空
LockSupport.unpark(s.thread);//喚醒當前節點s
}
public final boolean release(int arg) {
if (tryRelease(arg)) {//子類實現如何釋放鎖
Node h = head;//獲取到頭結點
if (h != null && h.waitStatus != 0)//獲取到頭結點，如果頭結點不為空，等待狀態不為0,喚醒後續節點
unparkSuccessor(h);
return true;
}
return false;
}
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)//如果等待狀態小於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);//喚醒當前等待的有效節點S
}
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)//由子類實現
doAcquireShared(arg);
}
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，表示還有多餘的共用鎖。這裡等於0也需要進一步判斷。由於如果當執行到這裡時，有另外的執行緒釋放了共用鎖，如果不進行判斷，將會導致釋放鎖的執行緒沒辦法喚醒其他執行緒。所以這裡會偽喚醒一個節點，喚醒的節點後續如果沒有鎖釋放，依舊阻塞在當前parkAndCheckInterrupt方法中
setHeadAndPropagate(node, r);//將當前節點的等待狀態設定為Propagate。
p.next = null; // help GC
if (interrupted)//判斷是會否中斷過
selfInterrupt();//設定中斷標識位
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&//判斷是否應該阻塞等待
parkAndCheckInterrupt方法中())//阻塞並檢查中斷標識
interrupted = true;//重置中斷標識位
}
} finally {
if (failed)//如果失敗
cancelAcquire(node);//取消節點
}
}
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);//將當前節點置為頭結點
/*
* 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.
*/
if (propagate > 0 //可獲取的共用鎖也即讀鎖的數量，對於ReentrantReadWriteLock而言，永遠都是1，所以會繼續喚醒下一個讀執行緒
|| h == null //如果舊的頭結點為空
|| h.waitStatus < 0 ||//頭結點的等待狀態不為0
(h = head) == null || h.waitStatus < 0) {//舊頭節點不為空並且等待狀態小於0也即是有效節點
Node s = node.next;//獲取到node的下一個節點
if (s == null || s.isShared())//如果node的下一個節點為空或者是共用節點
doReleaseShared();//喚醒下一個執行緒
}
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {//子類實現釋放鎖
doReleaseShared();//喚醒後續執行緒
return true;//釋放成功
}
return false;//釋放是吧
}
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) {//如果狀態為SIGNAL
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))//cas將SIGNAL狀態置為0。SIGNAL標識後續有執行緒需要喚醒
continue; // loop to recheck cases
unparkSuccessor(h);//喚醒後續執行緒
}
else if (ws == 0 &&//如果當前狀態為0。表示有執行緒將其置為0
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))//cas將0狀態置為PROPAGATE。在多個共用鎖同時釋放時，方便繼續進行讀傳播，喚醒後續節點
continue; // loop on failed CAS
}
if (h == head)//如果頭結點沒有改變，證明沒有必要繼續迴圈等待了，直接退出吧，如果頭結點放生變化，可能有其他執行緒釋放了鎖。
break;
}
}
public final void await() throws InterruptedException {
if (Thread.interrupted())//執行緒是否發生中斷，是，就丟擲中斷異常
throw new InterruptedException();
Node node = addConditionWaiter();//加入條件等待佇列
int savedState = fullyRelease(node);//釋放鎖，並返回。因為當前執行緒需要等待
int interruptMode = 0;
while (!isOnSyncQueue(node)) {//判斷是否在競爭佇列中。AQS分為兩個佇列一個是競爭佇列，等待排程執行，一個是等待佇列等待在ConditionObject上。
LockSupport.park(this);//阻塞等待
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)//重新去獲取鎖並判斷當前中斷模式不是THROW_IE
interruptMode = REINTERRUPT;//將中斷模式置為REINTERRUPT
if (node.nextWaiter != null) // clean up if cancelled如果當前節點的下一個節點不為空
unlinkCancelledWaiters();//清除等待佇列中已經取消的節點
if (interruptMode != 0)//如果當前中斷模式不等於0
reportInterruptAfterWait(interruptMode);
}
private void reportInterruptAfterWait(int interruptMode)
throws InterruptedException {
if (interruptMode == THROW_IE)//如果是THROW_IE直接丟擲異常
throw new InterruptedException();
else if (interruptMode == REINTERRUPT)//如果是REINTERRUPT
selfInterrupt();//重置中斷標識位
}
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);//將當前節點封裝成node
if (t == null)//如果最後一個節點為空，表示當前節點是第一個入隊的節點
firstWaiter = node;
else
t.nextWaiter = node;//否則將當前node掛在連結串列末尾
lastWaiter = node;//設定最後節點的指標指向當前node
return node;
}
final int fullyRelease(Node node) {
boolean failed = true;
try {
int savedState = getState();//獲取當前state狀態
if (release(savedState)) {//釋放鎖嘗試
failed = false;
return savedState;//返回
} else {
throw new IllegalMonitorStateException();//丟擲釋放鎖異常
}
} finally {
if (failed)
node.waitStatus = Node.CANCELLED;//如果失敗將節點置為取消狀態
}
}
public final boolean release(int arg) {
if (tryRelease(arg)) {//嘗試釋放鎖，在CyclciBarrier中由於執行緒需要去阻塞，所以需要將鎖釋放，後續重新拿鎖
Node h = head;
if (h != null && h.waitStatus != 0)//從頭結點開始喚醒
unparkSuccessor(h);
return true;
}
return false;
}
final boolean isOnSyncQueue(Node node) {
if (node.waitStatus == Node.CONDITION || node.prev == null)//如果當前節點是Condition或者node.pre節點為空，標識不在競爭佇列中，返回faslse
return false;
if (node.next != null) // If has successor, it must be on queue 表示在競爭佇列中
return true;
/*
* node.prev can be non-null, but not yet on queue because
* the CAS to place it on queue can fail. So we have to
* traverse from tail to make sure it actually made it. It
* will always be near the tail in calls to this method, and
* unless the CAS failed (which is unlikely), it will be
* there, so we hardly ever traverse much.
*/
return findNodeFromTail(node);//從競爭佇列的尾結點開始找當前node，找到就返回true，否則為false
}
private boolean findNodeFromTail(Node node) {
Node t = tail;//獲取到尾結點
for (;;) {
if (t == node)
return true;
if (t == null)
return false;
t = t.prev;
}
}
private int checkInterruptWhileWaiting(Node node) {
return Thread.interrupted() ?//判斷當前是否中斷過
(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) ://如果移動到競爭佇列中併入隊成功，返回THROW_IE，否則返回REINTERRUPT
0;//沒有中斷過直接返回0
}
//走到這裡表示條件佇列的條件滿足，可以將節點移動到競爭佇列中執行
final boolean transferAfterCancelledWait(Node node) {
if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {//嘗試將當前為Condition的節點置為0，並移動到競爭佇列中
enq(node);
return true;
}
/*
* If we lost out to a signal(), then we can't proceed
* until it finishes its enq(). Cancelling during an
* incomplete transfer is both rare and transient, so just
* spin.
*/
while (!isOnSyncQueue(node))//如果不在競爭佇列中返回false
Thread.yield();
return false;
}
signalAll方法
public final void signalAll() {
if (!isHeldExclusively())//是不是持有獨佔鎖
throw new IllegalMonitorStateException();
Node first = firstWaiter;//獲取等待佇列的第一個節點
if (first != null)//如果節點不為空
doSignalAll(first);//喚醒所有執行緒
}
//從頭指標一直遍歷等待佇列，將其移動到競爭佇列中
private void doSignalAll(Node first) {
lastWaiter = firstWaiter = null;
do {
Node next = first.nextWaiter;
first.nextWaiter = null;
transferForSignal(first);//
first = next;
} while (first != null);
}
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))//cas自旋將其等待狀態改為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))//如果節點是取消狀態或者cas將其置為signal失敗，喚醒當前執行緒，讓他自己處理，後續在競爭佇列中會自動移除取消節點
LockSupport.unpark(node.thread);
return true;
}
總結：AQS提供了統一的模板，對於如何入隊出隊以及執行緒的喚醒都由AQS提供預設的實現，只需要子類實現自己上鎖和解鎖的邏輯。
import java.util.concurrent.Semaphore;
public class SemaphoreDemo {
public static void main(String[] args) {
//Semaphore s = new Semaphore(2);
Semaphore s = new Semaphore(2, true);
//允許一個執行緒同時執行
//Semaphore s = new Semaphore(1);
new Thread(() -> {
try {
s.acquire();
System.out.println("T1 running...");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
s.release();
}
}).start();
new Thread(() -> {
try {
s.acquire();
System.out.println("T2 running...");
s.release();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
s.release();
}
}).start();
}
}
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 1192457210091910933L;
Sync(int permits) {
setState(permits);//設定號誌
}
final int getPermits() {
return getState();//獲得號誌
}
final int nonfairTryAcquireShared(int acquires) {//非公平鎖的搶鎖方式
for (;;) {
int available = getState();//獲取state中的可用號誌
int remaining = available - acquires;//減1
if (remaining < 0 ||//號誌小於0，直接返回
compareAndSetState(available, remaining))//嘗試cas搶鎖
return remaining;//返回剩餘的號誌
}
}
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();//獲取當前state
int next = current + releases;//將state+1.也即號誌加1
if (next < current) // overflow 非法條件判斷，超過最大數量
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))//cas嘗試釋放鎖
return true;//釋放成功返回
}
}
//減少號誌
final void reducePermits(int reductions) {
for (;;) {
int current = getState();//獲取當前state
int next = current - reductions;
if (next > current) // underflow
throw new Error("Permit count underflow");
if (compareAndSetState(current, next))//cas嘗試減少號誌
return;
}
}
//清空訊號數量
final int drainPermits() {
for (;;) {
int current = getState();//獲取當前state狀態
if (current == 0 || compareAndSetState(current, 0))//當前訊號為0 或者將state置為0也即將訊號數量置為0
return current;
}
}
}
//公平鎖
static final class FairSync extends Sync {
private static final long serialVersionUID = 2014338818796000944L;
FairSync(int permits) {
super(permits);
}
protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors())//佇列中是否有執行緒在排隊
return -1;//獲取失敗
int available = getState();//可用的號誌
int remaining = available - acquires;//減去當前獲取的數量
if (remaining < 0 ||//可用的號誌小於0
compareAndSetState(available, remaining))//cas設定state變數.
return remaining;//返回可用的號誌
}
}
}
//非公平鎖
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898L;
NonfairSync(int permits) {
super(permits);
}
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);//詳情請看父類別的實現
}
}
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);//請檢視父類別實現，與acquireShared一致，不過加了一場處理
}
public void release() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {//Semaphore的類實現鎖獲取的方法。
doReleaseShared();//與AQS中一致，不過多贅述
return true;
}
return false;
}
到了這裡，其實AQS的原始碼基本已經覆蓋了，對於AQS的原始碼也應該有了清楚的認知。總結就是：一個volatile 的state變數，兩個等待佇列(競爭佇列，條件佇列)，通過cas的方式保證單變數的原子性。後續將會對Exchanger以及Phaser進行原始碼解析，到此基本AQS已經到了一個段落了。後續觀看原始碼時，請注意多考慮一下多執行緒並行時可能出現的情況，去理解doug lea寫程式碼的思路。