生活案例
停車場(chǎng)：停車場(chǎng)只有2個(gè)車位，即同時(shí)只能容納2輛車，車輛都是停一會(huì)再走的，如何保證同一時(shí)刻最多有2個(gè)車停在停車位,？請(qǐng)用代碼實(shí)現(xiàn)。
女廁所：女廁所里只有五個(gè)位置,，即最多只能有五位女性同時(shí)上廁所,，如何保證同一時(shí)刻最多有五位女性在上廁所？請(qǐng)用java代碼實(shí)現(xiàn),。（原本就只想用廁所來說,，沒想用女廁所來舉例，但是之前遇到杠精,，說男廁可以兩個(gè)人一起上,，無比尷尬）。
多線程讀某個(gè)文件：實(shí)現(xiàn)一個(gè)文件允許同一時(shí)刻的并發(fā)訪問數(shù),。
Semaphore入場(chǎng)
上面的兩個(gè)案例可以用JDK1.5出的Semaphore來實(shí)現(xiàn),。實(shí)現(xiàn)停車場(chǎng)案例后，第二個(gè)案例也就很輕松實(shí)現(xiàn)了,。公司大樓來了6輛車,，怎么辦？保安大哥手里只有兩張停車卡（一個(gè)車位一張卡）,。
import java.util.Random;
import java.util.concurrent.Semaphore;
public class SemaphoreDemo {
//車位個(gè)數(shù)
private static final int PARK_LOT = 2;
//車輛數(shù)
private static final int CARS = 6;
private static Semaphore semaphore = new Semaphore(PARK_LOT);
private static void park() {
for (int i = 1; i <= CARS; i++) {
int finalI = i;
new Thread(() -> {
try {
//看看有沒有空車位
if (semaphore.availablePermits() == 0) {
System.out.println("第" + finalI + "輛司機(jī)看了看,，哎，還沒有空停車位，繼續(xù)排隊(duì)");
}
//嘗試進(jìn)入停車位
semaphore.acquire();
System.out.println("第" + finalI + "成功進(jìn)入停車場(chǎng)");
Thread.sleep(new Random().nextInt(10000));//模擬車輛在停車場(chǎng)停留的時(shí)間
System.out.println("第" + finalI + "駛出停車場(chǎng)");
//離開停車場(chǎng)
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
public static void main(String[] args) {
park();
}
}
運(yùn)行輸出
四輛車等待,，兩輛車進(jìn)入停車位,。然后后面就是出來一輛進(jìn)入一輛，這個(gè)停車場(chǎng)確實(shí)保證了任何時(shí)可最多可以停兩輛車了,。
可能前面排隊(duì)等著別人開出來,，可是有的人不按規(guī)矩出牌，看到有車位一下就進(jìn)去把車位搶了,。所以這個(gè)停車也分公平和非公平的,。
整個(gè)過程：
假設(shè)上面的場(chǎng)景是：司機(jī)們能看到車位是否為空，而且只能看到自己前面的車,，更前面的車看不見,。
通常有六種做法：
1.看到有空車位，便直接開過去,、不用排隊(duì)了（非公平）
2.看到有車位了，還得看看前面還有沒有車在排隊(duì)（公平）
3.不然,，就一直看車位是否為空,，以及自己前面還有沒有車（死循環(huán)--自旋）
4.能夠停車了，嘗試往空車位開去,，但卻被插隊(duì)（不公平）的那家伙搶了車位（CAS）
5.有時(shí)候保安也會(huì)說：今天車位滿了,，不讓停進(jìn)來了，回去吧（中斷）
6.保安叫你回去了,，你卻傻傻地不走,，想再等等看（中斷不響應(yīng)）
7.汽車從車位上離開，每離開一個(gè),，空車位就多一個(gè)（資源釋放）
以上是自己對(duì)生活和技術(shù)的理解,，如有不對(duì)的地方希望指正。
Semaphore翻譯
Semaphore是Doug Lea在JDK1.5時(shí)候搞出來的,。Semaphore是一種基于計(jì)數(shù)的信號(hào)量,。它可以設(shè)定一個(gè)閾值，基于此,，多個(gè)線程競(jìng)爭(zhēng)獲取許可信號(hào)，做完自己的申請(qǐng)后歸還,，超過閾值后,，線程申請(qǐng)?jiān)S可信號(hào)將會(huì)被阻塞。
UML類圖
方法和主要屬性
上面案例中我們一共使用了如下幾個(gè)關(guān)鍵方法：
new Semaphore(2);
availablePermits()
acquire();
release();
構(gòu)造方法
從構(gòu)造方法new Semaphore(PARK_LOT)為入口,，咱們一步一步把這個(gè)Semaphore的源碼給看一遍,，然后梳理一下，最后總結(jié)一下,。
//兩個(gè)車位 permits=2
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
有兩個(gè)構(gòu)造方法,，并且這里我們看到了兩個(gè)熟悉的面孔：
FairSync 公平的
NonfairSync 非公平的
上面的案例中,，使用的是一個(gè)單個(gè)參數(shù)的構(gòu)造方法，即此時(shí)使用的是非公平的,。就想上面停車的場(chǎng)景,，你排隊(duì)在前面也沒用，后面的直接插隊(duì),，停到車位上去了,。
我們來看看這個(gè)非公平NonfairSync 是怎么實(shí)現(xiàn)的
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898L;
//兩個(gè)車位 permits=2
NonfairSync(int permits) {
super(permits);
}
}
這里面什么邏輯，通過super(permits)到父類中
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 1192457210091910933L;
//兩個(gè)車位 permits=2
Sync(int permits) {
//就是給AQS設(shè)置state=2
setState(permits);
}
//返回AQS中state的值
final int getPermits() {
return getState();
}
}
構(gòu)造方法就是創(chuàng)建一個(gè)Semaphore對(duì)象,，給AQS中的state賦值,。
availablePermits() 方法--檢測(cè)是否有可用憑證（資源）
司機(jī)看看有沒有空車位
semaphore.availablePermits()
方法的源碼
public int availablePermits() {
return sync.getPermits();
}
其實(shí)這里就是獲取AQS中的sate的值。如果state==0則證明已滿,。否則semaphore.acquire();
acquire方法--中斷式
在Semaphore中獲取資源的有兩種方式：中斷式和不中斷式,。
當(dāng)前線程調(diào)用該方法的目的是希望獲取一個(gè)信號(hào)量資源。如果當(dāng)前信號(hào)量個(gè)數(shù)大于0,，則當(dāng)前信號(hào)量的計(jì)數(shù)會(huì)減1,，然后該方法直接返回。
否則如果當(dāng)前信號(hào)量個(gè)數(shù)等于0,，則當(dāng)前線程會(huì)被放入AQS的阻塞隊(duì)列,。
當(dāng)其他線程調(diào)用了當(dāng)前線程的interrupt（）方法中斷了當(dāng)前線程時(shí)，則當(dāng)前線程會(huì)拋出InterruptedException異常返回,。
其中acquire()為中斷式
public void acquire() throws InterruptedException {
//入?yún)⑹亲C明只需要一個(gè)信號(hào)
sync.acquireSharedInterruptibly(1);
}
是AQS中的方法
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//判斷是否中斷
if (Thread.interrupted()) throw new InterruptedException();
//tryAcquireShared 和 doAcquireSharedInterruptibly 方法
if (tryAcquireShared(arg) < 0){
doAcquireSharedInterruptibly(arg);
}
}
//空方法,，有子類去實(shí)現(xiàn)
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}
tryAcquireShared方法
該方法是在AQS中沒有實(shí)現(xiàn)，是一個(gè)空方法,，這里在Semaphore中有兩個(gè)實(shí)現(xiàn)類
由于我們前面使用就是非公平模式,，所以這里我們進(jìn)入NonfairSync
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898L;
NonfairSync(int permits) {
super(permits);
}
//這方法也沒有什么邏輯，直接調(diào)用nonfairTryAcquireShared
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
}
而nonfairTryAcquireShared方法則是在其父類里實(shí)現(xiàn)的
abstract static class Sync extends AbstractQueuedSynchronizer {
//acquires=1
final int nonfairTryAcquireShared(int acquires) {
//死循環(huán)--自旋
for (;;) {
//獲取AQS中的state,，就是我們前面給的permits=2
//也稱之為獲取剩余許可數(shù)
int available = getState();
int remaining = available - acquires;
// 剩余的許可小于0或者比較設(shè)置成功
if (remaining < 0 || compareAndSetState(available, remaining))
return remaining;
}
}
這里是一個(gè)CAS自旋,。因?yàn)镾emaphore是一個(gè)共享鎖，可能有多個(gè)線程同時(shí)申請(qǐng)共享資源,，因此CAS操作可能失敗,。直到成功獲取返回剩余資源數(shù)目，或者發(fā)現(xiàn)沒有剩余資源返回負(fù)值代表申請(qǐng)失敗,。在這里我們看看公平模式
static final class FairSync extends Sync {
protected int tryAcquireShared(int acquires) {
for (;;) {
//公平模式多了這個(gè)方法
if (hasQueuedPredecessors())
return -1;
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
}
公平模式下的tryAcquireShared方法在試圖獲取之前做了一個(gè)判斷,，如果發(fā)現(xiàn)等對(duì)隊(duì)列中有線程在等待獲取資源，就直接返回-1表示獲取失敗,。當(dāng)前線程會(huì)被上層的acquireSharedInterruptibly方法調(diào)用doAcquireShared方法放入等待隊(duì)列中,。這正是“公平”模式的語義：如果有線程先于我進(jìn)入等待隊(duì)列且正在等待，就直接進(jìn)入等待隊(duì)列，效果便是各個(gè)線程按照申請(qǐng)的順序獲得共享資源,，具有公平性,。
hasQueuedPredecessors這個(gè)方法我們?cè)谥暗腞eentranLock文章中說過了，這里就不在多說了,。
doAcquireSharedInterruptibly方法
這個(gè)方法我們?cè)谥暗腞eentranlock文章中也都講過了快速掌握并發(fā)編程---細(xì)說ReentrantLock和AQS,，在這里的主要是當(dāng)獲取鎖失敗后，他就是將當(dāng)前線程放入等待隊(duì)列并開始自旋檢測(cè)獲取資源,。
acquireUnInterruptibly方法-- 不中斷式
該方法與acquire()方法類似,，不同之處在于該方法對(duì)中斷不響應(yīng)，也就是當(dāng)當(dāng)前線程調(diào)用了acquireUninterruptibly獲取資源時(shí)（包含被阻塞后）,，其他線程調(diào)用了當(dāng)前線程的interrupt()方法設(shè)置了當(dāng)前線程的中斷標(biāo)志,，此時(shí)當(dāng)前線程并不會(huì)拋出InterruptedException異常而返回。
public void acquireUninterruptibly() {
sync.acquireShared(1);
}
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
release()方法--釋放資源
不管是在ReentranLock還是Semaphore里,，釋放鎖和釋放資源都不會(huì)公平性,。
該方法的作用是把當(dāng)前Semaphore對(duì)象的信號(hào)量值增加1，如果當(dāng)前有線程因?yàn)檎{(diào)用aquire方法被阻塞而被放入了AQS的阻塞隊(duì)列,，則會(huì)根據(jù)公平策略選擇一個(gè)信號(hào)量個(gè)數(shù)能被滿足的線程進(jìn)行激活,，激活的線程會(huì)嘗試獲取剛增加的信號(hào)量。
public void release() {
//注意入?yún)⑹?
sync.releaseShared(1);
}
//釋放掉資源后,，喚醒后繼
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
而在這個(gè)方法中一共調(diào)用了兩個(gè)方法tryReleaseShared和doReleaseShared
`tryReleaseShared`
tryReleaseShared方法的實(shí)現(xiàn)
既然不分公平和不公平，那么這個(gè)實(shí)現(xiàn)類就肯定是在他們兩的父類里實(shí)現(xiàn)的,，點(diǎn)進(jìn)去,，果然是在java.util.concurrent.Semaphore.Sync中實(shí)現(xiàn)的
protected final boolean tryReleaseShared(int releases) {
//自旋
for (;;) {
int current = getState();
//記得前面aquire方法是state-1
//而這里是+1，表示把憑證歸還到池子里了,，下一個(gè)人來就可以獲取憑證了
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
//CAS設(shè)置state的值=之前state的值+release
if (compareAndSetState(current, next))
return true;
}
}
`doReleaseShared`
此方法主要用于喚醒后繼
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
希望結(jié)合源碼的梳理和文章前面的說的幾種做法,，能讓您更好地理解Semaphore。
總結(jié)
核心知識(shí)點(diǎn)：利用了AQS中的state和同步阻塞隊(duì)列,、CAS,、死循環(huán)
