Java Semaphore信號量使用分析講解

更新時間：2022年12月02日 10:37:11 作者：肥肥技術宅

Semaphore實際上是一種共享鎖，因為它允許多個線程并發(fā)獲取共享的資源，在Semaphore對象創(chuàng)建時必須設置可用令牌的初始數量permits，用于控制并發(fā)時同時獲取資源權限的線程數量，這篇文章主要介紹了Java中的Semaphore如何使用,需要的朋友可以參考下

前言

大家應該都用過synchronized 關鍵字加鎖，用來保證某個時刻只允許一個線程運行。那么如果控制某個時刻允許指定數量的線程執(zhí)行，有什么好的辦法呢? 答案就是JUC提供的信號量Semaphore。

介紹和使用

Semaphore（信號量）可以用來限制能同時訪問共享資源的線程上限，它內部維護了一個許可的變量，也就是線程許可的數量

Semaphore的許可數量如果小于0個，就會阻塞獲取，直到有線程釋放許可

Semaphore是一個非重入鎖

API介紹

構造方法

public Semaphore(int permits)：permits 表示許可線程的數量
public Semaphore(int permits, boolean fair)：fair 表示公平性，如果設為 true，表示是公平，那么等待最久的線程先執(zhí)行

常用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. 創(chuàng)建 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 變?yōu)?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()
// 非公平，公平鎖會在循環(huán)內 hasQueuedPredecessors()方法判斷阻塞隊列是否有臨頭節(jié)點(第二個節(jié)點)
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();
            // 前驅節(jié)點是頭節(jié)點可以再次獲取許可
            if (p == head) {
                // 再次嘗試獲取許可，【返回剩余的許可證數量】
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 成功后本線程出隊（AQS）, 所在 Node設置為 head
                    // r 表示【可用資源數】, 為 0 則不會繼續(xù)傳播
                    setHeadAndPropagate(node, r); 
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // 不成功, 設置上一個節(jié)點 waitStatus = Node.SIGNAL, 下輪進入 park 阻塞
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        // 被打斷后進入該邏輯
        if (failed)
            cancelAcquire(node);
    }
}

最終的AQS狀態(tài)如下圖所示：

Thread-1、Thread-2、Thread-4正常運行
AQS的state也就是等于0
Thread-0、Thread-3再阻塞隊列中

釋放許可release()

現在Thread-4運行完畢，要釋放許可，Thread-0、Thread-3又是如何恢復執(zhí)行的呢？

調用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 成功, 下一個節(jié)點 unpark	
    // 如果 head.waitStatus == 0 ==> Node.PROPAGATE    
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                // 防止 unparkSuccessor 被多次執(zhí)行
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                // 喚醒后繼節(jié)點
                unparkSuccessor(h);
            }
            // 如果已經是 0 了，改為 -3，用來解決傳播性
            else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;
        }
        if (h == head)
            break;
    }
}

最終AQS狀態(tài)如下圖所示：