Semaphore的应用场景

Semaphore是JUC包内的一个多线程协调工具，是共享锁的一种实现，非常适合解决有限资源的共享问题。Semaphore内有一个最重要的变量permits，表示许可的数量。这里许可可以理解为共享资源分配的一种量化方式。一个Semaphore创建之后，permits数量已经确定。如果有任何线程想要访问共享资源，需要申请一定量的许可，如果没有足够的许可，该线程会阻塞。共享资源访问完毕后，需要释放同样数量的许可，用于唤醒排队中的因许可不足而等待的线程。

Semaphore的使用方法

Semaphore的构造函数：

// 创建具有10个permits的Semaphore，默认使用非公平锁
Semaphore defaultSemaphore = new Semaphore(10);
//  创建具有10个permits的Semaphore，使用公平锁方式
Semaphore fairSemaphore = new Semaphore(10, true);

获取公平锁的方式：

// 消耗一个permit获取共享锁
semaphore.acquire();

// 消耗10个permits获取共享锁
semaphore.acquire(10);

释放共享锁并归还permits：

// 释放共享锁，归还一个permit
semaphore.release();

// 释放共享锁，归还10个permit
semaphore.release(10);

接下来我们分析下Semaphore的实现原理。

acquire方法

和ReentrantLock一样，Semaphore内部也有一个Sync对象继承自AQS。Sync对象具有两个子类FairSync和NonFairSync，分别对应公平锁和非公平锁的实现。

Semaphore使用state变量来保存目前可用的permits数。获取共享锁计算state变量减去消耗的permits数。如果结果小于0，说明permits不够用，线程阻塞等待，如果大于等于0，使用CAS操作更新state变量。

释放共享锁的过程也是类似，使用CAS操作将归还的permits添加到state变量即可。

获取共享锁的入口方法为acquire代码如下：

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

Sync的方法统一在最后分析。

release方法

释放并归还permits的方法为release，代码如下所示：

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

公平方式tryAcquireShared

protected int tryAcquireShared(int acquires) {
    for (;;) {
        // 如果有其他线程等待，无法获取锁
        // 这里是实现公平锁的关键
        if (hasQueuedPredecessors())
            return -1;
        int available = getState();
        int remaining = available - acquires;
        // 计算剩余permit数，并CAS设置剩余permit数
        // 如果剩余数大于等于0，则成功获取锁
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

非公平方式tryAcquireShared

和公平方式相同，只不过少了判断是否有其他线程在排队等待的步骤。

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

tryReleaseShared方法

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");
        // 一旦CAS设置新state成功，唤醒排队等待的线程
        if (compareAndSetState(current, next))
            return true;
    }
}

本文为原创内容，欢迎大家讨论、批评指正与转载。转载时请注明出处。