一、Semaphore是什么
Semaphore 通常我們叫它信號量，可以用來控制同時訪問特定資源的線程數量，通過協調各個線程，以保證合理的使用資源。可以把它簡單的理解成我們停車場入口立着的那個顯示屏，每有一輛車進入停車場顯示屏就會顯示剩余車位減1，每有一輛車從停車場出去，顯示屏上顯示的剩余車輛就會加1，當顯示屏上的剩余車位為0時，停車場入口的欄桿就不會再打開，車輛就無法進入停車場了，直到有一輛車從停車場出去為止。相當於是一個計數信號量，用於控制共享資源的訪問，比如實例化時可以用N表示訪問共享資源的計數器。每訪問一次，都會將訪問的剩余次數進行減一。也是通過AQS來實現此功能的。
二、使用場景
通常用於那些資源有明確訪問數量限制的場景，常用於限流。
比如：數據庫連接池，同時進行連接的線程有數量限制，連接不能超過一定的數量，當連接達到了限制數量后，后面的線程只能排隊等前面的線程釋放了數據庫連接才能獲得數據庫連接。
比如：停車場場景，車位數量有限，同時只能容納多少台車，車位滿了之后只有等里面的車離開停車場外面的車才可以進入。
三、Semaphore常用方法說明
acquire() 獲取一個令牌，在獲取到令牌、或者被其他線程調用中斷之前線程一直處於阻塞狀態。 acquire(int permits) 獲取一個令牌，在獲取到令牌、或者被其他線程調用中斷、或超時之前線程一直處於阻塞狀態。 acquireUninterruptibly() 獲取一個令牌，在獲取到令牌之前線程一直處於阻塞狀態（忽略中斷）。 tryAcquire() 嘗試獲得令牌，返回獲取令牌成功或失敗，不阻塞線程。 tryAcquire(long timeout, TimeUnit unit) 嘗試獲得令牌，在超時時間內循環嘗試獲取，直到嘗試獲取成功或超時返回，不阻塞線程。 release() 釋放一個令牌，喚醒一個獲取令牌不成功的阻塞線程。 hasQueuedThreads() 等待隊列里是否還存在等待線程。 getQueueLength() 獲取等待隊列里阻塞的線程數。 drainPermits() 清空令牌把可用令牌數置為0，返回清空令牌的數量。 availablePermits() 返回可用的令牌數量。
四、用semaphore 實現停車場提示牌功能。
每個停車場入口都有一個提示牌，上面顯示着停車場的剩余車位還有多少，當剩余車位為0時，不允許車輛進入停車場，直到停車場里面有車離開停車場，這時提示牌上會顯示新的剩余車位數。
業務場景 ：
1、停車場容納總停車量10。
2、當一輛車進入停車場后，顯示牌的剩余車位數響應的減1.
3、每有一輛車駛出停車場后，顯示牌的剩余車位數響應的加1。
4、停車場剩余車位不足時，車輛只能在外面等待。
代碼：
public class TestCar { //停車場同時容納的車輛10 public static Semaphore semaphore = new Semaphore(10); public static void main(String[] args) { //模擬100輛車進入停車場 for(int i = 0; i < 100; i++){ Thread thread = new Thread(new Runnable() { @Override public void run() { try { System.out.println("===="+Thread.currentThread().getName()+"來到停車場"); if(semaphore.availablePermits() == 0){ System.out.println("車位不足，請耐心等待"); } //獲取令牌嘗試進入停車場 semaphore.acquire(); System.out.println(Thread.currentThread().getName()+"成功進入停車場"); //模擬車輛在停車場停留的時間 Thread.sleep(new Random().nextInt(10000)); System.out.println(Thread.currentThread().getName()+"駛出停車場"); //釋放令牌，騰出停車場車位 semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); } } },i+"號車"); thread.start(); } } }
五、Semaphore實現原理
(1)、Semaphore初始化。
Semaphore semaphore=new Semaphore(2);
1、當調用new Semaphore(2) 方法時，默認會創建一個非公平的鎖的同步阻塞隊列。
2、把初始令牌數量賦值給同步隊列的state狀態，state的值就代表當前所剩余的令牌數量。
初始化完成后同步隊列信息如下圖：
（2）獲取令牌
semaphore.acquire();
1、當前線程會嘗試去同步隊列獲取一個令牌，獲取令牌的過程也就是使用原子的操作去修改同步隊列的state ,獲取一個令牌則修改為state=state-1。
2、當計算出來的state<0，則代表令牌數量不足，此時會創建一個Node節點加入阻塞隊列，掛起當前線程。
3、當計算出來的state>=0，則代表獲取令牌成功。
源碼：
/** * 獲取1個令牌 */ public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); }
/** * 共享模式下獲取令牌，獲取成功則返回，失敗則加入阻塞隊列，掛起線程 * @param arg * @throws InterruptedException */ public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); //嘗試獲取令牌，arg為獲取令牌個數，當可用令牌數減當前令牌數結果小於0,則創建一個節點加入阻塞隊列，掛起當前線程。 if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }
/** * 1、創建節點，加入阻塞隊列， * 2、重雙向鏈表的head，tail節點關系，清空無效節點 * 3、掛起當前節點線程 * @param arg * @throws InterruptedException */ private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { //創建節點加入阻塞隊列 final Node node = addWaiter(Node.SHARED); boolean failed = true; try { for (;;) { //獲得當前節點pre節點 final Node p = node.predecessor(); if (p == head) { int r = tryAcquireShared(arg);//返回鎖的state if (r >= 0) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return; } } //重組雙向鏈表，清空無效節點，掛起當前線程 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
線程1、線程2、線程3、分別調用semaphore.acquire(),整個過程隊列信息變化如下圖：
(3)、釋放令牌
semaphore.release();
當調用semaphore.release() 方法時
1、線程會嘗試釋放一個令牌，釋放令牌的過程也就是把同步隊列的state修改為state=state+1的過程
2、釋放令牌成功之后，同時會喚醒同步隊列中的一個線程。
3、被喚醒的節點會重新嘗試去修改state=state-1 的操作，如果state>=0則獲取令牌成功，否則重新進入阻塞隊列，掛起線程。
源碼：
/** * 釋放令牌 */ public void release() { sync.releaseShared(1); }
/** *釋放共享鎖，同時會喚醒同步隊列中的一個線程。 * @param arg * @return */ public final boolean releaseShared(int arg) { //釋放共享鎖 if (tryReleaseShared(arg)) { //喚醒所有共享節點線程 doReleaseShared(); return true; } return false; }
/** * 喚醒同步隊列中的一個線程 */ private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) {//是否需要喚醒后繼節點 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))//修改狀態為初始0 continue; unparkSuccessor(h);//喚醒h.nex節點線程 } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)); } if (h == head) // loop if head changed break; } }
繼上面的圖，當我們線程1調用semaphore.release(); 時候整個流程如下圖：
六、源碼解讀
看一下Semaphore的兩個構造函數：
/** * Creates a {@code Semaphore} with the given number of * permits and nonfair fairness setting. * * @param permits the initial number of permits available. * This value may be negative, in which case releases * must occur before any acquires will be granted. */ public Semaphore(int permits) { sync = new NonfairSync(permits); } /** * Creates a {@code Semaphore} with the given number of * permits and the given fairness setting. * * @param permits the initial number of permits available. * This value may be negative, in which case releases * must occur before any acquires will be granted. * @param fair {@code true} if this semaphore will guarantee * first-in first-out granting of permits under contention, * else {@code false} */ public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); }
默認是非公平鎖。兩個構造方法都必須傳int permits值。
這個int值在實例化內部類時，被設置為AQS中的state。
Sync(int permits) { setState(permits); }
(1)、acquire()獲取信號
內部類Sync調用AQS中的acquireSharedInterruptibly()方法
public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }
調用tryAcquireShared()方法嘗試獲取信號。如果沒有可用信號，將當前線程加入等待隊列並掛起。tryAcquireShared()方法被Semaphore的內部類NonfairSync和FairSync重寫，實現有一些區別。
NonfairSync.tryAcquireShared()
final int nonfairTryAcquireShared(int acquires) { for (;;) { int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } }
可以看到，非公平鎖對於信號的獲取是直接使用CAS進行嘗試的。
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)) return remaining; } }
先調用hasQueuedPredecessors()方法，判斷隊列中是否有等待線程。如果有，直接返回-1，表示沒有可用信號。隊列中沒有等待線程，再使用CAS嘗試更新state，獲取信號。再看看acquireSharedInterruptibly()方法中，如果沒有可用信號加入隊列的方法doAcquireSharedInterruptibly()。
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.SHARED); // 1 boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head) { // 2 int r = tryAcquireShared(arg); if (r >= 0) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && // 3 parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
- 封裝一個Node節點，加入隊列尾部
- 在無限循環中，如果當前節點是頭節點，就嘗試獲取信號
- 不是頭節點，在經過節點狀態判斷后，掛起當前線程
(2)、release()釋放信號
public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { // 1 doReleaseShared(); // 2 return true; } return false; }
- 更新state加一
- 喚醒等待隊列頭節點線程
tryReleaseShared()方法在內部類Sync中被重寫
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; } }
這里也就是直接使用CAS算法，將state也就是可用信號，加1。