Semaphore訊號量是用來控制同時訪問特定資源的執行緒數量，它通過協調各個執行緒，以保證合理的使用公共資源。

舉個不適合吃飯時候看的例子：比如現在有五個茅坑，有50個人要蹲坑，那麼最先搶到的五個人先蹲，剩下45個人排對等著，等有人先蹲完出來了，告訴排在第一個的人

開始前建議先去看看我之前AQS的文章，以及很多共享鎖的程式碼在之前CountDownLatch文章裡介紹過了。

看看例子：

public class SemaphoreTrain {

    static class Worker extends Thread {
        private int n;
        private Semaphore semaphore;

        public Worker(int n, Semaphore semaphore) {
            this.n = n;
            this.semaphore = semaphore;
        }

        @Override
        public void run() {
            try {
                semaphore.acquire();
                System.out.println("Worker num " + n + " use machine");
                Thread.sleep(2000);
                System.out.println("Worker num " + n + " stop use");
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        int worker = 6;    //工人數
        int machine = 4;  //機器數
        Semaphore semaphore = new Semaphore(machine);
        for (int i = 0; i < worker; i++) {
            new Worker(i, semaphore).start();
        }
    }
}
 

Worker num 1 use machine
Worker num 3 use machine
Worker num 2 use machine
Worker num 0 use machine
Worker num 0 stop use
Worker num 1 stop use
Worker num 3 stop use
Worker num 2 stop use
Worker num 4 use machine
Worker num 5 use machine
Worker num 4 stop use
Worker num 5 stop use

只允許最多四個執行緒同時執行，其它的等待執行執行緒release退出。
來看看原始碼實現：

本質上於CountDownLatch一樣屬於共享鎖，有非公平與公平兩種模式。

    private final Sync sync;

    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1192457210091910933L;

        Sync(int permits) {
            setState(permits);
        }

        final int getPermits() {
            return getState();
        }

        // 以上面的例子來說明：非公平鎖的獲取鎖邏輯是先計算出空閒機器數remaining，若其小於0直接返回remaining
        //若大於等於0，CAS賦值state，返回remaining
        //根據AQS共享鎖的規則，返回值小於0則代表無法獲取到鎖需要入同步佇列等待。若大於等於0，執行緒不會被阻塞
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) { // 為什麼要將操作放在無限for迴圈裡？迴圈CAS + volatile = 執行緒安全
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }

        // 公平與非公平模式的嘗試釋放鎖的操作相同，即這裡的tryReleaseShared
        // 還以上面例子說明：一個工人用完機器了，他需要做的就是歸還機器，就是CAS更改state值，成功就返回true
        protected final boolean tryReleaseShared(int releases) {
            for (;;) { // 這裡為什麼也放在無限for迴圈裡？迴圈CAS + volatile = 執行緒安全
                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;
            }
        }
    }
 

這裡說點我對執行緒安全的思考：同步狀態state在AQS中是volatile修飾的，確保了可見性，迴圈CAS確保了符合操作的原子性，二者結合保證了state操作的執行緒安全性。這就是上面程式碼中兩個問題的答案，以nonfairTryAcquireShared為例，一個執行緒通過getState得到了state的最新值，計算出了remaining大於0，於是CAS賦值，那麼在這些個操作過程中，state很可能已經被改變，CAS會失敗，如何處理？迴圈再來一次唄，再次獲取state最新值進行計算不就可以了。Doug Lea大神在設計JUC框架時採用了這種非鎖式的不排他的方法來確保執行緒安全。

非公平鎖

    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -2694183684443567898L;

        NonfairSync(int permits) {
            super(permits);
        }

        // 實現AQS共享鎖的tryAcquireShared方法，內部呼叫了nonfairTryAcquireShared
        protected int tryAcquireShared(int acquires) {
            return nonfairTryAcquireShared(acquires);
        }
    }

公平鎖

    static final class FairSync extends Sync {
        private static final long serialVersionUID = 2014338818796000944L;

        FairSync(int permits) {
            super(permits);
        }

        // Semaphore公平模式獲取共享鎖邏輯
        protected int tryAcquireShared(int acquires) {
            for (;;) {
                // 之前文章說過公平就是不允許插隊，實現邏輯就在hasQueuedPredecessors裡。
                // 該方法當同步佇列裡有其它等待更久的執行緒就返回true，代表你不被允許執行乖乖後面排對去
                if (hasQueuedPredecessors())
                    return -1; // AQS共享鎖規定tryAcquireShared返回值<0則代表獲取鎖失敗，構成節點放到佇列
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
    }

建構函式

    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }

    // 布林值選擇模式
    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

acquire

獲取鎖操作：

    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

//-----------------------------方法在AQS中
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException(); // 直接拋異常
//小於零沒有獲得許可，構造節點加入佇列中等待
        if (tryAcquireShared(arg) < 0)  
            doAcquireSharedInterruptibly(arg);
    }

    private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.SHARED); // 構造節點加入到佇列。節點為共享節點代表共享鎖
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg); // r表示剩下的空餘位
                    if (r >= 0) { 
                        setHeadAndPropagate(node, r); // 將當前獲取鎖的節點設定為頭節點，根據情況喚醒後繼執行緒
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
//這裡shouldParkAfterFailedAcquire會將前面節點的狀態改為Signal
//對於等待佇列中節點的waitStatus初始為0，之後由後加入的節點改為SIGNAL
//SIGNAL這個狀態它表明你有後繼節點，release時喚醒它
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException(); //直接拋異常
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

setHeadAndPropagate

    // 該方法只在doAcquireShared中呼叫。當同步佇列中head.next節點被喚醒並獲取了共享鎖，它會呼叫該方法，
    //目的1，將該節點設為新的頭節點。
    //2，嘗試喚醒後繼節點。我麼來分析分析if的這些判斷條件：若引數propagate>0（表明仍有空位可爭）且有後繼的共享節點，則喚醒後繼節點
    //or 無空缺被佔滿了，propagate <=0，但h的狀態或是新的頭節點node的狀態waitStatus<0，表明有可能有後繼節點（因為PROPAGATE也<0），
    //這種情況下我們也去呼叫喚醒操作。哪怕是h為null，表示我們無法獲取到相關資訊的情況下仍然會去嘗試喚醒node的後繼節點（若其存在的化）。
    // 可以看出這種設計是保守的，它儘可能多的去嘗試喚醒等待的執行緒，跟獨佔鎖的操作截然不同，這就造成在併發競爭下可能會發生不必要的喚醒，
    //不過這些節點總是需要被喚醒的，不如就在這裡多嘗試。

    //對於上面說的waitStatus<0情況下節點狀態為PROPAGATE的情況，該種情況下同樣無法獲知有無後繼節點
    //關於PROPAGATE：該狀態只在doReleaseShared中被設定，有執行緒釋放了共享鎖，但由於當時頭節點狀態為0，
    //無法獲知有無後繼節點，於是將頭節點狀態改為PROPAGATE），
    //那麼此時情況可能已經變了，有了後繼節點，所以我們嘗試去喚醒後繼共享節點
    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node); //將node設定成新頭節點

        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared(); // doReleaseShared方法喚醒head.next節點的執行緒
        }
    }

總之setHeadAndPropagate喚醒後繼節點的操作偏於保守，它儘可能多的去喚醒後繼的執行緒。

release

    public void release() {
        sync.releaseShared(1);
    }

    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

    private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                // 表明有後繼節點
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h); // 喚醒後繼節點
                }
                // PROPAGATE這個標記只在一處用到，即這裡，該方法在同步佇列頭節點狀waitStatus=0
                //時將其設為PROPAGATE ，本身doReleaseShared是為了釋放後繼節點的，但是當頭節點狀態為0，我們不知
                //道有沒有後繼節點，所以就採用這種方式，將頭節點標記為PROPAGATE，意味著將共享鎖的釋放傳遞下去
                //並不會對之後的操作有什麼影響，之後進隊的節點獲取不到鎖就會呼叫
                //shouldParkAfterFailedAcquire，該方法會將頭節點改為SIGNAL。
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)               // loop if head changed
                break;
        }
    }

總結

Semaphore維持了一個同步狀態state（大小初始化時設定，代表最大允許執行執行緒數），在acquire時將state減1，小於零加入同步佇列等待，大於零則CAS更改值。
Semaphore.release happen-before Semaphore.acquire。

Actions prior to "releasing" synchronizer methods such as Lock.unlock, Semaphore.release, and CountDownLatch.countDown happen-before actions subsequent to a successful "acquiring" method such as Lock.lock, Semaphore.acquire, Condition.await, and CountDownLatch.await on the same synchronizer object in another thread.