AQS同步組件Semaphore信號(hào)量怎么使用

本篇內(nèi)容介紹了“AQS同步組件Semaphore信號(hào)量怎么使用”的有關(guān)知識(shí)，在實(shí)際案例的操作過程中，不少人都會(huì)遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

基本概念

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

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

• 用于保證同一時(shí)間并發(fā)訪問線程的數(shù)目。

• 信號(hào)量在操作系統(tǒng)中是很重要的概念，Java并發(fā)庫里的Semaphore就可以很輕松的完成類似操作系統(tǒng)信號(hào)量的控制。Semaphore可以很容易控制系統(tǒng)中某個(gè)資源被同時(shí)訪問的線程個(gè)數(shù)。

• 在數(shù)據(jù)結(jié)構(gòu)中我們學(xué)過鏈表，鏈表正常是可以保存無限個(gè)節(jié)點(diǎn)的，而Semaphore可以實(shí)現(xiàn)有限大小的列表。

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

源碼分析

構(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)的線程并不一定先獲得許可，誰搶到誰就獲得許可。
 */
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í)間可以看出來同一時(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()方法，沒有設(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é)果：

tryAcquire通過參數(shù)指定了2秒的等待時(shí)間。 上述代碼中同一時(shí)間最多執(zhí)行3個(gè)。第4個(gè)線程因前3個(gè)線程執(zhí)行需要耗時(shí)一秒未釋放許可，因此需要等待。

但是由于設(shè)置了2秒的等待時(shí)間，所以在5秒內(nèi)等待到了釋放的許可，繼續(xù)執(zhí)行，循環(huán)往復(fù)。

但是15個(gè)線程 ，每秒并發(fā)3個(gè)，2S是執(zhí)行不完的。所以上面執(zhí)行到第6個(gè)（0開始，顯示是5）就結(jié)束了，【每次執(zhí)行結(jié)果會(huì)有差異，取決于CPU】，并沒有全部執(zhí)行完15個(gè)線程。

“AQS同步組件Semaphore信號(hào)量怎么使用”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！