一、認識Semaphore

1.1、Semaphore 的使用場景

Semaphore 的使用場景主要用于流量控制，比如數(shù)據(jù)庫連接，同時使用的數(shù)據(jù)庫連接會有數(shù)量限制，數(shù)據(jù)庫連接不能超過一定的數(shù)量，當連接到達了限制數(shù)量后，后面的線程只能排隊等前面的線程釋放數(shù)據(jù)庫連接后才能獲得數(shù)據(jù)庫連接。

再比如交通公路上的紅綠燈，綠燈亮起時只能讓 100 輛車通過，紅燈亮起不允許車輛通過。

再比如停車場的場景中，一個停車場有有限數(shù)量的車位，同時能夠容納多少臺車，車位滿了之后只有等里面的車離開停車場外面的車才可以進入。

1.2、Semaphore 使用

下面我們就來模擬一下停車場的業(yè)務場景：在進入停車場之前會有一個提示牌，上面顯示著停車位還有多少，當車位為 0 時，不能進入停車場，當車位不為 0 時，才會允許車輛進入停車場。所以停車場有幾個關鍵因素：停車場車位的總容量，當一輛車進入時，停車場車位的總容量 - 1，當一輛車離開時，總容量 + 1，停車場車位不足時，車輛只能在停車場外等待。

public class CarParking {

    private static Semaphore semaphore = new Semaphore(10);

    public static void main(String[] args){

        for(int i = 0;i< 100;i++){

            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("歡迎 " + Thread.currentThread().getName() + " 來到停車場");
                    // 判斷是否允許停車
                    if(semaphore.availablePermits() == 0) {
                        System.out.println("車位不足，請耐心等待");
                    }
                    try {
                        // 嘗試獲取
                        semaphore.acquire();
                        System.out.println(Thread.currentThread().getName() + " 進入停車場");
                        Thread.sleep(new Random().nextInt(10000));// 模擬車輛在停車場停留的時間
                        System.out.println(Thread.currentThread().getName() + " 駛出停車場");
                        semaphore.release();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, i + "號車");

            thread.start();
        }

    }

}

在上面這段代碼中，我們給出了 Semaphore 的初始容量，也就是只有 10 個車位，我們用這 10 個車位來控制 100 輛車的流量，所以結果和我們預想的很相似，即大部分車都在等待狀態(tài)。但是同時仍允許一些車駛入停車場，駛入停車場的車輛，就會 semaphore.acquire 占用一個車位，駛出停車場時，就會 semaphore.release 讓出一個車位，讓后面的車再次駛入。

1.3、Semaphore 信號量的模型

上面代碼雖然比較簡單，但是卻能讓我們了解到一個信號量模型的五臟六腑。下面是一個信號量的模型：

來解釋一下 Semaphore ，Semaphore 有一個初始容量，這個初始容量就是 Semaphore 所能夠允許的信號量。在調用 Semaphore 中的 acquire 方法后，Semaphore 的容量 -1，相對的在調用 release 方法后，Semaphore 的容量 + 1，在這個過程中，計數(shù)器一直在監(jiān)控 Semaphore 數(shù)量的變化，等到流量超過 Semaphore 的容量后，多余的流量就會放入等待隊列中進行排隊等待。等到 Semaphore 的容量允許后，方可重新進入。

Semaphore 所控制的流量其實就是一個個的線程，因為并發(fā)工具最主要的研究對象就是線程。

它的工作流程如下

這幅圖應該很好理解吧，這里就不再過多解釋啦。

二、Semaphore 深入理解

在了解 Semaphore 的基本使用和 Semaphore 的模型后，下面我們還是得從源碼來和你聊一聊 Semaphore 的種種細節(jié)問題，因為我寫文章最核心的東西就是想讓我的讀者 了解 xxx，看這一篇就夠了，這是我寫文章的追求，好了話不多說，源碼走起來！

2.1、Semaphore 基本屬性

Semaphore 中只有一個屬性

private final Sync sync;

Sync 是 Semaphore 的同步實現(xiàn)，Semaphore 保證線程安全性的方式和 ReentrantLock 、CountDownLatch 類似，都是繼承于 AQS 的實現(xiàn)。同樣的，這個 Sync 也是繼承于 AbstractQueuedSynchronizer 的一個變量，也就是說，聊 Semaphore 也繞不開 AQS，所以說 AQS 真的太重要了。

2.2、Semaphore 的公平性和非公平性

那么我們進入 Sync 內部看看它實現(xiàn)了哪些方法

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;
    }
  }
}

首先是 Sync 的初始化，內部調用了 setState 并傳遞了 permits ，我們知道，AQS 中的 State 其實就是同步狀態(tài)的值，而 Semaphore 的這個 permits 就是代表了許可的數(shù)量。

getPermits 其實就是調用了 getState 方法獲取了一下線程同步狀態(tài)值。后面的 nonfairTryAcquireShared 方法其實是在 Semaphore 中構造了 NonfairSync 中的 tryAcquireShared 調用的

這里需要提及一下什么是 NonfairSync，除了 NonfairSync 是不是還有 FairSync 呢？查閱 JDK 源碼發(fā)現(xiàn)確實有。

那么這里的 FairSync 和 NonfairSync 都代表了什么？為什么會有這兩個類呢？

事實上，Semaphore 就像 ReentrantLock 一樣，也存在“公平”和"不公平"兩種，默認情況下 Semaphore 是一種不公平的信號量

Semaphore 的不公平意味著它不會保證線程獲得許可的順序，Semaphore 會在線程等待之前為調用 acquire 的線程分配一個許可，擁有這個許可的線程會自動將自己置于線程等待隊列的頭部。

當這個參數(shù)為 true 時，Semaphore 確保任何調用 acquire 的方法，都會按照先入先出的順序來獲取許可。

final int nonfairTryAcquireShared(int acquires) {
  for (;;) {
    // 獲取同步狀態(tài)值
    int available = getState();
    // state 的值 - 當前線程需要獲取的信號量（通常默認是 -1），只有
    // remaining > 0 才表示可以獲取。
    int remaining = available - acquires;
    // 先判斷是否小于 0 ，如果小于 0 則表示無法獲取，如果是正數(shù)
    // 就需要使用 CAS 判斷內存值和同步狀態(tài)值是否一致，然后更新為同步狀態(tài)值 - 1
    if (remaining < 0 ||
        compareAndSetState(available, remaining))
      return remaining;
  }
}

從上面這幅源碼對比圖可以看到，NonfairSync 和 FairSync 最大的區(qū)別就在于 tryAcquireShared 方法的區(qū)別。

NonfairSync 版本中，是不會管當前等待隊列中是否有排隊許可的，它會直接判斷信號許可量和 CAS 方法的可行性。

FairSync 版本中，它首先會判斷是否有許可進行排隊，如果有的話就直接獲取失敗。

這時候可能就會有讀者問了，你上面說公平性和非公平性的區(qū)別一直針對的是 acquire 方法來說的，怎么現(xiàn)在他們兩個主要的區(qū)別在于 tryAcquireShared 方法呢？

別急，讓我們進入到 acquire 方法一探究竟

可以看到，在 acquire 方法中，會調用 tryAcquireShared 方法，根據(jù)其返回值判斷是否調用 doAcquireSharedInterruptibly 方法。

這里需要注意下，acquire 方法具有阻塞性，而 tryAcquire 方法不具有阻塞性。

這也就是說，調用 acquire 方法如果獲取不到許可，那么 Semaphore 會阻塞，直到有可用的許可。而 tryAcquire 方法如果獲取不到許可會直接返回 false。

這里還需要注意下 acquireUninterruptibly 方法，其他 acquire 的相關方法要么是非阻塞，要么是阻塞可中斷，而 acquireUninterruptibly 方法不僅在沒有許可的情況下執(zhí)著的等待，而且也不會中斷，使用這個方法時需要注意，這個方法很容易在出現(xiàn)大規(guī)模線程阻塞而導致 Java 進程出現(xiàn)假死的情況。

有獲取許可相對應的就有釋放許可，但是釋放許可不會區(qū)分到底是公平釋放還是非公平釋放。不管方式如何都是釋放一個許可給 Semaphore ，同樣的 Semaphore 中的許可數(shù)量會增加。

在上圖中調用 tryReleaseShared 判斷是否能進行釋放后，再會調用 AQS 中的 releasedShared 方法進行釋放。

上面這個釋放流程只是釋放一個許可，除此之外，還可以釋放多個許可

public void release(int permits) {
  if (permits < 0) throw new IllegalArgumentException();
  sync.releaseShared(permits);
}

后面這個 releaseShared 的釋放流程和上面的釋放流程一致。

2.3、其他 Semaphore 方法

除了上面基本的 acquire 和 release 相關方法外，我們也要了解一下 Semaphore 的其他方法。Semaphore 的其他方法比較少，只有下面這幾個：

• drainPermits ： 獲取并退還所有立即可用的許可，其實相當于使用 CAS 方法把內存值置為 0
• reducePermits：和 nonfairTryAcquireShared 方法類似，只不過 nonfairTryAcquireShared 是使用 CAS 使內存值 + 1，而 reducePermits 是使內存值 - 1 。
• isFair：對 Semaphore 許可的爭奪是采用公平還是非公平的方式，對應到內部的實現(xiàn)就是 FairSync 和 NonfairSync。
• hasQueuedThreads：當前是否有線程由于要獲取 Semaphore 許可而進入阻塞。
• getQueuedThreads：返回一個包含了等待獲取許可的線程集合。
• getQueueLength：獲取正在排隊而進入阻塞狀態(tài)的線程個數(shù)

以上就是分析Java并發(fā)編程之信號量Semaphore的詳細內容，更多關于Java并發(fā)編程 信號量Semaphore的資料請關注腳本之家其它相關文章！

最新評論