JUC 中提供的限流利器-Semaphore（訊號量）
- 2020 年 3 月 31 日
- 筆記
在 JUC 包下，有一個 Semaphore 類，翻譯成訊號量，Semaphore（訊號量）是用來控制同時訪問特定資源的執行緒數量，它通過協調各個執行緒，以保證合理的使用公共資源。Semaphore 跟鎖（synchronized、Lock）有點相似，不同的地方是，鎖同一時刻只允許一個執行緒訪問某一資源，而 Semaphore 則可以控制同一時刻多個執行緒訪問某一資源。
Semaphore（訊號量）並不是 Java 語言特有的，幾乎所有的並發語言都有。所以也就存在一個訊號量模型的概念，如下圖所示：
訊號量模型比較簡單，可以概括為：一個計數器、一個隊列、三個方法。
計數器：記錄當前還可以運行多少個資源訪問資源。
隊列：待訪問資源的執行緒
三個方法：
- init()：初始化計數器的值，可就是允許多少執行緒同時訪問資源。
- up()：計數器加1，有執行緒歸還資源時，如果計數器的值大於或者等於 0 時，從等待隊列中喚醒一個執行緒
- down()：計數器減 1，有執行緒佔用資源時，如果此時計數器的值小於 0 ，執行緒將被阻塞。
這三個方法都是原子性的，由實現方保證原子性。例如在 Java 語言中，JUC 包下的 Semaphore 實現了訊號量模型，所以 Semaphore 保證了這三個方法的原子性。
Semaphore 是基於 AbstractQueuedSynchronizer 介面實現訊號量模型的。AbstractQueuedSynchronizer 提供了一個基於 FIFO 隊列，可以用於構建鎖或者其他相關同步裝置的基礎框架，利用了一個 int 來表示狀態，通過類似 acquire 和 release 的方式來操縱狀態。關於 AbstractQueuedSynchronizer 更多的介紹，可以點擊鏈接：
AbstractQueuedSynchronizer 在 Semaphore 類中的實現類如下：
abstract static class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 1192457210091910933L; Sync(int permits) { setState(permits); } final int getPermits() { return getState(); } final int nonfairTryAcquireShared(int acquires) { for (;;) { int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } } protected final boolean tryReleaseShared(int releases) { for (;;) { int current = getState(); int next = current + releases; if (next < current) // overflow throw new Error("Maximum permit count exceeded"); if (compareAndSetState(current, next)) return true; } } final void reducePermits(int reductions) { for (;;) { int current = getState(); int next = current - reductions; if (next > current) // underflow throw new Error("Permit count underflow"); if (compareAndSetState(current, next)) return; } } final int drainPermits() { for (;;) { int current = getState(); if (current == 0 || compareAndSetState(current, 0)) return current; } } } 在 Semaphore 類中，實現了兩種訊號量：公平的訊號量和非公平的訊號量，公平的訊號量就是大家排好隊，先到先進，非公平的訊號量就是不一定先到先進，允許插隊。非公平的訊號量效率會高一些，所以默認使用的是非公平訊號量。具體的可以查看 Semaphore 類實現源碼。
Semaphore 類中，主要有以下方法：
// 構造方法，參數表示許可證數量，用來創建訊號量 public Semaphore(int permits); // 從訊號量中獲取許可，相當於獲取到執行權 public void acquire() throws InterruptedException; // 嘗試獲取1個許可，不管是否能夠獲取成功，都立即返回，true表示獲取成功，false表示獲取失敗 public boolean tryAcquire(); // 將許可還給訊號量 public void release(); Semaphore 類的實現就了解的差不多了。可能你會有疑問 Semaphore 的應用場景是什麼？Semaphore 可以用來限流（流量控制），在一些公共資源有限的場景下，Semaphore 可以派上用場。比如在做日誌清洗時，可能有幾十個執行緒在並發清洗，但是將清洗的數據存入到資料庫時，可能只給資料庫分配了 10 個連接池，這樣兩邊的執行緒數就不對等了，我們必須保證同時只能有 10 個執行緒獲取資料庫鏈接，否則就會存在大量執行緒無法鏈接上資料庫。
用 Semaphore 訊號量來模擬這操作，程式碼如下：
public class SemaphoreDemo { /** * semaphore 訊號量，可以限流 * * 模擬並發資料庫操作，同時有三十個請求，但是系統每秒只能處理 5 個 */ private static final int THREAD_COUNT = 30; private static ExecutorService threadPool = Executors .newFixedThreadPool(THREAD_COUNT); // 初始化訊號量，個數為 5 private static Semaphore s = new Semaphore(5); public static void main(String[] args) { for (int i = 0; i < THREAD_COUNT; i++) { threadPool.execute(new Runnable() { @Override public void run() { try { // 獲取許可 s.acquire(); System.out.println(Thread.currentThread().getName()+" 完成資料庫操作 ,"+ new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format( new Date())); // 休眠兩秒鐘，效果更直觀 Thread.sleep(2000); // 釋放許可 s.release(); } catch (InterruptedException e) { } } }); } // 關閉連接池 threadPool.shutdown(); } } 運行效果如下：
從結果中，可以看出，每秒只有 5 個執行緒在執行，這符合我們的預期。
好了，關於 Semaphore 的內容就結束了，更加詳細的還請您查閱相關資料和閱讀 Semaphore 源碼。希望這篇文章對您的學習或者工作有所幫助。
感謝您的閱讀，祝好。
最後
目前互聯網上很多大佬都有 Semaphore（訊號量） 相關文章，如有雷同，請多多包涵了。原創不易，碼字不易，還希望大家多多支援。若文中有所錯誤之處，還望提出，謝謝。
