(相關資料圖)
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會釋放一個信號量。通過這種方式可以實現限流。
2.Semaphore的底層實現?Semaphore的底層實現依舊依賴于AQS的共享鎖機制。
2.AQS源碼Node節點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;}
AbstractQueuedSynchronizer類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;//最重要的一個變量 }
ConditionObject類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;}
accquire方法public final void acquire(int arg) { if (!tryAcquire(arg) &&//嘗試獲取鎖 acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//如果獲取鎖失敗,添加到隊列中,由于ReentrantLock是獨占鎖所以節點必須是EXCLUSIVE類型 selfInterrupt();//添加中斷標識位}
addWaiter方法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; }
enq方法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; } } }}
acquireQueued方法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 }
shouldParkAfterFailedAcquire方法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;}
cancelAcquire方法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 將引用指向自身 } }
unparkSuccessor方法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 }
release方法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}
acquireShared方法public final void acquireShared(int arg) { if (tryAcquireShared(arg) < 0)//由子類實現 doAcquireShared(arg);}
doAcquireShared方法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);//取消節點 }}
setHeadAndPropagate方法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();//喚醒下一個線程 } }
releaseShared方法public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {//子類實現釋放鎖 doReleaseShared();//喚醒后續線程 return true;//釋放成功 } return false;//釋放是吧}
doReleaseShared方法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; }}
await()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();//重置中斷標識位}
addConditionWaiter方法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;}
fullyRelease方法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;}
isOnSyncQueue方法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; }}
findNodeFromTail方法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);}
transferForSignal方法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提供默認的實現,只需要子類實現自己上鎖和解鎖的邏輯。
3.Semaphore基本使用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(); }}
Sync類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; } }}
FairSync與NonfairSync的類實現//公平鎖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);//詳情請看父類的實現 }}
acquire方法public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1);//請查看父類實現,與acquireShared一致,不過加了一場處理}
release方法:public void release() { sync.releaseShared(1);}public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {//Semaphore的類實現鎖獲取的方法。 doReleaseShared();//與AQS中一致,不過多贅述 return true; } return false;}
4.留言?到了這里,其實AQS的源碼基本已經覆蓋了,對于AQS的源碼也應該有了清楚的認知。總結就是:一個volatile 的state變量,兩個等待隊列(競爭隊列,條件隊列),通過cas的方式保證單變量的原子性。后續將會對Exchanger以及Phaser進行源碼解析,到此基本AQS已經到了一個段落了。后續觀看源碼時,請注意多考慮一下多線程并發時可能出現的情況,去理解doug lea寫代碼的思路。
關鍵詞:
版權與免責聲明:
1 本網注明“來源:×××”(非商業周刊網)的作品,均轉載自其它媒體,轉載目的在于傳遞更多信息,并不代表本網贊同其觀點和對其真實性負責,本網不承擔此類稿件侵權行為的連帶責任。
2 在本網的新聞頁面或BBS上進行跟帖或發表言論者,文責自負。
3 相關信息并未經過本網站證實,不對您構成任何投資建議,據此操作,風險自擔。
4 如涉及作品內容、版權等其它問題,請在30日內同本網聯系。