深入淺出Semaphore原始碼解析
阿新 • • 發佈：2020-07-18
> `Semaphore`通過permits的值來限制執行緒訪問臨界資源的總數，屬於有限制次數的共享鎖，不支援重入。
## 前提條件
在理解`Semaphore`時需要具備一些基本的知識：
### 理解AQS的實現原理
之前有寫過一篇[《深入淺出AQS原始碼解析》](https://www.cnblogs.com/pinxiong/p/13288201.html)關於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);
}
}
```
## 總結
+ 當`permits`初始化為`1`時，`Semaphore`變成了一個互斥的排他鎖
+ 當`permits`初始化為無窮大時，`Semaphore`變成了無鎖模式
+ 當`state`的值為`0`的時候，無法獲取資源，獲取資源的執行緒會進入AQS的同步佇列等待有資源釋放時被喚醒
+ 當`Semaphore`初始化成**非公平鎖**時，可能會出現有的執行緒餓死的情況，一般對於控制資源的使用而言，建議初始化為**公平鎖**
+ 可以呼叫`reducePermits`動態的縮減資源的數量，但是不能增加資源