本文接著分析Semaphore的實現原理

Semaphore是什么

Semaphore是一個計數信號量。Semaphore（信號）可以理解為一種許可，拿到許可的線程才可以繼續執行。Semaphore的計數器其實記錄的就是許可的數量，當許可數量為0時，acquire方法就會阻塞。這個系統和停車位系統非常相似，當停車場還有空位的時候，任何新來的車輛都可以進，當停車場滿的時候，新來的車輛必須要等到有空車位產生的時候才可以開進停車場。這里的停車位就相當于Semaphore的許可數量。

Semaphore的使用方法

public static void main(String[] args) throws InterruptedException{Semaphore semaphore = new Semaphore(3);for (int i = 1; i <= 5; i++) {final int threadNum = i;new Thread(() -> {try {semaphore.acquire();System.out.println("thread" + threadNum + ":entered");Thread.sleep(1000 * threadNum);System.out.println("thread" + threadNum + ":gone");semaphore.release();} catch (InterruptedException e) {System.out.println("thread" + threadNum + ":exception");}}).start();} }

看一下輸出結果：

thread2:entered thread3:entered thread1:entered thread1:gone thread4:entered thread2:gone thread5:entered thread3:gone thread4:gone thread5:goneProcess finished with exit code 0

首先我們new了一個信號量，給了3個許可，然后在新建5個線程搶占信號量。一開始1，2，3號線程可以拿到許可，然后4號線程來了，發現沒有許可了，4號線程阻塞，直到1號線程調用了release后有了許可后，4號線程被喚醒，以此類推。。。

Semaphore原理

Semaphore和Reentrant一樣分別實現了公平鎖和非公平鎖，同樣我們看非公平鎖。上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();//可用許可數量減去申請許可數量int remaining = available - acquires;if (remaining < 0 ||compareAndSetState(available, remaining))//返回remaining，小于0表示許可數量不夠，大于0表示許可數量足夠return remaining;}}//共享模式下釋放許可protected final boolean tryReleaseShared(int releases) {for (;;) {int current = getState();int next = current + releases;if (next < current) // overflowthrow new Error("Maximum permit count exceeded");if (compareAndSetState(current, next))return true;}}//減少許可數量final void reducePermits(int reductions) {for (;;) {int current = getState();int next = current - reductions;if (next > current) // underflowthrow new Error("Permit count underflow");if (compareAndSetState(current, next))return;}}//清空許可數量final int drainPermits() {for (;;) {int current = getState();if (current == 0 || compareAndSetState(current, 0))return current;}} }

了解了Semaphore對state的定義后，我們看一下acquire方法，該方法直接調用了sync的acquireSharedInterruptibly：

public final void acquireSharedInterruptibly(int arg)throws InterruptedException {//線程中斷標志為true，拋出中斷異常if (Thread.interrupted())throw new InterruptedException();//首先嘗試獲取需要的許可數量if (tryAcquireShared(arg) < 0)//當獲取失敗doAcquireSharedInterruptibly(arg); }

doAcquireSharedInterruptibly在CountdownLatch里分析過：當獲取許可失敗時，往等待隊列添加當前線程的node，如果隊列沒有初始化則初始化。然后在一個loop里從隊列head后第一個node開始嘗試獲取許可，為了不讓CPU空轉，當head后第一個node嘗試獲取許可失敗的時候，阻塞當前線程，第一個node后的弄的一樣都阻塞，等待被喚醒。

private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {final Node node = addWaiter(Node.SHARED);boolean failed = true;try {for (;;) {final Node p = node.predecessor();if (p == head) {int r = tryAcquireShared(arg);if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCfailed = false;return;}}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);} }

當調用了release方法后，意味著有新的許可被釋放，調用sync的releaseShared，接著調用Semaphore的內部類Sync實現的tryReleaseShared嘗試釋放許可。釋放成功后調用AQS的doReleaseShared，在CountdownLatch中也見過這個方法。在之前head后第一個node線程阻塞之前，已經將head狀態設置為SIGNAL，所以會喚醒第一個node的線程，該線程繼續執行之前的loop，嘗試獲取許可成功，并且當還有剩余的許可存在時向后傳播喚醒信號，喚醒后繼node的線程，獲取剩余的許可。

總結

Semaphore和CountdownLatch一樣使用了AQS的共享模式；
Semaphore在有許可釋放時喚醒第一個node的線程獲取許可，之后會根據是否還存在許可來決定是否繼續往后傳播喚醒線程的信號。
CountdownLatch在state為0的時候依次往后傳播喚醒信號，一直傳播到低，直到所有線程被喚醒。

總結

以上是生活随笔為你收集整理的AbstractQueuedSynchronizer理解之三（Semaphore）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。