深入淺出Semaphore原始碼解析

阿新 • • 發佈：2020-07-18

Semaphore通過permits的值來限制執行緒訪問臨界資源的總數，屬於有限制次數的共享鎖，不支援重入。

前提條件

在理解Semaphore時需要具備一些基本的知識：

理解AQS的實現原理

之前有寫過一篇《深入淺出AQS原始碼解析》關於AQS的文章，對AQS原理不瞭解的同學可以先看一下

Semaphore原始碼解析

Semaphore中有3個內部類，分別是Sync、NonfairSync和FairSync，Sync是NonfairSync和FairSync的抽象類，我們會從解讀Semaphore實現的功能開始入手逐漸去解析Sync、NonfairSync和FairSync的原始碼

public class Semaphore implements java.io.Serializable {

    private final Sync sync;

    /**
 

     * 初始化permits個資源

     */

    public Semaphore(int permits) {

        sync = new NonfairSync(permits);

    }

    /**

     * 初始化permits個資源，根據fair來決定是使用公平鎖還是非公平鎖的方式

     */

    public Semaphore(int permits, boolean fair) {

        sync = fair ? new FairSync(permits) : new NonfairSync(permits);

    }

    /**
 

     * 中斷方式獲取一個資源

     */

    public void acquire() throws InterruptedException {

        sync.acquireSharedInterruptibly(1);

    }

    /**

     * 非中斷方式獲取一個資源

     */

    public void acquireUninterruptibly() {

        sync.acquireShared(1);

    }

    /**

     * 嘗試獲取一個資源

     */

    public boolean tryAcquire() {
 

        return sync.nonfairTryAcquireShared(1) >= 0;

    }

    /**

     * 嘗試超時獲取一個資源

     */

    public boolean tryAcquire(long timeout, TimeUnit unit)

        throws InterruptedException {

        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));

    }

    /**

     * 釋放一個資源

     */

    public void release() {

        sync.releaseShared(1);

    }

    /**

     * 中斷方式獲取permits個資源

     */

    public void acquire(int permits) throws InterruptedException {

        if (permits < 0) throw new IllegalArgumentException();

        sync.acquireSharedInterruptibly(permits);

    }

    /**

     * 非中斷方式獲取permits個資源

     */

    public void acquireUninterruptibly(int permits) {

        if (permits < 0) throw new IllegalArgumentException();

        sync.acquireShared(permits);

    }

    /**

     * 嘗試獲取permits個資源

     */

    public boolean tryAcquire(int permits) {

        if (permits < 0) throw new IllegalArgumentException();

        return sync.nonfairTryAcquireShared(permits) >= 0;

    }

    /**

     * 嘗試超時獲取permits個資源

     */

    public boolean tryAcquire(int permits, long timeout, TimeUnit unit)

        throws InterruptedException {

        if (permits < 0) throw new IllegalArgumentException();

        return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));

    }

    /**

     * 釋放permits個資源

     */

    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();

    }

    /**

     * 縮減reduction個資源

     */

    protected void reducePermits(int reduction) {

        if (reduction < 0) throw new IllegalArgumentException();

        sync.reducePermits(reduction);

    }

}

雖然Semaphore中的方法比較多，但是都比較簡單，都是轉呼叫Sync中的方法，通過解析Sync中的原始碼來幫助大家理解這些方法是如何實現的

Sync類原始碼解析

abstract static class Sync extends AbstractQueuedSynchronizer {

    // 獲取所有可用的資源數量

    final int getPermits() {

        return getState();

    }

    // 非公平的方式嘗試獲取acquires個可用的資源

    final int nonfairTryAcquireShared(int acquires) {

        // 無限迴圈，嘗試獲取acquires個資源

        // 如果資源數量不夠，返回剩餘資源數量

        // 如果資源數量足夠且獲取成功，返回剩餘的資源數量

        for (;;) {

            int available = getState();

            int remaining = available - acquires;

            if (remaining < 0 ||

                compareAndSetState(available, remaining))

                return remaining;

        }

    }

    // 嘗試獲取releases個資源

    protected final boolean tryReleaseShared(int releases) {

        for (;;) {

            int current = getState();

            int next = current + releases;

            // 當releases不允許為負數

            if (next < current) // overflow

                throw new Error("Maximum permit count exceeded");

            // CAS操作嘗試修改state的值

            if (compareAndSetState(current, next))

                return true;

        }

    }

    // 縮減releases個資源

    final void reducePermits(int reductions) {

        for (;;) {

            int current = getState();

            int next = current - reductions;

            // 當releases不允許為負數，也就時不能通過該方法增加資源

            if (next > current) // underflow

                throw new Error("Permit count underflow");

            // CAS操作嘗試修改state的值

            if (compareAndSetState(current, next))

                return;

        }

    }

    // 清空所有的資源數量

    final int drainPermits() {

        for (;;) {

            int current = getState();

            // CAS操作嘗試將資源數量設定為0

            if (current == 0 || compareAndSetState(current, 0))

                return current;

        }

    }

}

FairSync類原始碼解析

FairSync中的原始碼很簡單，直接上程式碼

static final class FairSync extends Sync {

    FairSync(int permits) {

        super(permits);

    }

    protected int tryAcquireShared(int acquires) {

        /**

         * 具體思路如下：

         * 1、如果AQS的同步佇列中有等待的執行緒，直接獲取失敗，會加入到AQS的同步佇列中

         * 2、如果AQS的同步佇列為空，嘗試修改state的值來獲取acquires個資源

         * 3、一直重複步驟1和2，直到有結果返回才退出無限迴圈

         */

        for (;;) {

            if (hasQueuedPredecessors())

                return -1;

            int available = getState();

            int remaining = available - acquires;

            if (remaining < 0 ||

                compareAndSetState(available, remaining))

                return remaining;

        }

    }

}

NonfairSync類原始碼解析

NonfairSync中的原始碼就更簡單，解析如下：

static final class NonfairSync extends Sync {

    NonfairSync(int permits) {

        super(permits);

    }

    // 搶佔式的獲取acquires個資源

    protected int tryAcquireShared(int acquires) {

        return nonfairTryAcquireShared(acquires);

    }

}