JUC同步鎖原理源碼解析四----Semaphore

　　

(相關(guān)資料圖)

JUC同步鎖原理源碼解析四----SemaphoreSemaphore1.Semaphore的來(lái)源

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.

?一組數(shù)量的信號(hào)，只有獲取到信號(hào)的線程才允許執(zhí)行。通過(guò)acquire進(jìn)行獲取，如果獲取不到則需要阻塞等待直到一個(gè)信號(hào)可用。release會(huì)釋放一個(gè)信號(hào)量。通過(guò)這種方式可以實(shí)現(xiàn)限流。

2.Semaphore的底層實(shí)現(xiàn)

?Semaphore的底層實(shí)現(xiàn)依舊依賴(lài)于AQS的共享鎖機(jī)制。

2.AQS源碼Node節(jié)點(diǎn)

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類(lèi)

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;//最重要的一個(gè)變量       }

ConditionObject類(lèi)

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))//如果獲取鎖失敗，添加到隊(duì)列中，由于ReentrantLock是獨(dú)占鎖所以節(jié)點(diǎn)必須是EXCLUSIVE類(lèi)型        selfInterrupt();//添加中斷標(biāo)識(shí)位}

addWaiter方法

private Node addWaiter(Node mode) {     Node node = new Node(Thread.currentThread(), mode);//新建節(jié)點(diǎn)     // Try the fast path of enq; backup to full enq on failure     Node pred = tail;//獲取到尾指針     if (pred != null) {//尾指針不等于空，將當(dāng)前節(jié)點(diǎn)替換為尾指針         node.prev = pred;         if (compareAndSetTail(pred, node)) {//采用尾插法，充分利用時(shí)間局部性和空間局部性。尾插的節(jié)點(diǎn)一般不容易被取消。             pred.next = node;             return node;         }     }     enq(node);//cas失敗后執(zhí)行入隊(duì)操作，繼續(xù)嘗試     return node; }

enq方法

private Node enq(final Node node) {    for (;;) {        Node t = tail;//獲取尾指針        if (t == null) { //代表當(dāng)前隊(duì)列沒(méi)有節(jié)點(diǎn)            if (compareAndSetHead(new Node()))//將當(dāng)前節(jié)點(diǎn)置為頭結(jié)點(diǎn)                tail = head;        } else {//當(dāng)前隊(duì)列有節(jié)點(diǎn)            node.prev = t;//            if (compareAndSetTail(t, node)) {//將當(dāng)前節(jié)點(diǎn)置為尾結(jié)點(diǎn)                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();//找到當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)            if (p == head && tryAcquire(arg)) {//前驅(qū)節(jié)點(diǎn)等于頭節(jié)點(diǎn)嘗試cas搶鎖。                setHead(node);//搶鎖成功將當(dāng)前節(jié)點(diǎn)設(shè)置為頭節(jié)點(diǎn)                p.next = null; // help GC  當(dāng)頭結(jié)點(diǎn)置空                failed = false;                return interrupted;            }            if (shouldParkAfterFailedAcquire(p, node) &&//當(dāng)隊(duì)列中有節(jié)點(diǎn)在等待，判斷是否應(yīng)該阻塞                parkAndCheckInterrupt())//阻塞等待，檢查中斷標(biāo)識(shí)位                interrupted = true;//將中斷標(biāo)識(shí)位置為true        }    } finally {        if (failed)//            cancelAcquire(node);//取消當(dāng)前節(jié)點(diǎn)    }} private void cancelAcquire(Node node) {     // Ignore if node doesn"t exist     if (node == null)//當(dāng)前節(jié)點(diǎn)為空直接返回         return;     node.thread = null;//要取消了將當(dāng)前節(jié)點(diǎn)的線程置為空     // Skip cancelled predecessors     Node pred = node.prev;//獲取到當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)     while (pred.waitStatus > 0)//如果當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)的狀態(tài)大于0,代表是取消狀態(tài)，一直找到不是取消狀態(tài)的節(jié)點(diǎn)         node.prev = pred = pred.prev;     Node predNext = pred.next;//將當(dāng)前要取消的節(jié)點(diǎn)斷鏈     node.waitStatus = Node.CANCELLED;//將當(dāng)前節(jié)點(diǎn)的等待狀態(tài)置為CANCELLED     // If we are the tail, remove ourselves.     if (node == tail && compareAndSetTail(node, pred)) {//如果當(dāng)前節(jié)點(diǎn)是尾結(jié)點(diǎn)，將尾結(jié)點(diǎn)替換為淺語(yǔ)節(jié)點(diǎn)         compareAndSetNext(pred, predNext, null);//將當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)置為空，因?yàn)楫?dāng)前節(jié)點(diǎn)是最后一個(gè)節(jié)點(diǎn)沒(méi)有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 &&//前驅(qū)節(jié)點(diǎn)不等于頭結(jié)點(diǎn)             ((ws = pred.waitStatus) == Node.SIGNAL ||//前驅(qū)節(jié)點(diǎn)的狀態(tài)不等于SIGNAL              (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&//前驅(qū)節(jié)點(diǎn)的狀態(tài)小于0，并且cas將前驅(qū)節(jié)點(diǎn)的等待置為SIGNAL             pred.thread != null) {//前驅(qū)節(jié)點(diǎn)的線程補(bǔ)位空             Node next = node.next;//獲取當(dāng)前節(jié)點(diǎn)的next指針             if (next != null && next.waitStatus <= 0)//如果next指針不等于空并且等待狀態(tài)小于等于0，標(biāo)識(shí)節(jié)點(diǎn)有效                 compareAndSetNext(pred, predNext, next);//將前驅(qū)節(jié)點(diǎn)的next指針指向下一個(gè)有效節(jié)點(diǎn)         } else {             unparkSuccessor(node);//喚醒后續(xù)節(jié)點(diǎn) 條件：1.前驅(qū)節(jié)點(diǎn)是頭結(jié)點(diǎn) 2.當(dāng)前節(jié)點(diǎn)不是signal,在ReentransLock中基本不會(huì)出現(xiàn)，在讀寫(xiě)鎖時(shí)就會(huì)出現(xiàn)         }         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;//獲取當(dāng)前節(jié)點(diǎn)狀態(tài)     if (ws < 0)//如果節(jié)點(diǎn)為負(fù)數(shù)也即不是取消節(jié)點(diǎn)         compareAndSetWaitStatus(node, ws, 0);//cas將當(dāng)前節(jié)點(diǎn)置為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;//獲取到下一個(gè)節(jié)點(diǎn)     if (s == null || s.waitStatus > 0) {//下一個(gè)節(jié)點(diǎn)等于空或者下一個(gè)節(jié)點(diǎn)是取消節(jié)點(diǎn)         s = null;//將s置為空         for (Node t = tail; t != null && t != node; t = t.prev)//從尾結(jié)點(diǎn)遍歷找到一個(gè)不是取消狀態(tài)的節(jié)點(diǎn)             if (t.waitStatus <= 0)                 s = t;     }     if (s != null)//如果s不等于空         LockSupport.unpark(s.thread);//喚醒當(dāng)前節(jié)點(diǎn)s }

shouldParkAfterFailedAcquire方法

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {    int ws = pred.waitStatus;//獲取上一個(gè)節(jié)點(diǎn)的等待狀態(tài)    if (ws == Node.SIGNAL)//如果狀態(tài)為SIGNAL,代表后續(xù)節(jié)點(diǎn)有節(jié)點(diǎn)可以喚醒，可以安心阻塞去        /*             * This node has already set status asking a release             * to signal it, so it can safely park.             */        return true;    if (ws > 0) {//如果當(dāng)前狀態(tài)大于0，代表節(jié)點(diǎn)為CANCELLED狀態(tài)        /*             * Predecessor was cancelled. Skip over predecessors and             * indicate retry.             */        do {            node.prev = pred = pred.prev;//從尾節(jié)點(diǎn)開(kāi)始遍歷，找到下一個(gè)狀態(tài)不是CANCELLED的節(jié)點(diǎn)。將取消節(jié)點(diǎn)斷鏈移除        } 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時(shí)，已經(jīng)找到了一個(gè)不是取消狀態(tài)的前驅(qū)節(jié)點(diǎn)。        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);//將找到的不是CANCELLED節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)，將其等待狀態(tài)置為SIGNAL    }    return false;}

cancelAcquire方法

private void cancelAcquire(Node node) {     // Ignore if node doesn"t exist     if (node == null)//當(dāng)前節(jié)點(diǎn)為空直接返回         return;     node.thread = null;//要取消了將當(dāng)前節(jié)點(diǎn)的線程置為空     // Skip cancelled predecessors     Node pred = node.prev;//獲取到當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)     while (pred.waitStatus > 0)//如果當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)的狀態(tài)大于0,代表是取消狀態(tài)，一直找到不是取消狀態(tài)的節(jié)點(diǎn)         node.prev = pred = pred.prev;     Node predNext = pred.next;//將當(dāng)前要取消的節(jié)點(diǎn)斷鏈     node.waitStatus = Node.CANCELLED;//將當(dāng)前節(jié)點(diǎn)的等待狀態(tài)置為CANCELLED     // If we are the tail, remove ourselves.     if (node == tail && compareAndSetTail(node, pred)) {//如果當(dāng)前節(jié)點(diǎn)是尾結(jié)點(diǎn)，將尾結(jié)點(diǎn)替換為淺語(yǔ)節(jié)點(diǎn)         compareAndSetNext(pred, predNext, null);//將當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)置為空，因?yàn)楫?dāng)前節(jié)點(diǎn)是最后一個(gè)節(jié)點(diǎn)沒(méi)有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 &&//前驅(qū)節(jié)點(diǎn)不等于頭結(jié)點(diǎn)             ((ws = pred.waitStatus) == Node.SIGNAL ||//前驅(qū)節(jié)點(diǎn)的狀態(tài)不等于SIGNAL              (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&//前驅(qū)節(jié)點(diǎn)的狀態(tài)小于0，并且cas將前驅(qū)節(jié)點(diǎn)的等待置為SIGNAL             pred.thread != null) {//前驅(qū)節(jié)點(diǎn)的線程補(bǔ)位空             Node next = node.next;//獲取當(dāng)前節(jié)點(diǎn)的next指針             if (next != null && next.waitStatus <= 0)//如果next指針不等于空并且等待狀態(tài)小于等于0，標(biāo)識(shí)節(jié)點(diǎn)有效                 compareAndSetNext(pred, predNext, next);//將前驅(qū)節(jié)點(diǎn)的next指針指向下一個(gè)有效節(jié)點(diǎn)         } else {             unparkSuccessor(node);//喚醒后續(xù)節(jié)點(diǎn) 條件：1.前驅(qū)節(jié)點(diǎn)是頭結(jié)點(diǎn) 2.當(dāng)前節(jié)點(diǎn)不是signal,在ReentransLock中基本不會(huì)出現(xiàn)，在讀寫(xiě)鎖時(shí)就會(huì)出現(xiàn)         }         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;//獲取當(dāng)前節(jié)點(diǎn)狀態(tài)     if (ws < 0)//如果節(jié)點(diǎn)為負(fù)數(shù)也即不是取消節(jié)點(diǎn)         compareAndSetWaitStatus(node, ws, 0);//cas將當(dāng)前節(jié)點(diǎn)置為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;//獲取到下一個(gè)節(jié)點(diǎn)     if (s == null || s.waitStatus > 0) {//下一個(gè)節(jié)點(diǎn)等于空或者下一個(gè)節(jié)點(diǎn)是取消節(jié)點(diǎn)         s = null;//將s置為空         for (Node t = tail; t != null && t != node; t = t.prev)//從尾結(jié)點(diǎn)遍歷找到一個(gè)不是取消狀態(tài)的節(jié)點(diǎn)             if (t.waitStatus <= 0)                 s = t;     }     if (s != null)//如果s不等于空         LockSupport.unpark(s.thread);//喚醒當(dāng)前節(jié)點(diǎn)s }

release方法

public final boolean release(int arg) {    if (tryRelease(arg)) {//子類(lèi)實(shí)現(xiàn)如何釋放鎖        Node h = head;//獲取到頭結(jié)點(diǎn)        if (h != null && h.waitStatus != 0)//獲取到頭結(jié)點(diǎn)，如果頭結(jié)點(diǎn)不為空，等待狀態(tài)不為0,喚醒后續(xù)節(jié)點(diǎn)            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;//獲取節(jié)點(diǎn)的等待狀態(tài)    if (ws < 0)//如果等待狀態(tài)小于0,標(biāo)識(shí)節(jié)點(diǎn)屬于有效節(jié)點(diǎn)        compareAndSetWaitStatus(node, ws, 0);//將當(dāng)前節(jié)點(diǎn)的等待狀態(tài)置為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;//獲取到下一個(gè)節(jié)點(diǎn)    if (s == null || s.waitStatus > 0) {//如果節(jié)點(diǎn)是空，或者是取消狀態(tài)的節(jié)點(diǎn)，就找到一個(gè)非取消狀態(tài)的節(jié)點(diǎn)，將取消狀態(tài)的節(jié)點(diǎn)斷鏈后由垃圾回收器進(jìn)行回收        s = null;        for (Node t = tail; t != null && t != node; t = t.prev)            if (t.waitStatus <= 0)                s = t;    }    if (s != null)//節(jié)點(diǎn)不用空        LockSupport.unpark(s.thread);//喚醒當(dāng)前等待的有效節(jié)點(diǎn)S}

acquireShared方法

public final void acquireShared(int arg) {    if (tryAcquireShared(arg) < 0)//由子類(lèi)實(shí)現(xiàn)        doAcquireShared(arg);}

doAcquireShared方法

private void doAcquireShared(int arg) {    final Node node = addWaiter(Node.SHARED);//將共享節(jié)點(diǎn)也即讀線程入隊(duì)并返回    boolean failed = true;    try {        boolean interrupted = false;        for (;;) {            final Node p = node.predecessor();//找到節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)            if (p == head) {//如果前驅(qū)節(jié)點(diǎn)等于頭結(jié)點(diǎn)                int r = tryAcquireShared(arg);//嘗試獲取共享鎖數(shù)量                if (r >= 0) {//如果鎖的數(shù)量大于0，表示還有多余的共享鎖。這里等于0也需要進(jìn)一步判斷。由于如果當(dāng)執(zhí)行到這里時(shí)，有另外的線程釋放了共享鎖，如果不進(jìn)行判斷，將會(huì)導(dǎo)致釋放鎖的線程沒(méi)辦法喚醒其他線程。所以這里會(huì)偽喚醒一個(gè)節(jié)點(diǎn)，喚醒的節(jié)點(diǎn)后續(xù)如果沒(méi)有鎖釋放，依舊阻塞在當(dāng)前parkAndCheckInterrupt方法中                    setHeadAndPropagate(node, r);//將當(dāng)前節(jié)點(diǎn)的等待狀態(tài)設(shè)置為Propagate。                    p.next = null; // help GC                    if (interrupted)//判斷是會(huì)否中斷過(guò)                        selfInterrupt();//設(shè)置中斷標(biāo)識(shí)位                    failed = false;                    return;                }            }            if (shouldParkAfterFailedAcquire(p, node) &&//判斷是否應(yīng)該阻塞等待                parkAndCheckInterrupt方法中())//阻塞并檢查中斷標(biāo)識(shí)                interrupted = true;//重置中斷標(biāo)識(shí)位        }    } finally {        if (failed)//如果失敗            cancelAcquire(node);//取消節(jié)點(diǎn)    }}

setHeadAndPropagate方法

private void setHeadAndPropagate(Node node, int propagate) {        Node h = head; // Record old head for check below        setHead(node);//將當(dāng)前節(jié)點(diǎn)置為頭結(jié)點(diǎn)        /*         * 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 //可獲取的共享鎖也即讀鎖的數(shù)量，對(duì)于ReentrantReadWriteLock而言，永遠(yuǎn)都是1，所以會(huì)繼續(xù)喚醒下一個(gè)讀線程            || h == null //如果舊的頭結(jié)點(diǎn)為空            || h.waitStatus < 0 ||//頭結(jié)點(diǎn)的等待狀態(tài)不為0            (h = head) == null || h.waitStatus < 0) {//舊頭節(jié)點(diǎn)不為空并且等待狀態(tài)小于0也即是有效節(jié)點(diǎn)            Node s = node.next;//獲取到node的下一個(gè)節(jié)點(diǎn)            if (s == null || s.isShared())//如果node的下一個(gè)節(jié)點(diǎn)為空或者是共享節(jié)點(diǎn)                doReleaseShared();//喚醒下一個(gè)線程        }    }

releaseShared方法

public final boolean releaseShared(int arg) {    if (tryReleaseShared(arg)) {//子類(lèi)實(shí)現(xiàn)釋放鎖        doReleaseShared();//喚醒后續(xù)線程        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;//獲取到當(dāng)前頭結(jié)點(diǎn)        if (h != null && h != tail) {//如果頭結(jié)點(diǎn)不為空并且不等于尾結(jié)點(diǎn)            int ws = h.waitStatus;//獲取當(dāng)前節(jié)點(diǎn)的等待狀態(tài)            if (ws == Node.SIGNAL) {//如果狀態(tài)為SIGNAL                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))//cas將SIGNAL狀態(tài)置為0。SIGNAL標(biāo)識(shí)后續(xù)有線程需要喚醒                    continue;            // loop to recheck cases                unparkSuccessor(h);//喚醒后續(xù)線程            }            else if (ws == 0 &&//如果當(dāng)前狀態(tài)為0。表示有線程將其置為0                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))//cas將0狀態(tài)置為PROPAGATE。在多個(gè)共享鎖同時(shí)釋放時(shí)，方便繼續(xù)進(jìn)行讀傳播，喚醒后續(xù)節(jié)點(diǎn)                continue;                // loop on failed CAS        }        if (h == head)//如果頭結(jié)點(diǎn)沒(méi)有改變，證明沒(méi)有必要繼續(xù)循環(huán)等待了，直接退出吧，如果頭結(jié)點(diǎn)放生變化，可能有其他線程釋放了鎖。            break;    }}

await()

public final void await() throws InterruptedException {    if (Thread.interrupted())//線程是否發(fā)生中斷，是，就拋出中斷異常        throw new InterruptedException();    Node node = addConditionWaiter();//加入條件等待隊(duì)列    int savedState = fullyRelease(node);//釋放鎖，并返回。因?yàn)楫?dāng)前線程需要等待    int interruptMode = 0;    while (!isOnSyncQueue(node)) {//判斷是否在競(jìng)爭(zhēng)隊(duì)列中。AQS分為兩個(gè)隊(duì)列一個(gè)是競(jìng)爭(zhēng)隊(duì)列，等待調(diào)度執(zhí)行，一個(gè)是等待隊(duì)列等待在ConditionObject上。        LockSupport.park(this);//阻塞等待        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)            break;    }    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)//重新去獲取鎖并判斷當(dāng)前中斷模式不是THROW_IE        interruptMode = REINTERRUPT;//將中斷模式置為REINTERRUPT    if (node.nextWaiter != null) // clean up if cancelled如果當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)不為空        unlinkCancelledWaiters();//清除等待隊(duì)列中已經(jīng)取消的節(jié)點(diǎn)    if (interruptMode != 0)//如果當(dāng)前中斷模式不等于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();//重置中斷標(biāo)識(shí)位}

addConditionWaiter方法

private Node addConditionWaiter() {    Node t = lastWaiter;//獲取到最后一個(gè)節(jié)點(diǎn)    // If lastWaiter is cancelled, clean out.    if (t != null && t.waitStatus != Node.CONDITION) {//最后一個(gè)節(jié)點(diǎn)不等于空，并且等待狀態(tài)不等于CONDITION        unlinkCancelledWaiters();//將取消節(jié)點(diǎn)斷鏈,標(biāo)準(zhǔn)的鏈表操作        t = lastWaiter;//獲取到最后一個(gè)有效的節(jié)點(diǎn)    }    Node node = new Node(Thread.currentThread(), Node.CONDITION);//將當(dāng)前節(jié)點(diǎn)封裝成node    if (t == null)//如果最后一個(gè)節(jié)點(diǎn)為空，表示當(dāng)前節(jié)點(diǎn)是第一個(gè)入隊(duì)的節(jié)點(diǎn)        firstWaiter = node;    else        t.nextWaiter = node;//否則將當(dāng)前node掛在鏈表末尾    lastWaiter = node;//設(shè)置最后節(jié)點(diǎn)的指針指向當(dāng)前node    return node;}

fullyRelease方法

final int fullyRelease(Node node) {    boolean failed = true;    try {        int savedState = getState();//獲取當(dāng)前state狀態(tài)        if (release(savedState)) {//釋放鎖嘗試            failed = false;            return savedState;//返回        } else {            throw new IllegalMonitorStateException();//拋出釋放鎖異常        }    } finally {        if (failed)            node.waitStatus = Node.CANCELLED;//如果失敗將節(jié)點(diǎn)置為取消狀態(tài)    }}public final boolean release(int arg) {    if (tryRelease(arg)) {//嘗試釋放鎖，在CyclciBarrier中由于線程需要去阻塞，所以需要將鎖釋放，后續(xù)重新拿鎖        Node h = head;        if (h != null && h.waitStatus != 0)//從頭結(jié)點(diǎn)開(kāi)始喚醒            unparkSuccessor(h);        return true;    }    return false;}

isOnSyncQueue方法

final boolean isOnSyncQueue(Node node) {    if (node.waitStatus == Node.CONDITION || node.prev == null)//如果當(dāng)前節(jié)點(diǎn)是Condition或者node.pre節(jié)點(diǎn)為空，標(biāo)識(shí)不在競(jìng)爭(zhēng)隊(duì)列中，返回faslse        return false;    if (node.next != null) // If has successor, it must be on queue  表示在競(jìng)爭(zhēng)隊(duì)列中        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);//從競(jìng)爭(zhēng)隊(duì)列的尾結(jié)點(diǎn)開(kāi)始找當(dāng)前node，找到就返回true，否則為false}private boolean findNodeFromTail(Node node) {    Node t = tail;//獲取到尾結(jié)點(diǎn)    for (;;) {        if (t == node)            return true;        if (t == null)            return false;        t = t.prev;    }}

findNodeFromTail方法

private int checkInterruptWhileWaiting(Node node) {    return Thread.interrupted() ?//判斷當(dāng)前是否中斷過(guò)        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) ://如果移動(dòng)到競(jìng)爭(zhēng)隊(duì)列中并入隊(duì)成功，返回THROW_IE，否則返回REINTERRUPT    0;//沒(méi)有中斷過(guò)直接返回0}//走到這里表示條件隊(duì)列的條件滿足，可以將節(jié)點(diǎn)移動(dòng)到競(jìng)爭(zhēng)隊(duì)列中執(zhí)行final boolean transferAfterCancelledWait(Node node) {    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {//嘗試將當(dāng)前為Condition的節(jié)點(diǎn)置為0，并移動(dòng)到競(jìng)爭(zhēng)隊(duì)列中        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))//如果不在競(jìng)爭(zhēng)隊(duì)列中返回false        Thread.yield();    return false;}

signalAll方法

public final void signalAll() {    if (!isHeldExclusively())//是不是持有獨(dú)占鎖        throw new IllegalMonitorStateException();    Node first = firstWaiter;//獲取等待隊(duì)列的第一個(gè)節(jié)點(diǎn)    if (first != null)//如果節(jié)點(diǎn)不為空        doSignalAll(first);//喚醒所有線程}//從頭指針一直遍歷等待隊(duì)列，將其移動(dòng)到競(jìng)爭(zhēng)隊(duì)列中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自旋將其等待狀態(tài)改為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);//將其放入競(jìng)爭(zhēng)隊(duì)列    int ws = p.waitStatus;//獲取節(jié)點(diǎn)的等待狀態(tài)    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))//如果節(jié)點(diǎn)是取消狀態(tài)或者cas將其置為signal失敗，喚醒當(dāng)前線程，讓他自己處理，后續(xù)在競(jìng)爭(zhēng)隊(duì)列中會(huì)自動(dòng)移除取消節(jié)點(diǎn)        LockSupport.unpark(node.thread);    return true;}

總結(jié)：AQS提供了統(tǒng)一的模板，對(duì)于如何入隊(duì)出隊(duì)以及線程的喚醒都由AQS提供默認(rèn)的實(shí)現(xiàn)，只需要子類(lèi)實(shí)現(xiàn)自己上鎖和解鎖的邏輯。

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);        //允許一個(gè)線程同時(shí)執(zhí)行        //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類(lèi)

abstract static class Sync extends AbstractQueuedSynchronizer {    private static final long serialVersionUID = 1192457210091910933L;    Sync(int permits) {        setState(permits);//設(shè)置信號(hào)量    }    final int getPermits() {        return getState();//獲得信號(hào)量    }    final int nonfairTryAcquireShared(int acquires) {//非公平鎖的搶鎖方式        for (;;) {            int available = getState();//獲取state中的可用信號(hào)量            int remaining = available - acquires;//減1            if (remaining < 0 ||//信號(hào)量小于0，直接返回                compareAndSetState(available, remaining))//嘗試cas搶鎖                return remaining;//返回剩余的信號(hào)量        }    }    protected final boolean tryReleaseShared(int releases) {        for (;;) {            int current = getState();//獲取當(dāng)前state            int next = current + releases;//將state+1.也即信號(hào)量加1            if (next < current) // overflow 非法條件判斷，超過(guò)最大數(shù)量                throw new Error("Maximum permit count exceeded");            if (compareAndSetState(current, next))//cas嘗試釋放鎖                return true;//釋放成功返回        }    }    //減少信號(hào)量    final void reducePermits(int reductions) {        for (;;) {            int current = getState();//獲取當(dāng)前state            int next = current - reductions;            if (next > current) // underflow                throw new Error("Permit count underflow");            if (compareAndSetState(current, next))//cas嘗試減少信號(hào)量                return;        }    }    //清空信號(hào)數(shù)量    final int drainPermits() {        for (;;) {            int current = getState();//獲取當(dāng)前state狀態(tài)            if (current == 0 || compareAndSetState(current, 0))//當(dāng)前信號(hào)為0 或者將state置為0也即將信號(hào)數(shù)量置為0                return current;        }    }}

FairSync與NonfairSync的類(lèi)實(shí)現(xiàn)

//公平鎖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())//隊(duì)列中是否有線程在排隊(duì)                return -1;//獲取失敗            int available = getState();//可用的信號(hào)量            int remaining = available - acquires;//減去當(dāng)前獲取的數(shù)量            if (remaining < 0 ||//可用的信號(hào)量小于0                compareAndSetState(available, remaining))//cas設(shè)置state變量.                return remaining;//返回可用的信號(hào)量        }    }}//非公平鎖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);//詳情請(qǐng)看父類(lèi)的實(shí)現(xiàn)    }}

acquire方法

public void acquire() throws InterruptedException {    sync.acquireSharedInterruptibly(1);//請(qǐng)查看父類(lèi)實(shí)現(xiàn)，與acquireShared一致，不過(guò)加了一場(chǎng)處理}

release方法:

public void release() {    sync.releaseShared(1);}public final boolean releaseShared(int arg) {    if (tryReleaseShared(arg)) {//Semaphore的類(lèi)實(shí)現(xiàn)鎖獲取的方法。        doReleaseShared();//與AQS中一致，不過(guò)多贅述        return true;    }    return false;}

4.留言

?到了這里，其實(shí)AQS的源碼基本已經(jīng)覆蓋了，對(duì)于AQS的源碼也應(yīng)該有了清楚的認(rèn)知?？偨Y(jié)就是：一個(gè)volatile 的state變量，兩個(gè)等待隊(duì)列(競(jìng)爭(zhēng)隊(duì)列，條件隊(duì)列)，通過(guò)cas的方式保證單變量的原子性。后續(xù)將會(huì)對(duì)Exchanger以及Phaser進(jìn)行源碼解析，到此基本AQS已經(jīng)到了一個(gè)段落了。后續(xù)觀看源碼時(shí)，請(qǐng)注意多考慮一下多線程并發(fā)時(shí)可能出現(xiàn)的情況，去理解doug lea寫(xiě)代碼的思路。

關(guān)鍵詞：