/**
 * 保存某個AQS子類實例
 */
private final Sync sync;

/**
 * 創(chuàng)建具有給定的信號量數(shù)和非公平的公平模式的 Semaphore。
 *
 * @param permits 初始的可用信號量數(shù)目。此值可能為負(fù)數(shù)，在這種情況下，必須在授予任何獲取信號量前進(jìn)行釋放信號量。
 */
public Semaphore(int permits) {
    //默認(rèn)初始化NonfairSync實例
    sync = new NonfairSync(permits);
}

/**
 * 創(chuàng)建具有給定的信號量數(shù)和給定的公平設(shè)置的 Semaphore。
 *
 * @param permits 初始的可用信號量數(shù)目。此值可能為負(fù)數(shù)，在這種情況下，必須在授予任何獲取信號量前進(jìn)行釋放信號量。
 * @param fair    如果此信號量保證在爭用時按先進(jìn)先出的順序授予信號量，則為 true；否則為 false。
 */
public Semaphore(int permits, boolean fair) {
    //根據(jù)fair參數(shù)選擇初始化一個公平FairSync類或者非公平NonfairSync類的實例
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}


/**
 * 非公平模式的實現(xiàn)
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = -2694183684443567898L;

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

    //…………其他方法后面再講

}

/**
 * 公平模式的實現(xiàn)
 */
static final class FairSync extends Sync {
    private static final long serialVersionUID = 2014338818796000944L;

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

    //…………其他方法后面再講

}

/**
 * 信號量的同步實現(xiàn)。 使用 AQS 的state狀態(tài)表示信號量。子分類為公平和非公平模式。
 */
abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 1192457210091910933L;

    /**
     * 構(gòu)造器
     *
     * @param permits 初始的可用信號量數(shù)目。
     */
    Sync(int permits) {
        //被設(shè)置為state值
        setState(permits);
    }

    //…………其他方法后面再講
}

/**
 * Semaphore的acquire方法
 * 從信號量獲取一個信號量，沒有則一直阻塞，直到在其他線程提供信號量并喚醒或者線程被中斷。
 *
 * @throws InterruptedException 如果調(diào)用此方法時已被中斷或者等待時被中斷
 */
public void acquire() throws InterruptedException {
    //內(nèi)部調(diào)用AQS的acquireSharedInterruptibly方法
    //這實際上就是共享式可中斷獲取資源模版方法
    sync.acquireSharedInterruptibly(1);
}

/**
 * 從信號量獲取permits個信號量，沒有則一直阻塞，直到在其他線程提供信號量并喚醒或者線程被中斷。
 *
 * @param permits 需要獲取的信號量數(shù)量
 * @throws InterruptedException 如果調(diào)用此方法時已被中斷或者等待時被中斷
 */
public void acquire(int permits) throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    //參數(shù)就是permits
    sync.acquireSharedInterruptibly(permits);
}

/**
 1. AQS的acquireSharedInterruptibly方法
 2. 共享式可中斷獲取信號量資源的模版方法
 3.  4. @param arg 需要獲取的信號量資源數(shù)量
 5. @throws InterruptedException 如果調(diào)用此方法時已被中斷或者等待時被中斷
 */
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    //最開始就檢查一次，如果當(dāng)前線程是被中斷狀態(tài)，直接拋出異常
    if (Thread.interrupted())
        throw new InterruptedException();
    //調(diào)用tryAcquireShared嘗試獲取共享信號量資源，這個方法是子類自己重寫的
    //如果返回值小于0，表示當(dāng)前線程共享信號量資源失敗，否則表示成功
    //Semaphore的FairSync和NonfairSync對tryAcquireShared分別做出了公平和不公平的實現(xiàn)
    if (tryAcquireShared(arg) < 0)
        //獲取不到就執(zhí)行doAcquireSharedInterruptibly方法
        doAcquireSharedInterruptibly(arg);
}

/**
 * 公平模式
 */
static final class FairSync extends Sync {
    /**
     * 嘗試公平的獲取共享信號量資源
     *
     * @param acquires 獲取信號量資源數(shù)量
     * @return 如果返回值小于0，表示當(dāng)前線程共享信號量資源失敗，否則表示成功
     */
    protected int tryAcquireShared(int acquires) {
        /*開啟一個循環(huán)嘗試獲取共享信號量資源*/
        for (; ; ) {
            //這是AQS實現(xiàn)公平模式的預(yù)定義的方法，AQS幫我們實現(xiàn)好了。該方法用于查詢是否有任何線程等待獲取信號量資源的時間超過當(dāng)前線程
            //如果該方法返回true，則表示有線程比當(dāng)前線程更早地請求獲取信號量資源。由于是公平的的，因此當(dāng)前線程不應(yīng)該獲取信號量資源，直接返回-1，表示獲取信號量資源失敗
            if (hasQueuedPredecessors())
                return -1;
            //到這里，表示當(dāng)前線程就是最早請求獲取信號量資源，可以嘗試獲取

            //獲取state的值available，我們知道state代表信號量資源數(shù)量
            int available = getState();
            //remaining為available減去需要獲取的信號量資源數(shù)量之后的差值
            int remaining = available - acquires;
            //如果remaining小于0，那么返回remaining值，由于是負(fù)數(shù)，因此獲取失敗
            //如果大于等于0，那么表示可以獲取成功，嘗試CAS的更新state，更新成功之后同樣返回remaining，由于是大于等于0的數(shù)，因此獲取成功
            if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                return remaining;
            //如果remaining大于等于0，但是CAS更新state失敗，那么循環(huán)重試
        }
    }
}

/**
 * 非公平模式
 */
static final class NonfairSync extends Sync {

    /**
     * 嘗試非公平的獲取共享信號量資源
     *
     * @param acquires 獲取信號量資源數(shù)量
     * @return 如果返回值小于0，表示當(dāng)前線程共享信號量資源失敗，否則表示成功
     */
    protected int tryAcquireShared(int acquires) {
        //調(diào)用父類Sync的nonfairTryAcquireShared方法
        //為什么該方法要實現(xiàn)在父類中的，因為無論是指定的公平模式還是非公平模式，
        //它們的tryAcquire方法都是調(diào)用的nonfairTryAcquireShared方法，即非公平的，因此實現(xiàn)在父類中
        return nonfairTryAcquireShared(acquires);
    }
}

/**
 * AQS的實現(xiàn)，作為公平和非公平模式的父類，有一些共享方法
 */
abstract static class Sync extends AbstractQueuedSynchronizer {

    /**
     * 嘗試非公平的獲取共享信號量資源
     *
     * @param acquires 獲取信號量資源數(shù)量
     * @return 如果返回值小于0，表示當(dāng)前線程共享信號量資源失敗，否則表示成功
     */
    final int nonfairTryAcquireShared(int acquires) {
        /*開啟一個循環(huán)嘗試獲取共享信號量資源*/
        for (; ; ) {
            //相比于公平模式，少了hasQueuedPredecessors的實現(xiàn)
            //獲取state的值available，我們知道state代表信號量資源數(shù)量
            int available = getState();
            //remaining為available減去需要獲取的信號量資源數(shù)量之后的差值
            int remaining = available - acquires;
            //如果remaining小于0，那么返回remaining值，由于是負(fù)數(shù)，因此獲取失敗
            //如果大于等于0，那么表示可以獲取成功，嘗試CAS的更新state，更新成功之后同樣返回remaining，由于是大于等于0的數(shù)，因此獲取成功
            if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                return remaining;
            //如果remaining大于等于0，但是CAS更新state失敗，那么循環(huán)重試
        }
    }
}

/**
 * 獲取一個信號量，沒有則一直阻塞，直到在其他線程提供信號量并喚醒該線程。
 * 獲取一個信號量就立即返回，將可用的信號量數(shù)減 1。
 */
public void acquireUninterruptibly() {
    //內(nèi)部調(diào)用AQS的acquireShared方法
    //這實際上就是共享式不可中斷獲取資源模版方法
    sync.acquireShared(1);
}

/**
 * AQS的acquireShared方法
 * 共享式不可中斷獲取資源模版方法
 *
 * @param arg 獲取的資源數(shù)量
 */
public final void acquireShared(int arg) {
    //并沒有檢查中斷
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}


/**
 * 獲取permits個信號量，沒有則一直阻塞，直到在其他線程提供信號量并喚醒該線程。
 *
 * @param permits 獲取的信號量數(shù)量
 */
public void acquireUninterruptibly(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    //參數(shù)就是permits
    sync.acquireShared(permits);
}

/**
 * @param timeout 超時時間
 * @param unit    時間單位
 * @return 是否獲取資源成功
 * @throws InterruptedException 如果調(diào)用此方法時已被中斷或者等待時被中斷
 */
public boolean tryAcquire(long timeout, TimeUnit unit)
        throws InterruptedException {
    //實際上就是調(diào)用的AQS的共享式超時獲取資源的方法，獲取1個資源
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

/**
 * @param permits 獲取的資源數(shù)量
 * @param timeout 超時時間
 * @param unit    時間單位
 * @return 是否獲取資源成功
 * @throws InterruptedException 如果調(diào)用此方法時已被中斷或者等待時被中斷
 */
public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
        throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    //實際上就是調(diào)用的AQS的共享式超時獲取資源的方法，獲取permits個資源
    return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
}

/**
 * AQS的共享式超時獲取資源的模版方法，支持中斷
 *
 * @param arg          參數(shù)
 * @param nanosTimeout 超時時間，納秒
 * @return 是否獲取資源成功
 * @throws InterruptedException 如果調(diào)用此方法時已被中斷或者等待時被中斷
 */
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    //最開始就檢查一次，如果當(dāng)前線程是被中斷狀態(tài)，直接拋出異常
    if (Thread.interrupted())
        throw new InterruptedException();
    //下面是一個||運算進(jìn)行短路連接的代碼，同樣左邊是調(diào)用子類實現(xiàn)的tryAcquireShared嘗試獲取資源，獲取到了直接返回true
    //獲取不到資源就執(zhí)行doAcquireSharedNanos方法，這個方法是AQS的方法，因此超時機制是AQS幫我們實現(xiàn)的！
    return tryAcquireShared(arg) >= 0 ||
            doAcquireSharedNanos(arg, nanosTimeout);
}

public boolean tryAcquire(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    //調(diào)用nonfairTryAcquireShared方法
    return sync.nonfairTryAcquireShared(permits) >= 0;
}

public boolean tryAcquire(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    //調(diào)用nonfairTryAcquireShared方法
    return sync.nonfairTryAcquireShared(permits) >= 0;
}

/**
 * 釋放一個信號量，信號量總數(shù)加1。
 */
public void release() {
    //內(nèi)部調(diào)用AQS的releaseShared方法
    //這實際上就是共享式釋放資源的模版方法
    sync.releaseShared(1);
}

/**
 * 釋放permits個信號量，信號量總數(shù)加permits。
 *
 * @param permits 釋放的信號量個數(shù)
 */
public void release(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    //參數(shù)就是permits
    sync.releaseShared(permits);
}


/**
 * AQS的共享模式下釋放資源的模版方法。
 * 如果成功釋放則會調(diào)用doReleaseShared
 */
public final boolean releaseShared(int arg) {
    //tryReleaseShared釋放信號量資源，該方法由子類自己實現(xiàn)
    if (tryReleaseShared(arg)) {
        //釋放成功,必定調(diào)用doReleaseShared嘗試喚醒后繼結(jié)點，即阻塞的線程
        doReleaseShared();
        return true;
    }
    return false;
}

/**
 * Sync的tryReleaseShared實現(xiàn)
 *
 * @param releases 要釋放的資源數(shù)量
 * @return true 成功 false 失敗
 */
protected final boolean tryReleaseShared(int releases) {
    for (; ; ) {
        //很簡單，就是嘗試CAS的增加state值，增加releases
        int current = getState();
        int next = current + releases;
        //這里可以知道，信號量資源數(shù)量不可超過int的最大值
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        //CAS的增加state值，CAS成功之后返回true，否則循環(huán)重試
        if (compareAndSetState(current, next))
            return true;
    }
}

/**
 * @author lx
 */
public class SemaphoreTest {

    /**
     * 機器數(shù)目，實際上就是信號量為5，非公平模式
     */
    private static Semaphore semaphore = new Semaphore(5, false);
    /**
     * 機器數(shù)目，實際上就是信號量為5，公平模式
     */
    //private static Semaphore semaphore = new Semaphore(5, true);

    /**
     * 當(dāng)所有工人都完成任務(wù)，那么統(tǒng)計工作量
     */
    private static CountDownLatch countDownLatch = new CountDownLatch(10);

    /**
     * 工人數(shù)目，8
     */
    private static final int NUM = 10;

    /**
     * 當(dāng)前時間
     */
    private static final long NOW = System.nanoTime();

    /**
     * 納秒單位
     */
    private static final long NANOUNIT = 1000000000;

    /**
     * 工作量
     */
    private static final LongAdder WORKLOAD = new LongAdder();


    static class Worker implements Runnable {
        public Worker(int num) {
            this.num = num;
        }

        private int num;
        private long timed = 20 * NANOUNIT;

        @Override
        public void run() {
            while (true) {
                //獲取信號量
                try {
                    if (semaphore.tryAcquire(timed, TimeUnit.NANOSECONDS)) {
                        System.out.println("工人" + this.num + "占用一個機器在生產(chǎn)...");
                        //占用一定時間
                        LockSupport.parkNanos((long) (NANOUNIT * num * 0.5));
                        //統(tǒng)一調(diào)整為2秒，將會看到更明顯的Semaphore效果
                        //LockSupport.parkNanos((long) (NANOUNIT * 2));

                        System.out.println("工人" + this.num + "生產(chǎn)完畢，釋放出機器");
                        //釋放信號量
                        //每個工人最多執(zhí)行20秒
                        WORKLOAD.increment();
                        if ((timed = timed - (System.nanoTime() - NOW)) <= 0) {
                            semaphore.release();
                            countDownLatch.countDown();
                            break;
                        }
                        semaphore.release();
                    } else {
                        countDownLatch.countDown();
                        break;
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < NUM; i++) {
            executorService.execute(new Worker(i));
        }
        executorService.shutdown();
        countDownLatch.await();
        System.out.println("工作完畢，空閑機器為：" + semaphore.availablePermits());
        System.out.println("總工作量為：" + WORKLOAD.sum());
    }
}