Java Semaphore號誌使用分析講解

2022-12-03 14:01:27

前言

大家應該都用過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)：表示一個執行緒釋放n個許可，這個數量由引數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的加鎖、解鎖。

為了更好的搞懂原理，我們通過一個例子來幫助我們理解。

假設Semaphore 的 permits為 3，這時 5 個執行緒來獲取資源，其中Thread-1，Thread-2，Thread-4CAS 競爭成功，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狀態如下圖所示：