【JUC】訊號量Semaphore詳解
阿新 • • 發佈:2022-11-30
歡迎關注專欄【JAVA併發】
歡迎關注個人公眾號—— JAVA旭陽
前言
大家應該都用過synchronized
關鍵字加鎖,用來保證某個時刻只允許一個執行緒執行。那麼如果控制某個時刻允許指定數量的執行緒執行,有什麼好的辦法呢? 答案就是JUC提供的訊號量Semaphore
。
介紹和使用
-
Semaphore
(訊號量)可以用來限制能同時訪問共享資源的執行緒上限,它內部維護了一個許可的變數,也就是執行緒許可的數量 -
Semaphore
的許可數量如果小於0個,就會阻塞獲取,直到有執行緒釋放許可 -
Semaphore
是一個非重入鎖
API介紹
- 構造方法
-
public Semaphore(int permits)
permits
表示許可執行緒的數量 -
public Semaphore(int permits, boolean fair)
:fair
表示公平性,如果設為true
,表示是公平,那麼等待最久的執行緒先執行
- 常用API
-
public void acquire()
:表示一個執行緒獲取1個許可,那麼執行緒許可數量相應減少一個 -
public void release()
:表示釋放1個許可,那麼執行緒許可數量相應會增加
- 其他API
-
void acquire(int permits)
:表示一個執行緒獲取n個許可,這個數量由引數permits
決定 -
void release(int permits)
permits
決定 -
int availablePermits()
:返回當前訊號量執行緒許可數量 -
int getQueueLength()
: 返回等待獲取許可的執行緒數的預估值
基本使用
public static void main(String[] args) { // 1. 建立 semaphore 物件 Semaphore semaphore = new Semaphore(2); // 2. 10個執行緒同時執行 for (int i = 0; i < 8; i++) { new Thread(() -> { // 3. 獲取許可 try { semaphore.acquire(); } catch (InterruptedException e) { e.printStackTrace(); } try { log.debug("running..."); sleep(1); log.debug("end..."); } catch (InterruptedException e) { e.printStackTrace(); } finally { // 4. 釋放許可 semaphore.release(); } }).start(); } }
執行結果:
原理介紹
上面是Semaphore
的類結構圖,其中FairSync
和NonfairSync
是它的內部類,他們共同繼承了AQS類,AQS的共享模式提供了Semaphore
的加鎖、解鎖。
如果對AQS不瞭解的請移步深入淺出理解Java併發AQS的共享鎖模式
為了更好的搞懂原理,我們通過一個例子來幫助我們理解。
假設Semaphore
的 permits
為 3,這時 5 個執行緒來獲取資源,其中Thread-1
,Thread-2
,Thread-4
CAS 競爭成功,permits
變為 0,而 Thread-0
和 Thread-3
競爭失敗。
獲取許可acquire()
-
acquire()
主方法會呼叫sync.acquireSharedInterruptibly(1)
方法 -
acquireSharedInterruptibly()
方法會先呼叫tryAcquireShared()
方法返回許可的數量,如果小於0個,呼叫doAcquireSharedInterruptibly()
方法進入阻塞
// acquire() -> sync.acquireSharedInterruptibly(1),可中斷
public final void acquireSharedInterruptibly(int arg) {
if (Thread.interrupted())
throw new InterruptedException();
// 嘗試獲取通行證,獲取成功返回 >= 0的值
if (tryAcquireShared(arg) < 0)
// 獲取許可證失敗,進入阻塞
doAcquireSharedInterruptibly(arg);
}
-
tryAcquireShared()
方法在終會呼叫到Sync#nonfairTryAcquireShared()
方法 -
nonfairTryAcquireShared()
方法中會減去獲取的許可數量,返回剩餘的許可數量
// tryAcquireShared() -> nonfairTryAcquireShared()
// 非公平,公平鎖會在迴圈內 hasQueuedPredecessors()方法判斷阻塞佇列是否有臨頭節點(第二個節點)
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
// 獲取 state ,state 這裡【表示通行證】
int available = getState();
// 計算當前執行緒獲取通行證完成之後,通行證還剩餘數量
int remaining = available - acquires;
// 如果許可已經用完, 返回負數, 表示獲取失敗,
if (remaining < 0 ||
// 許可證足夠分配的,如果 cas 重試成功, 返回正數, 表示獲取成功
compareAndSetState(available, remaining))
return remaining;
}
}
- 如果剩餘的許可數量<0, 會呼叫
doAcquireSharedInterruptibly()
方法將當前執行緒加入到阻塞佇列中阻塞 - 方法中呼叫
parkAndCheckInterrupt()
阻塞當前執行緒
private void doAcquireSharedInterruptibly(int arg) {
// 將呼叫 Semaphore.aquire 方法的執行緒,包裝成 node 加入到 AQS 的阻塞佇列中
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) {
// 成功後本執行緒出隊(AQS), 所在 Node設定為 head
// r 表示【可用資源數】, 為 0 則不會繼續傳播
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 不成功, 設定上一個節點 waitStatus = Node.SIGNAL, 下輪進入 park 阻塞
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
// 被打斷後進入該邏輯
if (failed)
cancelAcquire(node);
}
}
最終的AQS狀態如下圖所示:
-
Thread-1
、Thread-2
、Thread-4
正常執行 - AQS的
state
也就是等於0 -
Thread-0
、Thread-3
再阻塞佇列中
釋放許可release()
現在Thread-4
執行完畢,要釋放許可,Thread-0
、Thread-3
又是如何恢復執行的呢?
- 呼叫
release()
方法釋放許可,最終呼叫Sync#releaseShared()
方法 - 如果方法
tryReleaseShared(arg)
嘗試釋放許可成功,那麼呼叫doReleaseShared();
進行喚醒
// release() -> releaseShared()
public final boolean releaseShared(int arg) {
// 嘗試釋放鎖
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
-
tryReleaseShared()
方法主要是嘗試釋放許可 - 獲取當前許可數量 + 釋放的數量,然後通過cas設定回去
protected final boolean tryReleaseShared(int releases) {
for (;;) {
// 獲取當前鎖資源的可用許可證數量
int current = getState();
int next = current + releases;
// 索引越界判斷
if (next < current)
throw new Error("Maximum permit count exceeded");
// 釋放鎖
if (compareAndSetState(current, next))
return true;
}
}
- 呼叫
doReleaseShared()
方法喚醒佇列中的執行緒 - 其中
unparkSuccessor()
方法是喚醒的核心操作
// 喚醒
private void doReleaseShared() {
// 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一個節點 unpark
// 如果 head.waitStatus == 0 ==> Node.PROPAGATE
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
// 防止 unparkSuccessor 被多次執行
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
// 喚醒後繼節點
unparkSuccessor(h);
}
// 如果已經是 0 了,改為 -3,用來解決傳播性
else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
if (h == head)
break;
}
}
最終AQS狀態如下圖所示:
- 許可state變回1
- 然後
Thread-0
開始競爭,如果競爭成功,如下圖所示:
- 由於Thread-0競爭成功,再次獲取到許可,許可數量減1,最終又變回0
- 然後等待佇列中剩餘
Thread-3
總結
Semaphore
訊號量類基於AQS的共享鎖實現,有公平鎖和非公平鎖兩個版本,它用來限制能同時訪問共享資源的執行緒上限,典型的應用場景是可以用來保護有限的公共資源,比如資料庫連線等。
如果本文對你有幫助的話,請留下一個贊吧