本文接著分析Semaphore的實現(xiàn)原理

Semaphore是什么

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

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號線程來了，發(fā)現(xiàn)沒有許可了，4號線程阻塞，直到1號線程調用了release后有了許可后，4號線程被喚醒，以此類推。。。

Semaphore原理

Semaphore和Reentrant一樣分別實現(xiàn)了公平鎖和非公平鎖，同樣我們看非公平鎖。上Sync內部類代碼：

abstract static class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID = 1192457210091910933L;//構造函數(shù)，接收許可的個數(shù)Sync(int permits) {setState(permits);}final int getPermits() {return getState();}//共享模式下，非公平鎖搶占final int nonfairTryAcquireShared(int acquires) {for (;;) {int available = getState();//可用許可數(shù)量減去申請許可數(shù)量int remaining = available - acquires;if (remaining < 0 ||compareAndSetState(available, remaining))//返回remaining，小于0表示許可數(shù)量不夠，大于0表示許可數(shù)量足夠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;}}//減少許可數(shù)量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;}}//清空許可數(shù)量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();//首先嘗試獲取需要的許可數(shù)量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實現(xiàn)的tryReleaseShared嘗試釋放許可。釋放成功后調用AQS的doReleaseShared，在CountdownLatch中也見過這個方法。在之前head后第一個node線程阻塞之前，已經(jīng)將head狀態(tài)設置為SIGNAL，所以會喚醒第一個node的線程，該線程繼續(xù)執(zhí)行之前的loop，嘗試獲取許可成功，并且當還有剩余的許可存在時向后傳播喚醒信號，喚醒后繼node的線程，獲取剩余的許可。

總結

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

總結

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

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