快捷導航

Java Semaphore信號量使用分析講解

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

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

前言

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

介紹和使用

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

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

Semaphore是一個非重入鎖

API介紹

構(gòu)造方法

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

常用API

public void acquire()：表示一個線程獲取1個許可，那么線程許可數(shù)量相應減少一個
public void release()：表示釋放1個許可，那么線程許可數(shù)量相應會增加

其他API

void acquire(int permits)：表示一個線程獲取n個許可，這個數(shù)量由參數(shù)permits決定
void release(int permits)：表示一個線程釋放n個許可，這個數(shù)量由參數(shù)permits決定
int availablePermits()：返回當前信號量線程許可數(shù)量
int getQueueLength(): 返回等待獲取許可的線程數(shù)的預估值

基本使用

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();
        }
    }

運行結(jié)果：

原理介紹

上面是Semaphore的類結(jié)構(gòu)圖，其中FairSync和NonfairSync是它的內(nèi)部類，他們共同繼承了AQS類，AQS的共享模式提供了Semaphore的加鎖、解鎖。

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

假設(shè)Semaphore 的 permits為 3，這時 5 個線程來獲取資源，其中Thread-1，Thread-2，Thread-4CAS 競爭成功，permits 變?yōu)?0，而 Thread-0 和 Thread-3 競爭失敗。

獲取許可acquire()

acquire()主方法會調(diào)用 sync.acquireSharedInterruptibly(1)方法
acquireSharedInterruptibly()方法會先調(diào)用tryAcquireShared()方法返回許可的數(shù)量，如果小于0個，調(diào)用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()方法在終會調(diào)用到Sync#nonfairTryAcquireShared()方法
nonfairTryAcquireShared()方法中會減去獲取的許可數(shù)量，返回剩余的許可數(shù)量

// tryAcquireShared() -> nonfairTryAcquireShared()
// 非公平，公平鎖會在循環(huán)內(nèi) hasQueuedPredecessors()方法判斷阻塞隊列是否有臨頭節(jié)點(第二個節(jié)點)
final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        // 獲取 state ，state 這里【表示通行證】
        int available = getState();
        // 計算當前線程獲取通行證完成之后，通行證還剩余數(shù)量
        int remaining = available - acquires;
        // 如果許可已經(jīng)用完, 返回負數(shù), 表示獲取失敗,
        if (remaining < 0 ||
            // 許可證足夠分配的，如果 cas 重試成功, 返回正數(shù), 表示獲取成功
            compareAndSetState(available, remaining))
            return remaining;
    }
}

如果剩余的許可數(shù)量<0，會調(diào)用doAcquireSharedInterruptibly()方法將當前線程加入到阻塞隊列中阻塞
方法中調(diào)用parkAndCheckInterrupt()阻塞當前線程

private void doAcquireSharedInterruptibly(int arg) {
    // 將調(diào)用 Semaphore.aquire 方法的線程，包裝成 node 加入到 AQS 的阻塞隊列中
    final Node node = addWaiter(Node.SHARED);
    // 獲取標記
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            // 前驅(qū)節(jié)點是頭節(jié)點可以再次獲取許可
            if (p == head) {
                // 再次嘗試獲取許可，【返回剩余的許可證數(shù)量】
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 成功后本線程出隊（AQS）, 所在 Node設(shè)置為 head
                    // r 表示【可用資源數(shù)】, 為 0 則不會繼續(xù)傳播
                    setHeadAndPropagate(node, r); 
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // 不成功, 設(shè)置上一個節(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()

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

調(diào)用release()方法釋放許可，最終調(diào)用 Sync#releaseShared()方法
如果方法tryReleaseShared(arg)嘗試釋放許可成功，那么調(diào)用doReleaseShared();進行喚醒

// release() -> releaseShared()
public final boolean releaseShared(int arg) {
    // 嘗試釋放鎖
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }    
    return false;
}

tryReleaseShared()方法主要是嘗試釋放許可
獲取當前許可數(shù)量 + 釋放的數(shù)量，然后通過cas設(shè)置回去

protected final boolean tryReleaseShared(int releases) {    
    for (;;) {
        // 獲取當前鎖資源的可用許可證數(shù)量
        int current = getState();
        int next = current + releases;
        // 索引越界判斷
        if (next < current)            
            throw new Error("Maximum permit count exceeded");        
        // 釋放鎖
        if (compareAndSetState(current, next))            
            return true;    
    }
}

調(diào)用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);
            }
            // 如果已經(jīng)是 0 了，改為 -3，用來解決傳播性
            else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;
        }
        if (h == head)
            break;
    }
}

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