AQS同步組件Semaphore信號量怎么使用

本篇內容介紹了“AQS同步組件Semaphore信號量怎么使用”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

基本概念

Semaphore也是一個線程同步的輔助類，可以維護當前訪問自身的線程個數，并提供了同步機制。使用Semaphore可以控制并發訪問資源的線程個數。

作用和使用場景

• 用于保證同一時間并發訪問線程的數目。

• 信號量在操作系統中是很重要的概念，Java并發庫里的Semaphore就可以很輕松的完成類似操作系統信號量的控制。Semaphore可以很容易控制系統中某個資源被同時訪問的線程個數。

• 在數據結構中我們學過鏈表，鏈表正常是可以保存無限個節點的，而Semaphore可以實現有限大小的列表。

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

源碼分析

構造函數

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

常用方法

acquire() 獲取一個許可

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

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

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

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

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

availablePermits() ： 返回此信號量中當前可用的許可證數

release() 釋放許可

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

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

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

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

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

使用案例

acquire()獲取單個許可

/**
     * 線程數量
     */
    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);
    }

輸出結果：

根據輸出結果的時間可以看出來同一時間最多只能3個線程執行，符合預期

acquire(int permits)獲取多個許可

/**
     * 線程數量
     */
    private final static int threadCount = 15;
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        //信號量設置為3，也就是最大并發量為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);
    }

輸出結果：

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

tryAcquire()獲取許可

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

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

輸出結果：

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

tryAcquire(long timeout, TimeUnit unit)

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

/**
     * 線程數量
     */
    private final static int threadCount = 15;
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        //信號量設置為3，也就是最大并發量為3，同時只允許3個線程獲得許可
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    //設置了獲取許可等待時間為2秒，如果兩秒后還是未獲得許可的線程便得不到執行
                    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);
    }

輸出結果：

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

但是由于設置了2秒的等待時間，所以在5秒內等待到了釋放的許可，繼續執行，循環往復。

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

“AQS同步組件Semaphore信號量怎么使用”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！