AQS同步組件Semaphore信號量案例剖析

更新時間：2022年08月07日 14:45:35 作者：共飲一杯無

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

基本概念

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

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

作用和使用場景

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

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

源碼分析

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

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

常用方法

acquire() 獲取一個許可

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

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

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

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

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

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

release() 釋放許可

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

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

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

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

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

使用案例

acquire()獲取單個許可

/**
     * 線程數(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 {
                    //獲取一個許可
                    semaphore.acquire();
                    test(threadNum);
                    //釋放一個許可
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        exec.shutdown();
    }
    private static void test(int threadNum) throws Exception {
        // 模擬請求的耗時操作
        Thread.sleep(1000);
        log.info("{}", threadNum);
    }

輸出結(jié)果：

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

acquire(int permits)獲取多個許可

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

輸出結(jié)果：

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

tryAcquire()獲取許可

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

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

輸出結(jié)果：

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

tryAcquire(long timeout, TimeUnit unit)

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

/**
     * 線程數(shù)量
     */
    private final static int threadCount = 15;
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        //信號量設(shè)置為3，也就是最大并發(fā)量為3，同時只允許3個線程獲得許可
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    //設(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 {
        // 模擬請求的耗時操作
        Thread.sleep(1000);
        log.info("{}", threadNum);
    }

輸出結(jié)果：