在上一章，我們學習了信號量（Semaphore）是如何請求許可證的璧针，下面我們來看看要如何歸還許可證嚷炉。
可以看到當我們要歸還許可證時，不論是調(diào)用release()或是release(int permits)探橱，都會調(diào)用AQS實現(xiàn)的releaseShared(int arg)方法申屹。在releaseShared(int arg)方法中會先調(diào)用子類實現(xiàn)的tryReleaseShared(int arg)方法绘证，這個方法會向信號量歸還許可證，在歸還完畢后哗讥，會調(diào)用doReleaseShared()方法嘗試喚醒信號量等待隊列中需要許可證的線程嚷那，這也印證了筆者之前所說的線程在歸還信號量后，會嘗試喚醒等待隊列中等待許可證的線程杆煞。
那我們來看看信號量（Semaphore）靜態(tài)內(nèi)部類Sync實現(xiàn)的tryReleaseShared(int releases)是怎么完成歸還許可證魏宽，首先會調(diào)用getState()獲取信號量當前剩余的許可證，加上外部線程歸還的許可證數(shù)量算出總許可證數(shù)量：current + releases决乎，如果能用CAS的方式修改成功队询，則退出方法，否則一直輪詢直到歸還成功构诚，這里CAS失敗的原因有可能是外部也在請求和歸還許可證蚌斩，可能在執(zhí)行完代碼<1>處后和執(zhí)行代碼<2>處之前，信號量內(nèi)部的許可證數(shù)量已經(jīng)變了范嘱，所以CAS失敗送膳。歸還信號量成功后就會調(diào)用doReleaseShared()，這個方法前面已經(jīng)講解過了丑蛤，這里就不再贅述了肠缨。
public class Semaphore implements java.io.Serializable {
? ? //...
? ? abstract static class Sync extends AbstractQueuedSynchronizer {
? ? ? ? //...
? ? ? ? protected final boolean tryReleaseShared(int releases) {
? ? ? ? ? ? for (;;) {
? ? ? ? ? ? ? ? int current = getState();//<1>
? ? ? ? ? ? ? ? int next = current + releases;
? ? ? ? ? ? ? ? if (next < current) // overflow
? ? ? ? ? ? ? ? ? ? throw new Error("Maximum permit count exceeded");
? ? ? ? ? ? ? ? if (compareAndSetState(current, next))//<2>
? ? ? ? ? ? ? ? ? ? return true;
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? //...
? ? }
? ? //...
? ? public void release() {
? ? ? ? sync.releaseShared(1);
? ? }
? ? //...
? ? public void release(int permits) {
? ? ? ? if (permits < 0) throw new IllegalArgumentException();
? ? ? ? sync.releaseShared(permits);
? ? }
? ? //...
}
public abstract class AbstractQueuedSynchronizer
? ? extends AbstractOwnableSynchronizer
? ? implements java.io.Serializable {
? ? //...
? ? public final boolean releaseShared(int arg) {
? ? ? ? if (tryReleaseShared(arg)) {
? ? ? ? ? ? doReleaseShared();
? ? ? ? ? ? return true;
? ? ? ? }
? ? ? ? return false;
? ? }
? ? //...
? ? protected boolean tryReleaseShared(int arg) {
? ? ? ? throw new UnsupportedOperationException();
? ? }
? ? //...
}
下面我們再來看看tryAcquire(long timeout, TimeUnit unit)和tryAcquire(int permits, long timeout, TimeUnit unit)的實現(xiàn)，這兩個方法會在給定的時間范圍內(nèi)嘗試獲取許可證盏阶，如果獲取成功則返回true晒奕，獲取失敗則返回false。
這兩個方法都會調(diào)用AQS實現(xiàn)的tryAcquireSharedNanos(int arg, long nanosTimeout)名斟，這個方法其實和先前講得doAcquireShared(int arg)十分相似脑慧，只是多了一個超時返回的功能。
這里筆者簡單過一下這個方法的實現(xiàn)：先在代碼<1>處算出超時時間砰盐，然后封裝線程對應的節(jié)點Node并將其入隊闷袒，如果判斷節(jié)點的前驅(qū)節(jié)點是頭節(jié)點，且申請許可證成功岩梳，這里會調(diào)用setHeadAndPropagate(node, r)將頭節(jié)點指向當前節(jié)點囊骤，并嘗試喚醒下一個節(jié)點對應的線程。如果申請許可證失敗冀值，會在<2>處算出還剩多少的阻塞時間nanosTimeout也物，如果剩余阻塞時間小于等于0，代表線程獲取許可證失敗列疗，這里會調(diào)用<3>處的cancelAcquire(node) 將節(jié)點從等待隊列中移除滑蚯，具體的移除邏輯可以看筆者寫的ReentrantLock源碼解析第二章。如果剩余阻塞時間大于0，則會執(zhí)行shouldParkAfterFailedAcquire(p, node)將前驅(qū)節(jié)點的等待狀態(tài)改為SIGNAL告材，在第二次循環(huán)時坤次，如果前驅(qū)節(jié)點的狀態(tài)為SIGNAL，且剩余阻塞時間大于SPIN_FOR_TIMEOUT_THRESHOLD（1000ns）斥赋，則陷入阻塞缰猴，直到被中斷拋出異常，或者被喚醒疤剑，檢查是否能獲取許可證滑绒，如果不能獲取許可證且超時，則會返回false表示在超時時間內(nèi)沒有獲取到許可證骚露。
public class Semaphore implements java.io.Serializable {
? ? //...
? ? public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
? ? ? ? throws InterruptedException {
? ? ? ? if (permits < 0) throw new IllegalArgumentException();
? ? ? ? return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
? ? }
? ? //...
? ? public boolean tryAcquire(long timeout, TimeUnit unit)
? ? ? ? throws InterruptedException {
? ? ? ? return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
? ? }
? ? //...
}
public abstract class AbstractQueuedSynchronizer
? ? extends AbstractOwnableSynchronizer
? ? implements java.io.Serializable {
? ? //...
? ? public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
? ? ? ? ? ? throws InterruptedException {
? ? ? ? if (Thread.interrupted())
? ? ? ? ? ? throw new InterruptedException();
? ? ? ? return tryAcquireShared(arg) >= 0 ||
? ? ? ? ? ? doAcquireSharedNanos(arg, nanosTimeout);
? ? }
? ? //...
? ? private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
? ? ? ? ? ? throws InterruptedException {
? ? ? ? if (nanosTimeout <= 0L)
? ? ? ? ? ? return false;
? ? ? ? final long deadline = System.nanoTime() + nanosTimeout;//<1>
? ? ? ? final Node node = addWaiter(Node.SHARED);
? ? ? ? try {
? ? ? ? ? ? for (;;) {
? ? ? ? ? ? ? ? final Node p = node.predecessor();
? ? ? ? ? ? ? ? if (p == head) {
? ? ? ? ? ? ? ? ? ? int r = tryAcquireShared(arg);
? ? ? ? ? ? ? ? ? ? if (r >= 0) {
? ? ? ? ? ? ? ? ? ? ? ? setHeadAndPropagate(node, r);
? ? ? ? ? ? ? ? ? ? ? ? p.next = null; // help GC
? ? ? ? ? ? ? ? ? ? ? ? return true;
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? nanosTimeout = deadline - System.nanoTime();//<2>
? ? ? ? ? ? ? ? if (nanosTimeout <= 0L) {
? ? ? ? ? ? ? ? ? ? cancelAcquire(node);//<3>
? ? ? ? ? ? ? ? ? ? return false;
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? if (shouldParkAfterFailedAcquire(p, node) &&
? ? ? ? ? ? ? ? ? ? nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD)
? ? ? ? ? ? ? ? ? ? LockSupport.parkNanos(this, nanosTimeout);
? ? ? ? ? ? ? ? if (Thread.interrupted())
? ? ? ? ? ? ? ? ? ? throw new InterruptedException();
? ? ? ? ? ? }
? ? ? ? } catch (Throwable t) {
? ? ? ? ? ? cancelAcquire(node);
? ? ? ? ? ? throw t;
? ? ? ? }
? ? }
? ? //...
}
下面我們對照一下FairSync和NonfairSync蹬挤，其實NonfairSync基本沒有什么實現(xiàn)缚窿，都是調(diào)用其父類Sync的方法棘幸，以非公平的方式競爭許可證也是調(diào)用其父類nonfairTryAcquireShared(acquires)方法谊却。而FairSync自身是有實現(xiàn)以公平的方式獲取許可證母剥，實現(xiàn)邏輯也非常簡單。先判斷信號量的等待隊列是否有節(jié)點定拟，有的話則返回獲取失敗扫茅，如果沒有再獲取當前的可用許可證數(shù)量available蹋嵌，扣去申請的許可證數(shù)量available - acquires，用CAS的方式把扣減完的值remaining存放進state葫隙，由于扣減的時候可能存在其他線程也在申請/歸還許可證栽烂，所以available的值并非一直有效，如果在獲取available后有其他線程也申請和歸還許可證恋脚，那么這里的CAS很可能會失敗腺办，判斷CAS失敗后，又會開始新的一輪嘗試獲取許可證邏輯糟描。
static final class FairSync extends Sync {
? ? private static final long serialVersionUID = 2014338818796000944L;
? ? FairSync(int permits) {
? ? ? ? super(permits);
? ? }
? ? 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;
? ? ? ? }
? ? }
}
static final class NonfairSync extends Sync {
? ? private static final long serialVersionUID = -2694183684443567898L;
? ? NonfairSync(int permits) {
? ? ? ? super(permits);
? ? }
? ? protected int tryAcquireShared(int acquires) {
? ? ? ? return nonfairTryAcquireShared(acquires);
? ? }
}
對照完公平FairSync和非公平NonfairSync的差別后怀喉，我們來看看Sync類實現(xiàn)的方法，Sync類的實現(xiàn)其實也不算復雜船响，主要就下面4個方法躬拢，其中：nonfairTryAcquireShared(int acquires)和tryReleaseShared(int releases)先前已經(jīng)將結(jié)果了，下面我們專注：reducePermits(int reductions)和drainPermits()见间。
abstract static class Sync extends AbstractQueuedSynchronizer {
? ? final int nonfairTryAcquireShared(int acquires) {
? ? ? ? //...
? ? }
? ? protected final boolean tryReleaseShared(int releases) {
? ? ? ? //...
? ? }
? ? final void reducePermits(int reductions) {
? ? ? ? //...
? ? }
? ? final int drainPermits() {
? ? ? ? //...
? ? }
}
Sync類實現(xiàn)的的reducePermits(int reductions)的作用是降低許可證數(shù)量聊闯，比如當雙11來臨時，淘寶京東可以對一些服務進行擴容和配置升級米诉，使得原本可以承受10W并發(fā)量的服務提高到可以承受50W馅袁，這里可以在不調(diào)用acquire()的前提下，調(diào)用release()方法增加信號量的許可證荒辕，當雙11的壓力過去后汗销，需要對服務進行縮容犹褒，由50W的并發(fā)量回到10W，這里可以用reducePermits(int reductions)降低許可證數(shù)量弛针。在這個方法中會先獲取當前許可證數(shù)量叠骑，減去我們要扣除的許可證數(shù)量current - reductions，并判斷其結(jié)果是否溢出削茁，如果溢出則拋出異常宙枷，沒有溢出用CAS的方式設置最新的許可證數(shù)量。
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public class Semaphore implements java.io.Serializable {
? ? //...
? ? abstract static class Sync extends AbstractQueuedSynchronizer {
? ? ? ? //...
? ? ? ? final void reducePermits(int reductions) {
? ? ? ? ? ? for (;;) {
? ? ? ? ? ? ? ? int current = getState();
? ? ? ? ? ? ? ? int next = current - reductions;
? ? ? ? ? ? ? ? if (next > current) // underflow
? ? ? ? ? ? ? ? ? ? throw new Error("Permit count underflow");
? ? ? ? ? ? ? ? if (compareAndSetState(current, next))
? ? ? ? ? ? ? ? ? ? return;
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? //...
? ? }
? ? //...
? ? protected void reducePermits(int reduction) {
? ? ? ? if (reduction < 0) throw new IllegalArgumentException();
? ? ? ? sync.reducePermits(reduction);
? ? }
? ? //...
}
需要注意兩點：
這個方法的訪問權限是protected茧跋，如果要使用此方法需要用一個類去繼承慰丛，并修改此方法的訪問權限。
這個方法可能導致信號量的剩余許可證數(shù)量為負瘾杭，比如一個信號量原先的許可證數(shù)量為10诅病，且被借走了9個許可證，當前許可證數(shù)量為1粥烁。這時想把許可證數(shù)量從原先的10扣降到3贤笆，向reducePermits(int reduction)傳入7，此時current-reductions=1-7=-6讨阻，如果CAS成功芥永，那么信號量目前的許可證數(shù)量為-6，不過沒關系钝吮，如果前面借走的9個許可證最終會歸還埋涧，信號量的許可證數(shù)量最終會回到3。
class MySemaphore extends Semaphore {
? ? public MySemaphore(int permits) {
? ? ? ? super(permits);
? ? }
? ? @Override
? ? public void reducePermits(int reduction) {
? ? ? ? super.reducePermits(reduction);
? ? }
}
public static void main(String[] args) {
? ? MySemaphore semaphore = new MySemaphore(8);
? ? System.out.println("初始信號量的許可證數(shù)量:" + semaphore.availablePermits());
? ? //初始化完信號量后奇瘦，增加信號量的許可證數(shù)量
? ? int add = 2;
? ? semaphore.release(add);
? ? System.out.printf("增加%d個許可證后棘催，許可證數(shù)量:%d\n", add, semaphore.availablePermits());
? ? //申請9個許可證
? ? int permits = 9;
? ? try {
? ? ? ? semaphore.acquire(permits);
? ? ? ? System.out.printf("申請%d個許可證后剩余許可證數(shù)量:%d\n", permits, semaphore.availablePermits());
? ? } catch (InterruptedException e) {
? ? ? ? e.printStackTrace();
? ? }
? ? //這里要將原先10個許可證扣除到只剩3個，所以傳入7链患，扣除7個許可證
? ? semaphore.reducePermits(7);
? ? System.out.println("扣除7個許可證數(shù)量后巧鸭，剩余許可證數(shù)量:" + semaphore.availablePermits());
? ? //歸還原先出借的9個許可證
? ? semaphore.release(permits);
? ? System.out.printf("歸還原先出借的%d信號量后，剩余信號量:%d\n", permits, semaphore.availablePermits());
}
執(zhí)行結(jié)果：
1
2
3
4
5
初始信號量的許可證數(shù)量:8
增加2個許可證后麻捻，許可證數(shù)量:10
申請9個許可證后剩余許可證數(shù)量:1
扣除7個許可證數(shù)量后纲仍，剩余許可證數(shù)量:-6
歸還原先出借的9信號量后，剩余信號量:3
Sync類實現(xiàn)的drainPermits()可以一次性扣除信號量目前所有的許可證數(shù)量并返回贸毕，通過這個API郑叠，我們可以得知資源目前最大的訪問限度。還是拿上一章遠程服務為例明棍，判定服務能承受的并發(fā)是5000乡革，用于限流的semaphore信號量的最大許可證數(shù)量也是5000。假設目前信號量剩余的許可證數(shù)量為2000，即有3000個線程正在并發(fā)訪問遠程服務沸版，我們可以通過drainPermits()方法獲取剩余的允許訪問數(shù)量2000嘁傀，然后創(chuàng)建2000個線程訪問遠程服務，這個API一般用于計算量大且計算內(nèi)容比較獨立的場景视粮。
public class Semaphore implements java.io.Serializable {
? ? //...
? ? abstract static class Sync extends AbstractQueuedSynchronizer {
? ? ? ? //...
? ? ? ? final int drainPermits() {
? ? ? ? ? ? for (;;) {
? ? ? ? ? ? ? ? int current = getState();
? ? ? ? ? ? ? ? if (current == 0 || compareAndSetState(current, 0))
? ? ? ? ? ? ? ? ? ? return current;
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? //...
? ? }
? ? //...
? ? public int drainPermits() {
? ? ? ? return sync.drainPermits();
? ? }
? ? //...
}
最后细办，筆者介紹一個Semaphore在JDK1.6.0_17時期的BUG，便結(jié)束對Semaphore的源碼解析蕾殴。
當時AQS的setHeadAndPropagate(Node node, int propagate)和releaseShared(int arg) 兩個方法的實現(xiàn)是下面這樣的笑撞，這個代碼可能導致隊列被阻塞。
private void setHeadAndPropagate(Node node, int propagate) {
? ? setHead(node);
? ? if (propagate > 0 && node.waitStatus != 0) {
? ? ? ? Node s = node.next;
? ? ? ? if (s == null || s.isShared())
? ? ? ? ? ? unparkSuccessor(node);
? ? }
}
public final boolean releaseShared(int arg) {
? ? if (tryReleaseShared(arg)) {
? ? ? ? Node h = head;
? ? ? ? if (h != null && h.waitStatus != 0)
? ? ? ? ? ? unparkSuccessor(h);
? ? ? ? return true;
? ? }
? ? return false;
}
按照上面代碼的實現(xiàn)钓觉，會讓下面的代碼出現(xiàn)隊列被阻塞的情況茴肥。t1和t2線程用于請求許可證，t3和t4線程用于歸還許可證荡灾，循環(huán)10000000次只是為了增加出現(xiàn)阻塞的概率瓤狐，現(xiàn)在說說什么樣的場景下會出現(xiàn)隊列被阻塞的情況。
程序開始時卧晓，信號量的許可證數(shù)量為0芬首，所以t1和t2只能進入隊列等待赴捞，t1和t2在隊列中的節(jié)點對應N1和N2逼裆，節(jié)點的排序為：head->N1->N2(tail)。t3歸還許可證時發(fā)現(xiàn)頭節(jié)點不為null且頭節(jié)點的等待狀態(tài)為SIGNAL赦政，于是會調(diào)用unparkSuccessor(h)方法喚醒頭節(jié)點的后繼節(jié)點N1對應的線程t1胜宇，在執(zhí)行unparkSuccessor(h)的時候會把head的等待狀態(tài)改為0。
t1被喚醒后獲取到許可證恢着，返回剩余許可證數(shù)量為0桐愉，即之后調(diào)用setHeadAndPropagate(Node node, int propagate)方法傳入的propagate為0，但尚未調(diào)用掰派。此時t4也歸還了許可證从诲，但發(fā)現(xiàn)head節(jié)點的等待狀態(tài)為0，就不會調(diào)用unparkSuccessor(h)靡羡。
t1執(zhí)行setHeadAndPropagate(Node node, int propagate)系洛，將頭節(jié)點指向自身線程對應的節(jié)點N1，雖然此時信號量里有剩余的許可證略步，但t1原先拿到的propagate為0描扯，所以不會執(zhí)行unparkSuccessor(node)喚醒t4。
那么新版本的setHeadAndPropagate(Node node, int propagate)和releaseShared(int arg)又是如何保證有許可證被歸還時喚醒隊列中被阻塞的線程呢趟薄？這里其實和PROPAGATE有關绽诚，讓我們按照新版的setHeadAndPropagate和releaseShared走一遍上面的流程。
t1和t2進入隊列中等待，t3歸還許可證發(fā)現(xiàn)頭節(jié)點不為null恩够，且頭節(jié)點等待狀態(tài)為SIGNAL卒落，于是調(diào)用unparkSuccessor(h)方法喚醒頭節(jié)點的后繼節(jié)點N1對應的線程t1，在執(zhí)行unparkSuccessor(h)的時候會把head的等待狀態(tài)改為0蜂桶。
t1被喚醒后獲取到許可證导绷，返回剩余許可證數(shù)量為0，在調(diào)用setHeadAndPropagate(Node node, int propagate)之前屎飘，t4歸還了許可證妥曲，發(fā)現(xiàn)頭節(jié)點的等待狀態(tài)為0，將其改為PROPAGATE钦购。
t1執(zhí)行setHeadAndPropagate(Node node, int propagate)檐盟，獲取原先頭節(jié)點h，并將頭節(jié)點指向N1押桃，此時雖然propagate為0葵萎，但原先頭節(jié)點h的等待狀態(tài)<0，可以執(zhí)行doReleaseShared()喚醒后繼節(jié)點N2對應的線程t2唱凯。
import java.util.concurrent.Semaphore;
public class TestSemaphore {
? ? private static Semaphore sem = new Semaphore(0);
? ? private static class Thread1 extends Thread {
? ? ? ? @Override
? ? ? ? public void run() {
? ? ? ? ? ? sem.acquireUninterruptibly();
? ? ? ? }
? ? }
? ? private static class Thread2 extends Thread {
? ? ? ? @Override
? ? ? ? public void run() {
? ? ? ? ? ? sem.release();
? ? ? ? }
? ? }
? ? public static void main(String[] args) throws InterruptedException {
? ? ? ? for (int i = 0; i < 10000000; i++) {
? ? ? ? ? ? Thread t1 = new Thread1();
? ? ? ? ? ? Thread t2 = new Thread1();
? ? ? ? ? ? Thread t3 = new Thread2();
? ? ? ? ? ? Thread t4 = new Thread2();
? ? ? ? ? ? t1.start();
? ? ? ? ? ? t2.start();
? ? ? ? ? ? t3.start();
? ? ? ? ? ? t4.start();
? ? ? ? ? ? t1.join();
? ? ? ? ? ? t2.join();
? ? ? ? ? ? t3.join();
? ? ? ? ? ? t4.join();
? ? ? ? ? ? System.out.println(i);
? ? ? ? }
? ? }
}
USB Microphone https://www.soft-voice.com/
Wooden Speakers? https://www.zeshuiplatform.com/
亞馬遜測評 www.yisuping.cn
深圳網(wǎng)站建設www.sz886.com
