Semaphore也是一個(gè)同步器，和前面兩篇說的CountDownLatch和CyclicBarrier不同，這是遞增的，初始化的時(shí)候可以指定一個(gè)值，但是不需要知道需要同步的線程個(gè)數(shù)，只需要在同步的地方調(diào)用acquire方法時(shí)指定需要同步的線程個(gè)數(shù)；
一.簡(jiǎn)單使用
同步兩個(gè)子線程，只有其中兩個(gè)子線程執(zhí)行完畢，主線程才會(huì)執(zhí)行：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
|
package com.example.demo.study;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;public class Study0217 { //創(chuàng)建一個(gè)信號(hào)量的實(shí)例，信號(hào)量初始值為0 static Semaphore semaphore = new Semaphore(0); public static void main(String[] args) throws Exception { ExecutorService pool = Executors.newFixedThreadPool(3); pool.submit(()->{ System.out.println("Thread1---start"); //信號(hào)量加一 semaphore.release(); }); pool.submit(()->{ System.out.println("Thread2---start"); //信號(hào)量加一 semaphore.release(); }); pool.submit(()->{ System.out.println("Thread3---start"); //信號(hào)量加一 semaphore.release(); }); //等待兩個(gè)子線程執(zhí)行完畢就放過，必須要信號(hào)量等于2才放過 semaphore.acquire(2); System.out.println("兩個(gè)子線程執(zhí)行完畢"); //關(guān)閉線程池，正在執(zhí)行的任務(wù)繼續(xù)執(zhí)行 pool.shutdown(); }} |
這個(gè)信號(hào)量也可以復(fù)用，類似CyclicBarrier：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
package com.example.demo.study;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;public class Study0217 { //創(chuàng)建一個(gè)信號(hào)量的實(shí)例，信號(hào)量初始值為0 static Semaphore semaphore = new Semaphore(0); public static void main(String[] args) throws Exception { ExecutorService pool = Executors.newFixedThreadPool(3); pool.submit(()->{ System.out.println("Thread1---start"); //信號(hào)量加一 semaphore.release(); }); pool.submit(()->{ System.out.println("Thread2---start"); //信號(hào)量加一 semaphore.release(); }); //等待兩個(gè)子線程執(zhí)行完畢就放過，必須要信號(hào)量等于2才放過 semaphore.acquire(2); System.out.println("子線程1，2執(zhí)行完畢"); pool.submit(()->{ System.out.println("Thread3---start"); //信號(hào)量加一 semaphore.release(); }); pool.submit(()->{ System.out.println("Thread4---start"); //信號(hào)量加一 semaphore.release(); }); semaphore.acquire(2); System.out.println("子線程3，4執(zhí)行完畢"); //關(guān)閉線程池，正在執(zhí)行的任務(wù)繼續(xù)執(zhí)行 pool.shutdown(); }} |
二.信號(hào)量原理
看看下面這個(gè)圖，可以知道信號(hào)量Semaphore還是根據(jù)AQS實(shí)現(xiàn)的，內(nèi)部有個(gè)Sync工具類操作AQS，還分為公平策略和非公平策略；
構(gòu)造器：
|
1
2
3
4
5
6
7
8
|
//默認(rèn)是非公平策略public Semaphore(int permits) { sync = new NonfairSync(permits);}//可以根據(jù)第二個(gè)參數(shù)選擇是公平策略還是非公平策略public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits);} |
acquire(int permits)方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
|
public void acquire(int permits) throws InterruptedException { if (permits < 0) throw new IllegalArgumentException(); sync.acquireSharedInterruptibly(permits);}//AQS中的方法public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); //這里根據(jù)子類是公平策略還是非公平策略 if (tryAcquireShared(arg) < 0) //獲取失敗會(huì)進(jìn)入這里，將線程放入阻塞隊(duì)列，然后再嘗試，還是失敗的話就調(diào)用park方法掛起當(dāng)前線程 doAcquireSharedInterruptibly(arg);}//非公平策略protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires);}final int nonfairTryAcquireShared(int acquires) { //一個(gè)無限循環(huán)，獲取state剩余的信號(hào)量，因?yàn)槊空{(diào)用一次release()方法的話，信號(hào)量就會(huì)加一，這里將 //最新的信號(hào)量減去傳進(jìn)來的參數(shù)比較，比如有兩個(gè)線程，其中一個(gè)線程已經(jīng)調(diào)用了release方法，然后調(diào)用acquire(2)方法，那么 //這里remaining的值就是-1,返回-1，然后當(dāng)前線程就會(huì)被丟到阻塞隊(duì)列中去了；如果另外一個(gè)線程也調(diào)用了release方法， //那么此時(shí)的remaining==0，所以在這里的if中會(huì)調(diào)用CAS將0設(shè)置到state // for (;;) { int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; }}//公平策略//和上面非公平差不多，只不過這里會(huì)查看阻塞隊(duì)列中當(dāng)前節(jié)點(diǎn)前面有沒有前驅(qū)節(jié)點(diǎn)，有的話直接返回-1，//就會(huì)把當(dāng)前線程丟到阻塞隊(duì)列中阻塞去了,沒有前驅(qū)節(jié)點(diǎn)的話，就跟非公平模式一樣的了protected int tryAcquireShared(int acquires) { for (;;) { if (hasQueuedPredecessors()) return -1; int available = getState(); int remaining = available - acquires; if (remaining < 0 ||compareAndSetState(available, remaining)) return remaining; }} |
再看看release(int permits)方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
|
//這個(gè)方法的作用就是將信號(hào)量加一public void release(int permits) { if (permits < 0) throw new IllegalArgumentException(); sync.releaseShared(permits);}//AQS中方法public final boolean releaseShared(int arg) { //tryReleaseShared嘗試釋放資源 if (tryReleaseShared(arg)) { //釋放資源成功就調(diào)用park方法喚醒喚醒AQS隊(duì)列中最前面的節(jié)點(diǎn)中的線程 doReleaseShared(); return true; } return false;}protected final boolean tryReleaseShared(int releases) { //一個(gè)無限循環(huán)，獲取state，然后加上傳進(jìn)去的參數(shù)，如果新的state的值小于舊的state，說明已經(jīng)超過了state的最大值，溢出了 //沒有溢出的話，就用CAS更新state的值 for (;;) { int current = getState(); int next = current + releases; if (next < current) // overflow throw new Error("Maximum permit count exceeded"); if (compareAndSetState(current, next)) return true; }}private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; //ws==Node.SIGNAL表示節(jié)點(diǎn)中線程需要被喚醒 if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases //調(diào)用阻塞隊(duì)列中線程的unpark方法喚醒線程 unparkSuccessor(h); } //ws == 0表示節(jié)點(diǎn)中線程是初始狀態(tài) else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) // loop if head changed break; }} |
以最上面的例子簡(jiǎn)單說一下，其實(shí)不是很難，首先線程1和線程2分別去調(diào)用release方法，這個(gè)方法里面會(huì)將AQS中的state加一，但是在執(zhí)行這個(gè)操作之前，主線程肯定會(huì)先到acquire(2)，在這個(gè)方法里面，假如默認(rèn)使用非公平策略，首先獲取當(dāng)前的信號(hào)量state(state的初始值是0)，用當(dāng)前信號(hào)量減去2,如果小于0，那么當(dāng)前主線程就會(huì)丟到AQS隊(duì)列中阻塞；
這個(gè)時(shí)候線程1的release方法執(zhí)行了，于是就把信號(hào)量state加一(此時(shí)state==1)，CAS更新state為一，成功的話，就調(diào)用doReleaseShared()方法喚醒AQS阻塞隊(duì)列中最先掛起的線程（這里就是因?yàn)檎{(diào)用acquire方法而阻塞的主線程），主線程喚醒之后又會(huì)去獲取最新的信號(hào)量，與2比較，發(fā)現(xiàn)還是小于0，于是又會(huì)阻塞；
線程2此時(shí)的release方法執(zhí)行完成，重復(fù)線程一的操作，主線程喚醒之后(此時(shí)state==2)，又去獲取最新的信號(hào)量發(fā)現(xiàn)是2，減去acquire方法的參數(shù)2等于0，于是就用CAS更新state的值，然后acquire方法也就執(zhí)行完畢，主線程繼續(xù)執(zhí)行后面的代碼；
其實(shí)信號(hào)量還是很有意思的，記得在項(xiàng)目里，有人利用信號(hào)量實(shí)現(xiàn)了一個(gè)故障隔離，什么時(shí)候我可以把整理之后的代碼貼出來分享一下，還是很有意思的，就跟springcloud的熔斷機(jī)制差不多，場(chǎng)景是：比如你在service的一個(gè)方法調(diào)用第三方的接口，你不知道調(diào)不調(diào)得通，而且你不希望每次前端過來都會(huì)去調(diào)用，比如當(dāng)調(diào)用失敗的次數(shù)超過100次，那么五分鐘之后才會(huì)再去實(shí)際調(diào)用這個(gè)第三方服務(wù)！這五分鐘內(nèi)前調(diào)用這個(gè)服務(wù)，就會(huì)觸發(fā)我們這個(gè)故障隔離的機(jī)制，向前端返回一個(gè)特定的錯(cuò)誤碼和錯(cuò)誤信息！
以上就是詳解Java 信號(hào)量Semaphore的詳細(xì)內(nèi)容，更多關(guān)于Java 信號(hào)量Semaphore的資料請(qǐng)關(guān)注服務(wù)器之家其它相關(guān)文章！
原文鏈接：https://www.cnblogs.com/wyq1995/p/12319707.html
