Semaphore源码分析

Semaphore的作用是在有限的资源里，当有N个线程进行访问时，如何进行资源的分配。

基本的使用方法是，创建一个Semaphore实例，并且指定permit的个数，当线程要获取许可时调用acquire()方法，当释放许可时调用release()方法，当没有可用的许可时，线程将会进行阻塞。

Semaphore是基于AQS进行实现的，所以它底层肯定是通过自定义AQS共享模式下的同步器来实现的，该同步器需要重写AQS提供的tryAcquireShared()和tryReleaseShared()方法，只需要告诉AQS是否尝试获取同步资源和释放同步资源成功即可。

AQS子类需要定义以及维护同步状态的值，在Semaphore中，同步状态state的值为同步资源的个数。

Semaphore源码分析-LMLPHP

Semaphore的结构

public class Semaphore implements java.io.Serializable {
    private static final long serialVersionUID = -3222578661600680210L;

    private final Sync sync;

    abstract static class Sync extends AbstractQueuedSynchronizer {
       // ......
    }

    /**
     * NonFair version
     */
    static final class NonfairSync extends Sync {
		// ......
    }

    /**
     * Fair version
     */
    static final class FairSync extends Sync {
        // ......
    }

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

    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

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

    public boolean tryAcquire() {
        return sync.nonfairTryAcquireShared(1) >= 0;
    }

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

    public void acquire(int permits) throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireSharedInterruptibly(permits);
    }

    public boolean tryAcquire(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.nonfairTryAcquireShared(permits) >= 0;
    }

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

    /**
     * 返回可用的许可
     */
    public int availablePermits() {
        return sync.getPermits();
    }

    /**
     * 耗尽许可
     */
    public int drainPermits() {
        return sync.drainPermits();
    }

    /**
     * 减少许可
     */
    protected void reducePermits(int reduction) {
        if (reduction < 0) throw new IllegalArgumentException();
        sync.reducePermits(reduction);
    }

    // 其他省略
}

可以看到Semaphore中定义了一个抽象同步器（Sync）、非公平同步器（NonfairSync）和公平同步器（FairSync），同时非公平同步器和公平同步器都继承抽象同步器。

同时Semaphore中存在一个全局的抽象同步器属性，然后通过构造方法来进行初始化，通过fair参数来指定到底是使用非公平的还是公平的同步器（默认非公平），通过permits参数来指定同步资源的个数。

同时Semaphore的acquire()方法将会直接调用AQS的acquireSharedInterruptibly()方法，tryAcquire()方法调用抽象同步器的nonfairTryAcquireShared()方法，release()方法直接调用AQS的releaseShared()方法。

剖析抽象同步器

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 (;;) {
            // 当前同步资源的个数 - 要获取的同步资源个数，如果大于等于0则表示获取同步资源成功，则通过CAS更新同步状态的值，然后返回剩余的可用资源个数，否则表示获取同步资源失败，返回负数
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }

    /**
     * 尝试释放同步资源
     */
    protected final boolean tryReleaseShared(int releases) {
        // 死循环保证操作最终肯定成功
        for (;;) {
            // 当前同步资源的个数 + 要释放的同步资源个数 ，然后通过CAS更新同步状态的值
            int current = getState();
            int next = current + releases;
            if (next < current) // 如果release是负数则抛出异常
                throw new Error("Maximum permit count exceeded");
            if (compareAndSetState(current, next))
                return true;
        }
    }

    /**
     * 减少许可
     */
    final void reducePermits(int reductions) {
        // 死循环保证操作最终肯定成功
        for (;;) {
            // 当前同步资源的个数 - 要减少的同步资源个数，然后通过CAS更新同步状态的值
            int current = getState();
            int next = current - reductions;
            if (next > current) // 如果reductions是负数，则抛出异常
                throw new Error("Permit count underflow");
            if (compareAndSetState(current, next))
                return;
        }
    }

    /**
     * 耗尽许可
     */
    final int drainPermits() {
        // 死循环保证操作最终肯定成功
        for (;;) {
            // 通过CAS将同步状态的值设置为0
            int current = getState();
            if (current == 0 || compareAndSetState(current, 0))
                return current;
        }
    }
}

可以看到抽象同步器的构建方法会初始化同步资源的个数，同时提供了nonfairTryAcquireShared()、tryReleaseShared()、reducePermits()、drainPermits()方法。

同时抽象同步器已经重写了AQS的tryReleaseShared()方法，因此抽象同步器的子类还需要重写AQS的tryAcquireShared()方法。

剖析非公平同步器

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

    /**
     * 显式调用父类的带permits参数的构建方法
     */
    NonfairSync(int permits) {
        super(permits);
    }

    /**
     * 尝试获取同步资源
     */
    protected int tryAcquireShared(int acquires) {
        // 直接调用抽象同步器的nonfairTryAcquireShared()方法
        return nonfairTryAcquireShared(acquires);
    }
}

可以看到非公平同步器的构造方法会显式调用父类的带permits参数的构建方法，用于初始化同步资源。

同时非公平同步器的tryAcquireShared()方法将会直接调用抽象同步器的nonfairTryAcquireShared()方法。

剖析公平同步器

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

    /**
     * 显式调用父类的带permits参数的构建方法
     */
    FairSync(int permits) {
        super(permits);
    }

    /**
     * 尝试获取同步资源
     */
    protected int tryAcquireShared(int acquires) {
        // 死循环保证操作最终肯定成功
        for (;;) {
            // 只有当当前线程就是等待队列中头节点的后继节点所封装的线程，或者当前等待队列为空或只有一个节点，才允许尝试获取同步资源，否则表示获取同步资源失败，返回负数
            if (hasQueuedPredecessors())
                return -1;

            // 当前同步资源的个数 - 要获取的同步资源个数，如果大于等于0则表示获取同步资源成功，则通过CAS更新同步状态的值，然后返回剩余的可用资源个数，否则表示获取同步资源失败，返回负数
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }
}

可以看到公平同步器的构造方法将会显示调用父类的带permits参数的构造方法，用于初始化同步资源。

同时公平同步器重写的tryAcquireShared()方法中，只有当当前线程是等待队列中头节点的后继节点所封装的线程，或者当前等待队列为空或只有一个节点时，才允许尝试获取同步资源，否则获取同步资源失败，返回负数。

非公平模式下的总结

1.当线程要获取许可时，可以直接调用Semaphore的acquire()和tryAcquire()方法。

2.如果调用acquire()方法，那么将会直接调用AQS的acquireShared()方法，该方法又会调用非公平同步器的tryAcquireShared()方法，该方法会直接调用抽象同步器提供的nonfairTryAcquireShared()方法（用当前同步资源的个数 - 要获取的同步资源个数，如果大于等于0则表示获取同步资源成功，则通过CAS更新同步状态的值，然后返回剩余的可用资源个数，否则表示获取同步资源失败，返回负数）

3.如果调用tryAcquire()方法，那么将会直接调用抽象同步器的提供的nonfairTryAcquireShared()方法尝试获取同步资源。

4.当线程要释放许可时，将会调用Semaphore的release()方法，该方法会直接调用AQS的releaseShared()方法，releaseShared()方法又会调用抽象同步器的tryReleaseShared()方法，将当前同步资源的个数 + 要释放的同步资源个数，然后通过CAS更新同步状态的值。

公平模式下的总结

1.当线程要获取许可时，可以直接调用Semaphore的acquire()和tryAcquire()方法。

2.如果调用acquire()方法，那么将会直接调用AQS的acquireShared()方法，该方法又会调用公平同步器的tryAcquireShared()方法，该方法只有当当前线程是等待队列中的头节点的后继节点所封装的线程，或者当前等待队列为空或只有一个节点时，才允许尝试获取同步资源（用当前同步资源的个数 - 要获取的同步资源个数，如果大于等于0则表示获取同步资源成功，则通过CAS更新同步状态的值，然后返回剩余的可用资源个数，否则表示获取同步资源失败，返回负数）

3.如果调用tryAcquire()方法，那么将会直接调用抽象同步器的提供的nonfairTryAcquireShared()方法尝试获取同步资源。

FAQ

非公平模式是如何保证非公平的？

主要体现在非公平同步器的tryAcquireShared()方法，直接用当前同步资源的个数 - 要获取的同步资源个数，如果大于等于0则表示获取同步资源成功，则通过CAS更新同步状态的值，然后返回剩余的可用资源个数，否则表示获取同步资源失败，返回负数。

当线程释放同步资源时，需要唤醒离头节点最近的同时等待状态不为CANCELLED的后继节点，然后在该节点尝试获取同步资源之前，如果其他线程直接调用了Semaphore的acquire()方法获取许可，由于acquire()方法直接调用AQS的acquireShared()方法，该方法又会调用非公平同步器的tryAcquireShared()方法，所以线程会直接尝试获取同步资源，这就是非公平的体现。

公平模式是如何保证公平的？

主要体现在公平同步器的tryAcquireShared()方法，只有当当前线程是等待队列中头节点的后继节点所封装的线程，或当前等待队列为空或只有一个节点时，才允许尝试获取同步资源，将当前同步资源的个数 - 要获取的同步资源个数，如果大于等于0则表示获取同步资源成功，则CAS更新同步状态的值，然后返回剩余的可用资源个数，否则表示获取同步资源失败，返回负数。

当线程释放同步资源时，需要唤醒离头节点最近的同时等待状态不为CANCELLED的后继节点，然后在该节点尝试获取同步资源之前，如果其他线程直接调用了Semaphore的acquire()方法获取许可，由于acquire()方法直接调用AQS的acquireShared()方法，而该方法又调用公平同步器的tryAcquireShared()方法，由于当前线程并非等待队列中头节点的后继节点所封装的线程，因此该线程也只能封装成Node节点，然后加入到等待队列当中。

辣鸡小篮子