AQS同步組件Semaphore信號(hào)量案例剖析

更新時(shí)間：2022年08月07日 14:45:35 作者：共飲一杯無(wú)

這篇文章主要為大家介紹了AQS同步組件Semaphore信號(hào)量案例剖析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

基本概念

Semaphore也是一個(gè)線程同步的輔助類(lèi)，可以維護(hù)當(dāng)前訪問(wèn)自身的線程個(gè)數(shù)，并提供了同步機(jī)制。使用Semaphore可以控制并發(fā)訪問(wèn)資源的線程個(gè)數(shù)。

例如排隊(duì)買(mǎi)票的情況，如果只有三個(gè)窗口，那么同一時(shí)間最多也只能有三個(gè)人買(mǎi)票。第四個(gè)人來(lái)了之后就必須在后面等著，只有其他人買(mǎi)好了，才可以去相應(yīng)的窗口進(jìn)行買(mǎi)票。

作用和使用場(chǎng)景

用于保證同一時(shí)間并發(fā)訪問(wèn)線程的數(shù)目。
信號(hào)量在操作系統(tǒng)中是很重要的概念，Java并發(fā)庫(kù)里的Semaphore就可以很輕松的完成類(lèi)似操作系統(tǒng)信號(hào)量的控制。Semaphore可以很容易控制系統(tǒng)中某個(gè)資源被同時(shí)訪問(wèn)的線程個(gè)數(shù)。
在數(shù)據(jù)結(jié)構(gòu)中我們學(xué)過(guò)鏈表，鏈表正常是可以保存無(wú)限個(gè)節(jié)點(diǎn)的，而Semaphore可以實(shí)現(xiàn)有限大小的列表。

使用場(chǎng)景：僅能提供有限訪問(wèn)的資源。比如數(shù)據(jù)庫(kù)連接。

源碼分析

構(gòu)造函數(shù)

/**
*接受一個(gè)整型的數(shù)字，表示可用的許可證數(shù)量。Semaphore（10）表*示允許10個(gè)線程獲取許可證，
*也就是最大并發(fā)數(shù)是10。 
*
* @param permits 可用許可證的初始數(shù)量。
**/
public Semaphore(int permits) {
        sync = new NonfairSync(permits);
}
/**
 * 使用給定的許可數(shù)量和給定的公平性設(shè)置
 *
 * @param permits 可用許可證的初始數(shù)量。
 *
 * @param fair 指定是公平模式還是非公平模式,默認(rèn)非公平模式 . 公平模式：先啟動(dòng)的線程優(yōu)先得到
 * 許可。 非公平模式：先啟動(dòng)的線程并不一定先獲得許可，誰(shuí)搶到誰(shuí)就獲得許可。
 */
public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

常用方法

acquire() 獲取一個(gè)許可

acquire(int permits) 獲取指定個(gè)數(shù)的許可

tryAcquire()方法嘗試獲取1個(gè)許可證

tryAcquire(long timeout, TimeUnit unit) 最大等待許可的時(shí)間

tryAcquire(int permits) 獲取指定個(gè)數(shù)的許可

tryAcquire(int permits, long timeout, TimeUnit unit) 最大等待許可的時(shí)間

availablePermits() ：返回此信號(hào)量中當(dāng)前可用的許可證數(shù)

release() 釋放許可

release(int permits) 釋放指定個(gè)數(shù)的許可

int getQueueLength() 返回正在等待獲取許可證的線程數(shù)。

boolean hasQueuedThreads() 是否有線程正在等待獲取許可證。

void reducePermits(int reduction) 減少reduction個(gè)許可證。是個(gè)protected方法。

Collection getQueuedThreads() 返回所有等待獲取許可證的線程集合。是個(gè)protected方法。

使用案例

acquire()獲取單個(gè)許可

/**
     * 線程數(shù)量
     */
    private final static int threadCount = 15;
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    //獲取一個(gè)許可
                    semaphore.acquire();
                    test(threadNum);
                    //釋放一個(gè)許可
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        exec.shutdown();
    }
    private static void test(int threadNum) throws Exception {
        // 模擬請(qǐng)求的耗時(shí)操作
        Thread.sleep(1000);
        log.info("{}", threadNum);
    }

輸出結(jié)果：

根據(jù)輸出結(jié)果的時(shí)間可以看出來(lái)同一時(shí)間最多只能3個(gè)線程執(zhí)行，符合預(yù)期

acquire(int permits)獲取多個(gè)許可

/**
     * 線程數(shù)量
     */
    private final static int threadCount = 15;
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        //信號(hào)量設(shè)置為3，也就是最大并發(fā)量為3，同時(shí)只允許3個(gè)線程獲得許可
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    //獲取多個(gè)許可
                    semaphore.acquire(3);
                    test(threadNum);
                    //釋放多個(gè)許可
                    semaphore.release(3);
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        exec.shutdown();
    }
    private static void test(int threadNum) throws Exception {
        // 模擬請(qǐng)求的耗時(shí)操作
        Thread.sleep(1000);
        log.info("{}", threadNum);
    }

輸出結(jié)果：

設(shè)置了3個(gè)許可，每個(gè)線程每次獲取3個(gè)許可，因此同一時(shí)間只能有1個(gè)線程執(zhí)行。

tryAcquire()獲取許可

tryAcquire()嘗試獲取一個(gè)許可,如果未獲取到，不等待，將直接丟棄該線程不執(zhí)行

/**
     * 線程數(shù)量
     */
    private final static int threadCount = 15;
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        //信號(hào)量設(shè)置為3，也就是最大并發(fā)量為3，同時(shí)只允許3個(gè)線程獲得許可
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    //嘗試獲取一個(gè)許可,如果未獲取到，不等待，將直接丟棄該線程不執(zhí)行
                    if(semaphore.tryAcquire()) {
                        test(threadNum);
                        //釋放許可
                        semaphore.release();
                    }
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        exec.shutdown();
    }
    private static void test(int threadNum) throws Exception {
        // 模擬請(qǐng)求的耗時(shí)操作
        Thread.sleep(1000);
        log.info("{}", threadNum);
    }

輸出結(jié)果：

從輸出可以看到，在3個(gè)線程獲取到3個(gè)許可后，因?yàn)槊總€(gè)線程調(diào)用的方法要執(zhí)行1秒中，最早的一個(gè)許可也要在1S后釋放，剩下的17個(gè)線程未獲取到許可，使用了semaphore.tryAcquire()方法，沒(méi)有設(shè)置等待時(shí)間，所以便直接被丟棄，不執(zhí)行了。

tryAcquire(long timeout, TimeUnit unit)

tryAcquire(long timeout, TimeUnit unit)未獲取到許可，設(shè)置等待時(shí)長(zhǎng)

/**
     * 線程數(shù)量
     */
    private final static int threadCount = 15;
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        //信號(hào)量設(shè)置為3，也就是最大并發(fā)量為3，同時(shí)只允許3個(gè)線程獲得許可
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    //設(shè)置了獲取許可等待時(shí)間為2秒，如果兩秒后還是未獲得許可的線程便得不到執(zhí)行
                    if(semaphore.tryAcquire(2000, TimeUnit.MILLISECONDS)) {
                        test(threadNum);
                        //釋放許可
                        semaphore.release();
                    }
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        exec.shutdown();
    }
    private static void test(int threadNum) throws Exception {
        // 模擬請(qǐng)求的耗時(shí)操作
        Thread.sleep(1000);
        log.info("{}", threadNum);
    }

輸出結(jié)果：