Semaphore訊號量源碼解析

介紹

Semaphore是什麼

Semaphore可以稱為訊號量，這個原本是作業系統中的概念，是一種執行緒同步方法，配合PV操作實現執行緒之間的同步功能。訊號量可以表示作業系統中某種資源的個數，因此可以用來控制同時訪問該資源的最大執行緒數，以保證資源的合理使用

Java的JUC對Semaphore做了具體的實現，其功能和作業系統中的訊號量基本一致，也是一種執行緒同步並發工具

使用場景

Semaphore常用於某些有限資源的並發使用場景，即限流

場景一
資料庫連接池對於同時連接的執行緒數有限制，當連接數達到限制後，接下來的執行緒必須等待前面的執行緒釋放連接才可以獲得資料庫連接

場景二
醫院叫號，放出的號數是固定的，不然醫院窗口來不及處理。因此只有取到號才能去門診，沒取到號的只能在外面等待放號

Semaphore常用方法介紹

構造函數

Semaphore(int permits)：創建一個許可數為permits的Semaphore
Semaphore(int permits, boolean fair)：創建一個許可數為permits的Semaphore，可以選擇公平模式或非公平模式

獲取許可

void acquire() throws InterruptedException：獲取一個許可，響應中斷，獲取不到則阻塞等待
void acquire(int permits) throws InterruptedException：獲取指定數量的許可，響應中斷，獲取不到則阻塞等待
void acquireUninterruptibly()：獲取一個許可，忽略中斷，獲取不到則阻塞等待
void acquireUninterruptibly(int permits)：獲取指定數量的許可，忽略中斷，獲取不到則阻塞等待
int drainPermits()：獲取當前所有可用的許可，並返回獲得的許可數

釋放許可

void release()：釋放一個許可
void release(int permits)：釋放指定數量的許可

嘗試獲取許可

boolean tryAcquire()：嘗試獲取一個許可，如果獲取失敗不會被阻塞，而是返回false。成功則返回true
boolean tryAcquire(int permits)：嘗試獲取指定數量的許可，如果獲取失敗不會被阻塞，而是返回false。成功則返回true
boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException：嘗試獲取一個許可，如果獲取失敗則會等待指定時間，如果超時還未獲得，則返回false。獲取成功則返回true。在等待過程中如果被中斷，則會立即拋出中斷異常
boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException：嘗試獲取指定數量的許可，如果獲取失敗則會等待指定時間，如果超時還未獲得，則返回false。獲取成功則返回true。在等待過程中如果被中斷，則會立即拋出中斷異常

其他方法

boolean isFair()：判斷訊號量是否是公平模式
int availablePermits()：返回當前可用的許可數
boolean hasQueuedThreads()：判斷是否有執行緒正在等待獲取許可
int getQueueLength()：返回當前等待獲取許可的執行緒的估計值。該值並不是一個確定值，因為在執行該函數時，執行緒數可能已經發生了變化

Semaphore使用示例

針對「使用場景」的場景二，假設醫院最多接待2個人，如果接待滿了，那麼所有人都必須在大廳等待（不能「忙等」）

程式碼如下：

無論如何，醫院內最多有5個病人同時接診。這也說明了Semaphore可以控制某種資源最多同時被指定數量的執行緒使用

作者：酒冽出處：//www.cnblogs.com/frankiedyz/p/15674098.html
版權：本文版權歸作者和部落格園共有
轉載：歡迎轉載，但未經作者同意，必須保留此段聲明；必須在文章中給出原文連接；否則必究法律責任

構造函數

Semaphore有兩個構造函數，都是需要設置許可總數，額外的另一個參數是用來控制公平模式or非公平模式的，如果不設置（默認）或設為false，則是非公平模式。如果設置true，則是公平模式

兩種構造函數的源碼如下：

private final Sync sync;

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

公平模式和非公平模式的區別，就是在於sync域是FairSync類還是NonfairSync類，這兩種子類分別對應這兩種模式。而permits參數代表許可的個數，作為這兩個類的構造函數參數傳入，源碼如下：

FairSync(int permits) {
    super(permits);
}

NonfairSync(int permits) {
    super(permits);
}

這兩個類的構造函數都是進一步調用父類Sync的構造函數：

Sync(int permits) {
    setState(permits);
}

從這裡就可以明白，許可個數就是state的值。這裡為AQS的state域賦了初值，為permits

重要結論：在Semaphore的語境中，AQS的state就表示許可的個數！對於Semaphore的任何獲取、釋放許可操作，本質上都是state的增減操作

獲取許可

響應中斷的獲取

acquire是響應中斷的獲取許可方法，有兩個重載版本。如果獲取成功，則返回；如果許可不夠而導致獲取失敗，則會進入AQS的同步隊列阻塞等待。在整個過程中如果被中斷，則會拋出中斷異常

acquire(int)

首先來看看可以獲取指定數量許可的acquire方法，其源碼如下：

public void acquire(int permits) throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    sync.acquireSharedInterruptibly(permits);
}

該方法會先檢查permits是否合法（非負數），再將後續的執行過程委託給Sync類的父類AQS的acquireSharedInterruptibly方法來執行，其源碼如下：

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

該方法響應中斷，首先檢查中斷狀態，如果已經被中斷則會拋出中斷異常。接下來調用鉤子方法tryAcquireShared。如果返回值小於0，說明獲取許可失敗，會調用doAcquireSharedInterruptibly進入同步隊列阻塞等待，等待過程中響應中斷

tryAcquireShared由Sync的兩個子類實現，分別對應公平模式、非公平模式下的獲取。這裡由於許可是共享資源，所以使用到的AQS的鉤子方法都是針對共享資源的獲取、釋放的。這也很合理，因為許可是可以被多個執行緒同時持有的，所以Semaphore中的許可是一種共享資源！

接下來分別看一看公平模式和非公平模式下，tryAcquireShared的實現方式是怎樣的

公平模式

FairSync類實現的tryAcquireShared方法如下：

protected int tryAcquireShared(int acquires) {
    for (;;) {
        if (hasQueuedPredecessors())	// 如果發現有執行緒在等待獲取許可，就選擇謙讓
            return -1;
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 || compareAndSetState(available, remaining))
            // 如果獲取許可數大於當前已有的許可數，獲取失敗，且返回值小於0
            // 如果CAS失敗，則重新循環直到獲取成功
            // 如果CAS成功，說明獲取成功，返回剩餘的許可數
            return remaining;	
    }
}

總體上是一個for循環，這是為了應對CAS失敗的情況。首先判斷是否有執行緒在等待獲取許可，如果有就選擇謙讓，這裡體現了公平性
接下來判斷獲取之後剩餘的許可數是否合法，如果小於0，說明沒有足夠的許可，獲取失敗，返回值小於0；如果大於0且CAS修改state成功，說明獲取許可成功，返回剩餘的許可數

這裡需要說明一下tryAcquireShared的返回值含義，這個其實在《全網最詳細的ReentrantReadWriteLock源碼剖析（萬字長文）》也有講解過：

負數：獲取失敗，執行緒會進入同步隊列阻塞等待

0：獲取成功，但是後續以共享模式獲取的執行緒都不可能獲取成功

正數：獲取成功，且後續以共享模式獲取的執行緒也可能獲取成功

非公平模式

NonfairSync類實現的tryAcquireShared方法如下：

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}

實際上委託給了父類Sync的nonfairTryAcquireShared方法來執行：

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

該方法是Sync類中的一個final方法，禁止子類重寫。邏輯上和公平模式的tryAcquireShared基本一致，只是沒有調用hasQueuedPredecessors，即使有其他執行緒在等待獲取許可，也不會謙讓，而是直接CAS競爭。這裡體現了非公平性

acquire()

這個不帶參數的acquire方法，默認獲取一個許可：

public void acquire() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);	// 默認獲取一個許可
}

和acquire(int)基本上是一樣的，只是這裡獲取數固定為1。在公平模式和非公平模式下的獲取方式也不相同。這裡不再解釋

忽略中斷的獲取

acquireUninterruptibly是忽略中斷的獲取許可方法，也有兩個重載版本。如果獲取成功，則返回；如果許可不夠而導致獲取失敗，則會進入AQS的同步隊列阻塞等待。在整個過程中如果被中斷，不會拋出中斷異常，只會記錄中斷狀態

acquireUninterruptibly(int)

首先來看看可以獲取指定數量許可的acquireUninterruptibly方法：

public void acquireUninterruptibly(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    sync.acquireShared(permits);
}

該方法會先檢查permits是否合法（非負數），再將後續的執行過程委託給Sync類的父類AQS的acquireShared方法來執行，其源碼如下：

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

這裡同樣會調用子類實現的tryAcquireShared方法，對於公平模式和非公平模式下的獲取方式略有不同，在上面都已經分析過，這裡不重複解釋了
如果tryAcquireShared獲取成功，則直接返回；如果獲取失敗，則調用doAcquireShared方法，進入同步隊列阻塞等待，在等待過程中忽略中斷，只記錄中斷狀態<

和響應中斷的acquire方法相比，唯一的區別就在於如果獲取失敗，acquire調用的是doAcquireSharedInterruptibly，響應中斷；而則這裡的acquireUninterruptibly調用的是doAcquireShared，忽略中斷

acquireUninterruptibly()

這個不帶參數的acquireUninterruptibly方法，默認獲取一個許可：

public void acquireUninterruptibly() {
    sync.acquireShared(1);	// 默認獲取一個許可
}

和acquireUninterruptibly(int)基本上是一樣的，只是這裡獲取數固定為1。在公平模式和非公平模式下的獲取方式也不相同。這裡也不再解釋

獲取剩餘所有許可

這個是Semaphore中比較特殊的一個獲取資源的方式，它提供了drainPermits方法，可以直接獲取當前剩餘的所有許可（資源），並返回獲得的個數。其源碼如下：

public int drainPermits() {
    return sync.drainPermits();
}

該方法實際上委託給了Sync類的drainPermits方法來執行：

final int drainPermits() {
    for (;;) {
        int current = getState();
        if (current == 0 || compareAndSetState(current, 0))
            return current;
    }
}

該方法是一個final方法，禁止子類重寫。整體上是一個for循環，為了應對CAS失敗的情況
首先通過AQS的getState方法獲取當前可用的許可數，再一次性全部獲取，並返回獲得的許可數，很簡單~

釋放許可

Semaphore提供了release作為釋放許可的方法，和獲取許可一樣，release也有兩個重載版本。但是，釋放許可和獲取有兩點不同：

釋放許可的方法都是忽略異常的，沒有響應異常的版本
對於公平模式和非公平模式來說，釋放許可的方式都是一樣的，因此在Sync類這一層就提供了統一的實現

release(int)

首先來看看釋放指定數量許可的release方法：

public void release(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    sync.releaseShared(permits);
}

該方法會先檢查permits是否合法（非負數），再將後續的執行過程委託給Sync類的父類AQS的releaseShared方法來執行，其源碼如下：

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

releaseShared會先調用子類重寫的tryReleaseShared方法，在公平模式和非公平模式下的釋放許可邏輯是一致的，因此在Sync類就對其進行了統一的實現，而沒有下放到子類中實現。Sync類的tryReleaseShared方法如下：

protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        int current = getState();
        int next = current + releases;
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        if (compareAndSetState(current, next))
            return true;
    }
}

這也是一個final方法，禁止子類重寫。整體上是一個for循環，為了應對CAS失敗的情況
循環體內對AQS的state進行修改，不過需要避免釋放許可後導致state溢出，否則會拋出錯誤

使用Semaphore的一點注意事項
從relsease(int)的邏輯中，並沒有發現對釋放的數量有所檢查，即一個執行緒可以釋放任意數量的許可，而不管它真正擁有多少許可
比如：一個執行緒可以釋放100個許可，即使它沒有獲得任何許可，這樣必然會導致程式錯誤
因此，使用Semaphore必須遵守「釋放許可的數量一定不能超過當前持有許可的數量」這一規定，即使Semaphore不會對其進行檢查！

源碼中的注釋也指出了這一點：

There is no requirement that a thread that releases a permit must have acquired that permit by calling acquire. Correct usage of a semaphore is established by programming convention in the application.

release()

這個版本的release默認釋放一個許可：

public void release() {
    sync.releaseShared(1);
}

其他和release(int)一致，這裡不再解釋，不過使用release()也必須注意遵守上面的規定，Semaphore也不會主動進行檢查

嘗試獲取許可

Semaphore雖然提供了阻塞版本的獲取方式，也提供了四個版本的嘗試獲取方式，包括兩種：一種是非計時等待版本，一種是計時等待版本

非計時等待

tryAcquire(int)

public boolean tryAcquire(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    return sync.nonfairTryAcquireShared(permits) >= 0;
}

該方法會先檢查permits是否合法（非負數），再將後續的執行過程委託給Sync類的nonfairTryAcquireShared方法來執行。此方法就是非公平版本的嘗試獲取許可方式，即使當前Semaphore使用的是公平策略。如果返回值不小於0，說明獲取成功，返回true；否則獲取失敗，返回false。不管成功與否，都會立即返回，不會阻塞等待

tryAcquire()

public boolean tryAcquire() {
    return sync.nonfairTryAcquireShared(1) >= 0;
}

這個就是默認獲取一個許可的tryAcquire版本，跟上面基本一樣，不解釋

計時等待

tryAcquire(int, long, TimeUnit)

public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
    throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
}

該方法會先檢查permits是否合法（非負數），再將後續的執行過程委託給Sync類的父類AQS的tryAcquireSharedNanos方法來執行。該方法會嘗試獲取許可，如果獲取成功，則立即返回；如果獲取不到，則阻塞一段時間。如果等待過程中被中斷，則會拋出中斷異常

tryAcquire(long, TimeUnit)

public boolean tryAcquire(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

這個就是默認獲取一個許可的計時等待tryAcquire版本，跟上面基本一樣，不解釋

其他方法

Semaphore還提供了一些方法以獲取訊號量的狀態，比如：

當前訊號量使用的公平策略
當前可獲取的許可數量
是否有執行緒正在等待獲取許可
因為獲取許可而被阻塞的執行緒數

下面一一來看

isFair

public boolean isFair() {
    return sync instanceof FairSync;
}

如果Semaphore的sync域是FariSync類對象，則說明使用的是公平策略，返回true，否則返回false

availablePermits

public int availablePermits() {
    return sync.getPermits();
}

final int getPermits() {
    return getState();
}

本質上調用的就是AQS的getState方法，返回state的值，而state就代表了當前可獲取的許可數量

hasQueuedThreads

public final boolean hasQueuedThreads() {
    return sync.hasQueuedThreads();
}

// AQS.hasQueuedThreads
public final boolean hasQueuedThreads() {
    return head != tail;
}

本質上調用的是AQS的hasQueuedThreads方法，判斷同步隊列的head和tail是否相同，如果相同，則說明隊列為空，沒有執行緒正在等待；否則說明有執行緒正在等待

getQueueLength

public final int getQueueLength() {
    return sync.getQueueLength();
}

// AQS.getQueueLength
public final int getQueueLength() {
    int n = 0;
    for (Node p = tail; p != null; p = p.prev) {
        if (p.thread != null)
            ++n;
    }
    return n;
}

這個方法實際調用的是AQS.getQueueLength方法。此方法會對同步隊列進行一個反向全遍歷，返回當前隊列長度的估計值。該值並不是一個確定值，因為在執行該函數時，其中的執行緒數可能已經發生了變化

Tags: JAVA 並發編程源碼閱讀 JUC源碼閱讀

Semaphore訊號量源碼解析

介紹

Semaphore是什麼

使用場景

Semaphore常用方法介紹

構造函數

獲取許可

釋放許可

嘗試獲取許可

其他方法

Semaphore使用示例

構造函數

獲取許可

響應中斷的獲取

acquire(int)

acquire()

忽略中斷的獲取

acquireUninterruptibly(int)

acquireUninterruptibly()

獲取剩餘所有許可

釋放許可

release(int)

release()

嘗試獲取許可

非計時等待

tryAcquire(int)

tryAcquire()

計時等待

tryAcquire(int, long, TimeUnit)

tryAcquire(long, TimeUnit)

其他方法

isFair

availablePermits

hasQueuedThreads

getQueueLength

VirMach 便宜 VPS

QNews

Semaphore訊號量源碼解析

介紹

Semaphore是什麼

使用場景

Semaphore常用方法介紹

構造函數

獲取許可

釋放許可

嘗試獲取許可

其他方法

Semaphore使用示例

構造函數

獲取許可

響應中斷的獲取

acquire(int)

acquire()

忽略中斷的獲取

acquireUninterruptibly(int)

acquireUninterruptibly()

獲取剩餘所有許可

釋放許可

release(int)

release()

嘗試獲取許可

非計時等待

tryAcquire(int)

tryAcquire()

計時等待

tryAcquire(int, long, TimeUnit)

tryAcquire(long, TimeUnit)

其他方法

isFair

availablePermits

hasQueuedThreads

getQueueLength

分享此文：

Related Posts

黑盒測試、白盒測試與灰盒測試方法

【PS演算法理論探討二】 Photoshop中圖層樣式之 投影樣式 演算法原理初探討。

【PS演算法理論探討二】 Photoshop中圖層樣式之 投影樣式 演算法原理初探討。

蘋果對國產屏用上癮了：iPhone 14無懸念

VirMach 便宜 VPS

QNews

熱門文章

熱門搜尋

【PS演算法理論探討二】 Photoshop中圖層樣式之投影樣式演算法原理初探討。

【PS演算法理論探討二】 Photoshop中圖層樣式之投影樣式演算法原理初探討。