Java中Semaphore信號量的方法解析

更新時間：2023年12月27日 08:36:59 作者：緣來如此09

這篇文章主要介紹了Java中Semaphore信號量的方法解析,??Semaphore信號量是用來控制同?時訪問?特定?資?源的?線?程數(shù)量,它通?過協(xié)調(diào)?各個?線?程,以保證合理的使用公共?資源,需要的朋友可以參考下

一、簡介

Semaphore（信號量）是用來控制同時訪問特定資源的線程數(shù)量，它通過協(xié)調(diào) 各個線程，以保證合理的使用公共資源。實現(xiàn)其實就是一個共享鎖,是基于AQS實現(xiàn)的，通過state變量來實現(xiàn)共享。通過調(diào)用acquire方法,對state值減去一,當調(diào)用release的時候,對state值加一。當state變量小于0的時候，在AQS隊列中阻塞等待

二、使用場景

當我們需要對某個任務限制資源使用時，比如我們這個系統(tǒng) 有多個接口，其中一個接口不重要其他接口特別重要，這時候就可以通過信號量限制這個接口可使用的線程數(shù)量防止再一些特殊情況這個接口使用超量的線程資源從而影響到重要任務的執(zhí)行

三、構(gòu)造方法

Semaphore可以實現(xiàn)公平和非公平

  public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

其中NonfairSync和FairSync都是繼承自Sync，而Sync繼承于AQS所以Semaphore就是通過AQS實現(xiàn)的

abstract static class Sync extends AbstractQueuedSynchronizer {
    Sync(int permits) {
        setState(permits);
    }
}

從這里可以看出來調(diào)用了AQS的setState方法，讀過前面的文章應該明白AQS的核心就是內(nèi)部維護著一個volatile修飾的同步狀態(tài)值state。所以說當我們new Semaphore(10)時候，實際上是在AQS的框架中初始化了一個同步狀態(tài)為10的值。

四、acquire方法

public void acquire() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}
final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

通過源代碼我們發(fā)現(xiàn)此處semphore處理獲取鎖的業(yè)務邏輯是:

獲取同步狀態(tài)值
每個線程進來就減去請求的值，此處請求的值是1.然后用可用同步狀態(tài)值減去請求的值得到同步狀態(tài)剩余的值。
如果請求的值大于可用的值或者CAS操作把可用值改為剩余可用的值那么就返回剩下可用的值。

五、release()釋放鎖

Semaphore semaphore=new Semaphore(10);
semaphore.release();
public void release() {
    sync.releaseShared(1);
}

此處sync調(diào)用了AQS中的方法releaseShared，在這個方法中如果釋放成功那么就調(diào)用doReleaseShared方法，此方法在前面AQS共享模式文章中已經(jīng)講解過，此處不在詳細講解。它主要作用就是釋放隊列中的節(jié)點。

六、整體流程

附上網(wǎng)圖

七、實例代碼

  public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
        Semaphore semaphore = new Semaphore(5);
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8, 10, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10));
        for (int i = 0; i < 20; i++) {
            threadPoolExecutor.execute(() -> {
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName() + "開始執(zhí)行");
                    Thread.sleep(2000);
                    System.out.println(Thread.currentThread().getName() + "執(zhí)行成功############");
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }

輸出：

可以看出最多只有五個線程在執(zhí)行