一行一行源碼分析清楚 AbstractQueuedSynchronizer (三)
轉自：https://javadoop.com/post/AbstractQueuedSynchronizer-3
這篇文章是 AQS 系列的最後一篇，第一篇，我們通過 ReentrantLock 公平鎖分析了 AQS 的核心，第二篇的重點是把 Condition 說明白，同時也說清楚了對于線程中斷的使用。
這篇，我們的關注點是 AQS 最後的部分，共享模式的使用。有前兩篇文章的鋪墊，剩下的源碼分析将會簡單很多。
本文先用 CountDownLatch 将共享模式說清楚，然後順着把其他 AQS 相關的類 CyclicBarrier、Semaphore 的源碼一起過一下。
老規矩：不放過任何一行代碼，沒有任何糊弄，沒有任何瞎說。
- AQS的共享模式
- CountDownLatch
- 使用例子
- 源碼分析
- CyclicBarrier
- Semaphore
- Exchanger
- 總結
在講CountDownLatch之前，先讓我們了解一下什麼是AQS中的共享模式和共享鎖。
深入淺出AQS之共享鎖模式
原文位址：http://www.jianshu.com/p/1161...
搞清楚AQS獨占鎖的實作原理之後，再看共享鎖的實作原理就會輕松很多。兩種鎖模式之間很多通用的地方本文隻會簡單說明一下，就不在贅述了，具體細節可以參考我的上篇文章。
一、執行過程概述
擷取鎖的過程：
- 當線程調用acquireShared()申請擷取鎖資源時，如果成功，則進入臨界區。
- 當擷取鎖失敗時，則建立一個共享類型的節點并進入一個FIFO等待隊列，然後被挂起等待喚醒。
- 當隊列中的等待線程被喚醒以後就重新嘗試擷取鎖資源，如果成功則喚醒後面還在等待的共享節點并把該喚醒事件傳遞下去，即會依次喚醒在該節點後面的所有共享節點，然後進入臨界區，否則繼續挂起等待。
釋放鎖過程：
- 當線程調用releaseShared()進行鎖資源釋放時，如果釋放成功，則喚醒隊列中等待的節點，如果有的話。
二、源碼深入分析
基于上面所說的共享鎖執行流程，我們接下來看下源碼實作邏輯：
首先來看下擷取鎖的方法acquireShared()，如下
public final void acquireShared(int arg) {
//嘗試擷取共享鎖，傳回值小于0表示擷取失敗
if (tryAcquireShared(arg) < 0)
//執行擷取鎖失敗以後的方法
doAcquireShared(arg);
} 這裡tryAcquireShared()方法是留給使用者去實作具體的擷取鎖邏輯的。關于該方法的實作有兩點需要特别說明：
一、該方法必須自己檢查目前上下文是否支援擷取共享鎖，如果支援再進行擷取。
二、該方法傳回值是個重點。其一、由上面的源碼片段可以看出傳回值小于0表示擷取鎖失敗，需要進入等待隊列。其二、如果傳回值等于0表示目前線程擷取共享鎖成功，但它後續的線程是無法繼續擷取的，也就是不需要把它後面等待的節點喚醒。最後、如果傳回值大于0，表示目前線程擷取共享鎖成功且它後續等待的節點也有可能繼續擷取共享鎖成功，也就是說此時需要把後續節點喚醒讓它們去嘗試擷取共享鎖。
有了上面的約定，我們再來看下doAcquireShared方法的實作：
//參數不多說，就是傳給acquireShared()的參數
private void doAcquireShared(int arg) {
//添加等待節點的方法跟獨占鎖一樣，唯一差別就是節點類型變為了共享型，不再贅述
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
//表示前面的節點已經擷取到鎖，自己會嘗試擷取鎖
if (p == head) {
int r = tryAcquireShared(arg);
//注意上面說的， 等于0表示不用喚醒後繼節點，大于0需要
if (r >= 0) {
//這裡是重點，擷取到鎖以後的喚醒操作，後面詳細說
setHeadAndPropagate(node, r);
p.next = null;
//如果是因為中斷醒來則設定中斷标記位
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
//挂起邏輯跟獨占鎖一樣，不再贅述
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
//擷取失敗的取消邏輯跟獨占鎖一樣，不再贅述
if (failed)
cancelAcquire(node);
}
} 獨占鎖模式擷取成功以後設定頭結點然後傳回中斷狀态，結束流程。而共享鎖模式擷取成功以後，調用了setHeadAndPropagate方法，從方法名就可以看出除了設定新的頭結點以外還有一個傳遞動作，一起看下代碼：
//兩個入參，一個是目前成功擷取共享鎖的節點，一個就是tryAcquireShared方法的傳回值，注意上面說的，它可能大于0也可能等于0
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; //記錄目前頭節點
//設定新的頭節點，即把目前擷取到鎖的節點設定為頭節點
//注：這裡是擷取到鎖之後的操作，不需要并發控制
setHead(node);
//這裡意思有兩種情況是需要執行喚醒操作
//1.propagate > 0 表示調用方指明了後繼節點需要被喚醒
//2.頭節點後面的節點需要被喚醒（waitStatus<0），不論是老的頭結點還是新的頭結點
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
//如果目前節點的後繼節點是共享類型擷取沒有後繼節點，則進行喚醒
//這裡可以了解為除非明确指明不需要喚醒（後繼等待節點是獨占類型），否則都要喚醒
if (s == null || s.isShared())
//後面詳細說
doReleaseShared();
}
}
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
} 最終的喚醒操作也很複雜，專門拿出來分析一下：
注：這個喚醒操作在releaseShare()方法裡也會調用。
private void doReleaseShared() {
for (;;) {
//喚醒操作由頭結點開始，注意這裡的頭節點已經是上面新設定的頭結點了
//其實就是喚醒上面新擷取到共享鎖的節點的後繼節點
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
//表示後繼節點需要被喚醒
if (ws == Node.SIGNAL) {
//這裡需要控制并發，因為入口有setHeadAndPropagate跟release兩個，避免兩次unpark
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
//執行喚醒操作
unparkSuccessor(h);
}
//如果後繼節點暫時不需要喚醒，則把目前節點狀态設定為PROPAGATE確定以後可以傳遞下去
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
//如果頭結點沒有發生變化，表示設定完成，退出循環
//如果頭結點發生變化，比如說其他線程擷取到了鎖，為了使自己的喚醒動作可以傳遞，必須進行重試
if (h == head)
break;
}
} 接下來看下釋放共享鎖的過程：
public final boolean releaseShared(int arg) {
//嘗試釋放共享鎖
if (tryReleaseShared(arg)) {
//喚醒過程，詳情見上面分析
doReleaseShared();
return true;
}
return false;
} 注：上面的setHeadAndPropagate()方法表示等待隊列中的線程成功擷取到共享鎖，這時候它需要喚醒它後面的共享節點（如果有），但是當通過releaseShared（）方法去釋放一個共享鎖的時候，接下來等待獨占鎖跟共享鎖的線程都可以被喚醒進行嘗試擷取。
三、總結
跟獨占鎖相比，共享鎖的主要特征在于當一個在等待隊列中的共享節點成功擷取到鎖以後（它擷取到的是共享鎖），既然是共享，那它必須要依次喚醒後面所有可以跟它一起共享目前鎖資源的節點，毫無疑問，這些節點必須也是在等待共享鎖（這是大前提，如果等待的是獨占鎖，那前面已經有一個共享節點擷取鎖了，它肯定是擷取不到的）。當共享鎖被釋放的時候，可以用讀寫鎖為例進行思考，當一個讀鎖被釋放，此時不論是讀鎖還是寫鎖都是可以競争資源的。
CountDownLatch 這個類是比較典型的 AQS 的共享模式的使用，這是一個高頻使用的類。latch 的中文意思是門栓、栅欄，具體怎麼解釋我就不廢話了，大家随意，看兩個例子就知道在哪裡用、怎麼用了。
我們看下 Doug Lea 在 java doc 中給出的例子，這個例子非常實用，我們經常會寫這個代碼。
假設我們有 N ( N > 0 ) 個任務，那麼我們會用 N 來初始化一個 CountDownLatch，然後将這個 latch 的引用傳遞到各個線程中，在每個線程完成了任務後，調用 latch.countDown() 代表完成了一個任務。
調用 latch.await() 的方法的線程會阻塞，直到所有的任務完成。
class Driver2 { // ...
void main() throws InterruptedException {
CountDownLatch doneSignal = new CountDownLatch(N);
Executor e = Executors.newFixedThreadPool(8);
// 建立 N 個任務，送出給線程池來執行
for (int i = 0; i < N; ++i) // create and start threads
e.execute(new WorkerRunnable(doneSignal, i));
// 等待所有的任務完成，這個方法才會傳回
doneSignal.await(); // wait for all to finish
}
}
class WorkerRunnable implements Runnable {
private final CountDownLatch doneSignal;
private final int i;
WorkerRunnable(CountDownLatch doneSignal, int i) {
this.doneSignal = doneSignal;
this.i = i;
}
public void run() {
try {
doWork(i);
// 這個線程的任務完成了，調用 countDown 方法
doneSignal.countDown();
} catch (InterruptedException ex) {
} // return;
}
void doWork() { ...}
}
是以說 CountDownLatch 非常實用，我們常常會将一個比較大的任務進行拆分，然後開啟多個線程來執行，等所有線程都執行完了以後，再往下執行其他操作。這裡例子中，隻有 main 線程調用了 await 方法。
我們再來看另一個例子，這個例子很典型，用了兩個 CountDownLatch：
class Driver { // ...
void main() throws InterruptedException {
CountDownLatch startSignal = new CountDownLatch(1);
CountDownLatch doneSignal = new CountDownLatch(N);
for (int i = 0; i < N; ++i) // create and start threads
new Thread(new Worker(startSignal, doneSignal)).start();
// 這邊插入一些代碼，確定上面的每個線程先啟動起來，才執行下面的代碼。
doSomethingElse(); // don't let run yet
// 因為這裡 N == 1，是以，隻要調用一次，那麼所有的 await 方法都可以通過
startSignal.countDown(); // let all threads proceed
doSomethingElse();
// 等待所有任務結束
doneSignal.await(); // wait for all to finish
}
}
class Worker implements Runnable {
private final CountDownLatch startSignal;
private final CountDownLatch doneSignal;
Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
this.startSignal = startSignal;
this.doneSignal = doneSignal;
}
public void run() {
try {
// 為了讓所有線程同時開始任務，我們讓所有線程先阻塞在這裡
// 等大家都準備好了，再打開這個門栓
startSignal.await();
doWork();
doneSignal.countDown();
} catch (InterruptedException ex) {
} // return;
}
void doWork() { ...}
}
這個例子中，doneSignal 同第一個例子的使用，我們說說這裡的 startSignal。N 個新開啟的線程都調用了startSignal.await() 進行阻塞等待，它們阻塞在栅欄上，隻有當條件滿足的時候（startSignal.countDown()），它們才能同時通過這個栅欄。
轉存失敗重新上傳取消
如果始終隻有一個線程調用 await 方法等待任務完成，那麼 CountDownLatch 就會簡單很多，是以之後的源碼分析讀者一定要在腦海中建構出這麼一個場景：有 m 個線程是做任務的，有 n 個線程在某個栅欄上等待這 m 個線程做完任務，直到所有 m 個任務完成後，n 個線程同時通過栅欄。
Talk is cheap, show me the code.
構造方法，需要傳入一個不小于 0 的整數：
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
// 老套路了，内部封裝一個 Sync 類繼承自 AQS
private static final class Sync extends AbstractQueuedSynchronizer {
Sync(int count) {
// 這樣就 state == count 了
setState(count);
}
...
}
代碼都是套路，先分析套路：AQS 裡面的 state 是一個整數值，這邊用一個 int count 參數其實初始化就是設定了這個值，所有調用了 await 方法的等待線程會挂起，然後有其他一些線程會做 state = state - 1 操作，當 state 減到 0 的同時，那個線程會負責喚醒調用了 await 方法的所有線程。都是套路啊，隻是 Doug Lea 的套路很深，代碼很巧妙，不然我們也沒有要分析源碼的必要。
對于 CountDownLatch，我們僅僅需要關心兩個方法，一個是 countDown() 方法，另一個是 await() 方法。countDown() 方法每次調用都會将 state 減 1，直到 state 的值為 0；而 await 是一個阻塞方法，當 state 減為 0 的時候，await 方法才會傳回。await 可以被多個線程調用，讀者這個時候腦子裡要有個圖：所有調用了 await 方法的線程阻塞在 AQS 的阻塞隊列中，等待條件滿足（state == 0），将線程從隊列中一個個喚醒過來。
我們用以下程式來分析源碼，t1 和 t2 負責調用 countDown() 方法，t3 和 t4 調用 await 方法阻塞：
public class CountDownLatchDemo {
public static void main(String[] args) {
CountDownLatch latch = new CountDownLatch(2);
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(5000);
} catch (InterruptedException ignore) {
}
// 休息 5 秒後(模拟線程工作了 5 秒)，調用 countDown()
latch.countDown();
}
}, "t1");
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(10000);
} catch (InterruptedException ignore) {
}
// 休息 10 秒後(模拟線程工作了 10 秒)，調用 countDown()
latch.countDown();
}
}, "t2");
t1.start();
t2.start();
Thread t3 = new Thread(new Runnable() {
@Override
public void run() {
try {
// 阻塞，等待 state 減為 0
latch.await();
System.out.println("線程 t3 從 await 中傳回了");
} catch (InterruptedException e) {
System.out.println("線程 t3 await 被中斷");
Thread.currentThread().interrupt();
}
}
}, "t3");
Thread t4 = new Thread(new Runnable() {
@Override
public void run() {
try {
// 阻塞，等待 state 減為 0
latch.await();
System.out.println("線程 t4 從 await 中傳回了");
} catch (InterruptedException e) {
System.out.println("線程 t4 await 被中斷");
Thread.currentThread().interrupt();
}
}
}, "t4");
t3.start();
t4.start();
}
}
上述程式，大概在過了 10 秒左右的時候，會輸出：
線程 t3 從 await 中傳回了
線程 t4 從 await 中傳回了
// 這兩條輸出，順序不是絕對的
// 後面的分析，我們假設 t3 先進入阻塞隊列
接下來，我們按照流程一步一步走：先 await 等待，然後被喚醒，await 方法傳回。
首先，我們來看 await() 方法，它代表線程阻塞，等待 state 的值減為 0。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
// 這也是老套路了，我在第二篇的中斷那一節說過了
if (Thread.interrupted())
throw new InterruptedException();
// t3 和 t4 調用 await 的時候，state 都大于 0。
// 也就是說，這個 if 傳回 true，然後往裡看
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
// 隻有當 state == 0 的時候，這個方法才會傳回 1
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
從方法名我們就可以看出，這個方法是擷取共享鎖，并且此方法是可中斷的（中斷的時候抛出 InterruptedException 退出這個方法）。
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 1. 入隊
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
// 同上，隻要 state 不等于 0，那麼這個方法傳回 -1
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 2
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
我們來仔細分析這個方法，線程 t3 經過第 1 步 addWaiter 入隊以後，我們應該可以得到這個：
由于 tryAcquireShared 這個方法會傳回 -1，是以 if (r >= 0) 這個分支不會進去。到 shouldParkAfterFailedAcquire 的時候，t3 将 head 的 waitStatus 值設定為 -1，如下：
然後進入到 parkAndCheckInterrupt 的時候，t3 挂起。
我們再分析 t4 入隊，t4 會将前驅節點 t3 所在節點的 waitStatus 設定為 -1，t4 入隊後，應該是這樣的：
然後，t4 也挂起。接下來，t3 和 t4 就等待喚醒了。
接下來，我們來看喚醒的流程，我們假設用 10 初始化 CountDownLatch。
當然，我們的例子中，其實沒有 10 個線程，隻有 2 個線程 t1 和 t2，隻是為了讓圖好看些罷了。
我們再一步步看具體的流程。首先，我們看 countDown() 方法:
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
// 隻有當 state 減為 0 的時候，tryReleaseShared 才傳回 true
// 否則隻是簡單的 state = state - 1 那麼 countDown 方法就結束了
if (tryReleaseShared(arg)) {
// 喚醒 await 的線程
doReleaseShared();
return true;
}
return false;
}
// 這個方法很簡單，用自旋的方法實作 state 減 1
protected boolean tryReleaseShared(int releases) {
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
countDown 方法就是每次調用都将 state 值減 1，如果 state 減到 0 了，那麼就調用下面的方法進行喚醒阻塞隊列中的線程：
// 調用這個方法的時候，state == 0
// 這個方法先不要看所有的代碼，按照思路往下到我寫注釋的地方，其他的之後還會仔細分析
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
// t3 入隊的時候，已經将頭節點的 waitStatus 設定為 Node.SIGNAL（-1） 了
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
// 就是這裡，喚醒 head 的後繼節點，也就是阻塞隊列中的第一個節點
// 在這裡，也就是喚醒 t3
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) // todo
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
一旦 t3 被喚醒後，我們繼續回到 await 的這段代碼，parkAndCheckInterrupt 傳回，我們先不考慮中斷的情況：
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r); // 2. 這裡是下一步
p.next = null; // help GC
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
// 1. 喚醒後這個方法傳回
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
接下來，t3 會進到 setHeadAndPropagate(node, r) 這個方法，先把 head 給占了，然後喚醒隊列中其他的線程：
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
// 下面說的是，喚醒目前 node 之後的節點，即 t3 已經醒了，馬上喚醒 t4
// 類似的，如果 t4 後面還有 t5，那麼 t4 醒了以後，馬上将 t5 給喚醒了
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
// 又是這個方法，隻是現在的 head 已經不是原來的空節點了，是 t3 的節點了
doReleaseShared();
}
}
又回到這個方法了，那麼接下來，我們好好分析 doReleaseShared 這個方法，我們根據流程，頭節點 head 此時是 t3 節點了：
// 調用這個方法的時候，state == 0
private void doReleaseShared() {
for (;;) {
Node h = head;
// 1. h == null: 說明阻塞隊列為空
// 2. h == tail: 說明頭結點可能是剛剛初始化的頭節點，
// 或者是普通線程節點，但是此節點既然是頭節點了，那麼代表已經被喚醒了，阻塞隊列沒有其他節點了
// 是以這兩種情況不需要進行喚醒後繼節點
if (h != null && h != tail) {
int ws = h.waitStatus;
// t4 将頭節點(此時是 t3)的 waitStatus 設定為 Node.SIGNAL（-1） 了
if (ws == Node.SIGNAL) {
// 這裡 CAS 失敗的場景請看下面的解讀
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
// 就是這裡，喚醒 head 的後繼節點，也就是阻塞隊列中的第一個節點
// 在這裡，也就是喚醒 t4
unparkSuccessor(h);
}
else if (ws == 0 &&
// 這個 CAS 失敗的場景是：執行到這裡的時候，剛好有一個節點入隊，入隊會将這個 ws 設定為 -1
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
// 如果到這裡的時候，前面喚醒的線程已經占領了 head，那麼再循環
// 否則，就是 head 沒變，那麼退出循環，
// 退出循環是不是意味着阻塞隊列中的其他節點就不喚醒了？當然不是，喚醒的線程之後還是會調用這個方法的
if (h == head) // loop if head changed
break;
}
}
我們分析下最後一個 if 語句，然後才能解釋第一個 CAS 為什麼可能會失敗：
- h == head：說明頭節點還沒有被剛剛用 unparkSuccessor 喚醒的線程（這裡可以了解為 t4）占有，此時 break 退出循環。
- h != head：頭節點被剛剛喚醒的線程（這裡可以了解為 t4）占有，那麼這裡重新進入下一輪循環，喚醒下一個節點（這裡是 t4 ）。我們知道，等到 t4 被喚醒後，其實是會主動喚醒 t5、t6、t7...，那為什麼這裡要進行下一個循環來喚醒 t5 呢？我覺得是出于吞吐量的考慮。
滿足上面的 2 的場景，那麼我們就能知道為什麼上面的 CAS 操作 compareAndSetWaitStatus(h, Node.SIGNAL, 0) 會失敗了？
因為目前進行 for 循環的線程到這裡的時候，可能剛剛喚醒的線程 t4 也剛剛好到這裡了，那麼就有可能 CAS 失敗了。
for 循環第一輪的時候會喚醒 t4，t4 醒後會将自己設定為頭節點，如果在 t4 設定頭節點後，for 循環才跑到 if (h == head)，那麼此時會傳回 false，for 循環會進入下一輪。t4 喚醒後也會進入到這個方法裡面，那麼 for 循環第二輪和 t4 就有可能在這個 CAS 相遇，那麼就隻會有一個成功了。
字面意思是“可重複使用的栅欄”，CyclicBarrier 相比 CountDownLatch 來說，要簡單很多，其源碼沒有什麼高深的地方，它是 ReentrantLock 和 Condition 的組合使用。看如下示意圖，CyclicBarrier 和 CountDownLatch 是不是很像，隻是 CyclicBarrier 可以有不止一個栅欄，因為它的栅欄（Barrier）可以重複使用（Cyclic）。
首先，CyclicBarrier 的源碼實作和 CountDownLatch 大相徑庭，CountDownLatch 基于 AQS 的共享模式的使用，而 CyclicBarrier 基于 Condition 來實作。
因為 CyclicBarrier 的源碼相對來說簡單許多，讀者隻要熟悉了前面關于 Condition 的分析，那麼這裡的源碼是毫無壓力的，就是幾個特殊概念罷了。
廢話結束，先上基本屬性和構造方法，往下拉一點點，和圖一起看：
public class CyclicBarrier {
// 我們說了，CyclicBarrier 是可以重複使用的，我們把每次從開始使用到穿過栅欄當做"一代"
private static class Generation {
boolean broken = false;
}
/** The lock for guarding barrier entry */
private final ReentrantLock lock = new ReentrantLock();
// CyclicBarrier 是基于 Condition 的
// Condition 是“條件”的意思，CyclicBarrier 的等待線程通過 barrier 的“條件”是大家都到了栅欄上
private final Condition trip = lock.newCondition();
// 參與的線程數
private final int parties;
// 如果設定了這個，代表越過栅欄之前，要執行相應的操作
private final Runnable barrierCommand;
// 目前所處的“代”
private Generation generation = new Generation();
// 還沒有到栅欄的線程數，這個值初始為 parties，然後遞減
// 還沒有到栅欄的線程數 = parties - 已經到栅欄的數量
private int count;
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
public CyclicBarrier(int parties) {
this(parties, null);
}
我用一圖來描繪下 CyclicBarrier 裡面的一些概念：
看圖我們也知道了，CyclicBarrier 的源碼最重要的就是 await() 方法了。
首先，先看怎麼開啟新的一代：
// 開啟新的一代，當最後一個線程到達栅欄上的時候，調用這個方法來喚醒其他線程，同時初始化“下一代”
private void nextGeneration() {
// 首先，需要喚醒所有的在栅欄上等待的線程
trip.signalAll();
// 更新 count 的值
count = parties;
// 重新生成“新一代”
generation = new Generation();
}
看看怎麼打破一個栅欄：
private void breakBarrier() {
// 設定狀态 broken 為 true
generation.broken = true;
// 重置 count 為初始值 parties
count = parties;
// 喚醒所有已經在等待的線程
trip.signalAll();
}
這兩個方法之後用得到，現在開始分析最重要的等待通過栅欄方法 await 方法：
// 不帶逾時機制
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
// 帶逾時機制，如果逾時抛出 TimeoutException 異常
public int await(long timeout, TimeUnit unit)
throws InterruptedException,
BrokenBarrierException,
TimeoutException {
return dowait(true, unit.toNanos(timeout));
}
繼續往裡看：
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
final ReentrantLock lock = this.lock;
// 先要擷取到鎖，然後在 finally 中要記得釋放鎖
// 如果記得 Condition 部分的話，我們知道 condition 的 await 會釋放鎖，signal 的時候需要重新擷取鎖
lock.lock();
try {
final Generation g = generation;
// 檢查栅欄是否被打破，如果被打破，抛出 BrokenBarrierException 異常
if (g.broken)
throw new BrokenBarrierException();
// 檢查中斷狀态，如果中斷了，抛出 InterruptedException 異常
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
// index 是這個 await 方法的傳回值
// 注意到這裡，這個是從 count 遞減後得到的值
int index = --count;
// 如果等于 0，說明所有的線程都到栅欄上了，準備通過
if (index == 0) { // tripped
boolean ranAction = false;
try {
// 如果在初始化的時候，指定了通過栅欄前需要執行的操作，在這裡會得到執行
final Runnable command = barrierCommand;
if (command != null)
command.run();
// 如果 ranAction 為 true，說明執行 command.run() 的時候，沒有發生異常退出的情況
ranAction = true;
// 喚醒等待的線程，然後開啟新的一代
nextGeneration();
return 0;
} finally {
if (!ranAction)
// 進到這裡，說明執行指定操作的時候，發生了異常，那麼需要打破栅欄
// 之前我們說了，打破栅欄意味着喚醒所有等待的線程，設定 broken 為 true，重置 count 為 parties
breakBarrier();
}
}
// loop until tripped, broken, interrupted, or timed out
// 如果是最後一個線程調用 await，那麼上面就傳回了
// 下面的操作是給那些不是最後一個到達栅欄的線程執行的
for (;;) {
try {
// 如果帶有逾時機制，調用帶逾時的 Condition 的 await 方法等待，直到最後一個線程調用 await
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
// 如果到這裡，說明等待的線程在 await（是 Condition 的 await）的時候被中斷
if (g == generation && ! g.broken) {
// 打破栅欄
breakBarrier();
// 打破栅欄後，重新抛出這個 InterruptedException 異常給外層調用的方法
throw ie;
} else {
// 到這裡，說明 g != generation, 說明新的一代已經産生，即最後一個線程 await 執行完成，
// 那麼此時沒有必要再抛出 InterruptedException 異常，記錄下來這個中斷資訊即可
// 或者是栅欄已經被打破了，那麼也不應該抛出 InterruptedException 異常，
// 而是之後抛出 BrokenBarrierException 異常
Thread.currentThread().interrupt();
}
}
// 喚醒後，檢查栅欄是否是“破的”
if (g.broken)
throw new BrokenBarrierException();
// 這個 for 循環除了異常，就是要從這裡退出了
// 我們要清楚，最後一個線程在執行完指定任務(如果有的話)，會調用 nextGeneration 來開啟一個新的代
// 然後釋放掉鎖，其他線程從 Condition 的 await 方法中得到鎖并傳回，然後到這裡的時候，其實就會滿足 g != generation 的
// 那什麼時候不滿足呢？barrierCommand 執行過程中抛出了異常，那麼會執行打破栅欄操作，
// 設定 broken 為true，然後喚醒這些線程。這些線程會從上面的 if (g.broken) 這個分支抛 BrokenBarrierException 異常傳回
// 當然，還有最後一種可能，那就是 await 逾時，此種情況不會從上面的 if 分支異常傳回，也不會從這裡傳回，會執行後面的代碼
if (g != generation)
return index;
// 如果醒來發現逾時了，打破栅欄，抛出異常
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
好了，我想我應該講清楚了吧，我好像幾乎沒有漏掉任何一行代碼吧？
下面開始收尾工作。
首先，我們看看怎麼得到有多少個線程到了栅欄上，處于等待狀态：
public int getNumberWaiting() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return parties - count;
} finally {
lock.unlock();
}
}
判斷一個栅欄是否被打破了，這個很簡單，直接看 broken 的值即可：
public boolean isBroken() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return generation.broken;
} finally {
lock.unlock();
}
}
前面我們在說 await 的時候也幾乎說清楚了，什麼時候栅欄會被打破，總結如下：
- 中斷，我們說了，如果某個等待的線程發生了中斷，那麼會打破栅欄，同時抛出 InterruptedException 異常；
- 逾時，打破栅欄，同時抛出 TimeoutException 異常；
- 指定執行的操作抛出了異常，這個我們前面也說過。
最後，我們來看看怎麼重置一個栅欄：
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // break the current generation
nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
}
我們設想一下，如果初始化時，指定了線程 parties = 4，前面有 3 個線程調用了 await 等待，在第 4 個線程調用 await 之前，我們調用 reset 方法，那麼會發生什麼？
首先，打破栅欄，那意味着所有等待的線程（3個等待的線程）會喚醒，await 方法會通過抛出 BrokenBarrierException 異常傳回。然後開啟新的一代，重置了 count 和 generation，相當于一切歸零了。
怎麼樣，CyclicBarrier 源碼很簡單吧。
有了 CountDownLatch 的基礎後，分析 Semaphore 會簡單很多。Semaphore 是什麼呢？它類似一個資源池（讀者可以類比線程池），每個線程需要調用 acquire() 方法擷取資源，然後才能執行，執行完後，需要 release 資源，讓給其他的線程用。
大概大家也可以猜到，Semaphore 其實也是 AQS 中共享鎖的使用，因為每個線程共享一個池嘛。
套路解讀：建立 Semaphore 執行個體的時候，需要一個參數 permits，這個基本上可以确定是設定給 AQS 的 state 的，然後每個線程調用 acquire 的時候，執行 state = state - 1，release 的時候執行 state = state + 1，當然，acquire 的時候，如果 state = 0，說明沒有資源了，需要等待其他線程 release。
構造方法：
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
這裡和 ReentrantLock 類似，用了公平政策和非公平政策。
看 acquire 方法：
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public void acquireUninterruptibly() {
sync.acquireShared(1);
}
public void acquire(int permits) throws InterruptedException {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireSharedInterruptibly(permits);
}
public void acquireUninterruptibly(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireShared(permits);
}
這幾個方法也是老套路了，大家基本都懂了吧，這邊多了兩個可以傳參的 acquire 方法，不過大家也都懂的吧，如果我們需要一次擷取超過一個的資源，會用得着這個的。
我們接下來看不抛出 InterruptedException 異常的 acquireUninterruptibly() 方法吧：
public void acquireUninterruptibly() {
sync.acquireShared(1);
}
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
前面說了，Semaphore 分公平政策和非公平政策，我們對比一下兩個 tryAcquireShared 方法：
// 公平政策：
protected int tryAcquireShared(int acquires) {
for (;;) {
// 差別就在于是不是會先判斷是否有線程在排隊，然後才進行 CAS 減操作
if (hasQueuedPredecessors())
return -1;
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
// 非公平政策：
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
也是老套路了，是以從源碼分析角度的話，我們其實不太需要關心是不是公平政策還是非公平政策，它們的差別往往就那麼一兩行。
我們再回到 acquireShared 方法，
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
由于 tryAcquireShared(arg) 傳回小于 0 的時候，說明 state 已經小于 0 了（沒資源了），此時 acquire 不能立馬拿到資源，需要進入到阻塞隊列等待，雖然貼了很多代碼，不在乎多這點了：
private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
這個方法我就不介紹了，線程挂起後等待有資源被 release 出來。接下來，我們就要看 release 的方法了：
// 任務介紹，釋放一個資源
public void release() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
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");
if (compareAndSetState(current, next))
return true;
}
}
tryReleaseShared 方法總是會傳回 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;
}
}
Semphore 的源碼确實很簡單，基本上都是分析過的老代碼的組合使用了。
此篇部落格所有源碼均來自JDK 1.8
前面三篇部落格分别介紹了CyclicBarrier、CountDownLatch、Semaphore，現在介紹并發工具類中的最後一個Exchange。Exchange是最簡單的也是最複雜的，簡單在于API非常簡單，就一個構造方法和兩個exchange()方法，最複雜在于它的實作是最複雜的（反正我是看暈了的）。
在API是這麼介紹的：可以在對中對元素進行配對和交換的線程的同步點。每個線程将條目上的某個方法呈現給 exchange 方法，與夥伴線程進行比對，并且在傳回時接收其夥伴的對象。Exchanger 可能被視為 SynchronousQueue 的雙向形式。Exchanger 可能在應用程式（比如遺傳算法和管道設計）中很有用。
Exchanger，它允許在并發任務之間交換資料。具體來說，Exchanger類允許在兩個線程之間定義同步點。當兩個線程都到達同步點時，他們交換資料結構，是以第一個線程的資料結構進入到第二個線程中，第二個線程的資料結構進入到第一個線程中。
功能簡介：
- Exchanger是一種線程間安全交換資料的機制。可以和之前分析過的SynchronousQueue對比一下：線程A通過SynchronousQueue将資料a交給線程B；線程A通過Exchanger和線程B交換資料，線程A把資料a交給線程B，同時線程B把資料b交給線程A。可見，SynchronousQueue是交給一個資料，Exchanger是交換兩個資料。
應用示例
Exchange實作較為複雜，我們先看其怎麼使用，然後再來分析其源碼。現在我們用Exchange來模拟生産-消費者問題：
public class ExchangerTest {
static class Producer implements Runnable{
//生産者、消費者交換的資料結構
private List<String> buffer;
//步生産者和消費者的交換對象
private Exchanger<List<String>> exchanger;
Producer(List<String> buffer,Exchanger<List<String>> exchanger){
this.buffer = buffer;
this.exchanger = exchanger;
}
@Override
public void run() {
for(int i = 1 ; i < 5 ; i++){
System.out.println("生産者第" + i + "次提供");
for(int j = 1 ; j <= 3 ; j++){
System.out.println("生産者裝入" + i + "--" + j);
buffer.add("buffer：" + i + "--" + j);
}
System.out.println("生産者裝滿，等待與消費者交換...");
try {
exchanger.exchange(buffer);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
static class Consumer implements Runnable {
private List<String> buffer;
private final Exchanger<List<String>> exchanger;
public Consumer(List<String> buffer, Exchanger<List<String>> exchanger) {
this.buffer = buffer;
this.exchanger = exchanger;
}
@Override
public void run() {
for (int i = 1; i < 5; i++) {
//調用exchange()與消費者進行資料交換
try {
buffer = exchanger.exchange(buffer);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("消費者第" + i + "次提取");
for (int j = 1; j <= 3 ; j++) {
System.out.println("消費者 : " + buffer.get(0));
buffer.remove(0);
}
}
}
}
public static void main(String[] args){
List<String> buffer1 = new ArrayList<String>();
List<String> buffer2 = new ArrayList<String>();
Exchanger<List<String>> exchanger = new Exchanger<List<String>>();
Thread producerThread = new Thread(new Producer(buffer1,exchanger));
Thread consumerThread = new Thread(new Consumer(buffer2,exchanger));
producerThread.start();
consumerThread.start();
}
} class ExchangerTest {
static class Producer implements Runnable{
//生産者、消費者交換的資料結構
private List<String> buffer;
//步生産者和消費者的交換對象
private Exchanger<List<String>> exchanger;
Producer(List<String> buffer,Exchanger<List<String>> exchanger){
this.buffer = buffer;
this.exchanger = exchanger;
}
@Override
public void run() {
for(int i = 1 ; i < 5 ; i++){
System.out.println("生産者第" + i + "次提供");
for(int j = 1 ; j <= 3 ; j++){
System.out.println("生産者裝入" + i + "--" + j);
buffer.add("buffer：" + i + "--" + j);
}
System.out.println("生産者裝滿，等待與消費者交換...");
try {
exchanger.exchange(buffer);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
static class Consumer implements Runnable {
private List<String> buffer;
private final Exchanger<List<String>> exchanger;
public Consumer(List<String> buffer, Exchanger<List<String>> exchanger) {
this.buffer = buffer;
this.exchanger = exchanger;
}
@Override
public void run() {
for (int i = 1; i < 5; i++) {
//調用exchange()與消費者進行資料交換
try {
buffer = exchanger.exchange(buffer);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("消費者第" + i + "次提取");
for (int j = 1; j <= 3 ; j++) {
System.out.println("消費者 : " + buffer.get(0));
buffer.remove(0);
}
}
}
}
public static void main(String[] args){
List<String> buffer1 = new ArrayList<String>();
List<String> buffer2 = new ArrayList<String>();
Exchanger<List<String>> exchanger = new Exchanger<List<String>>();
Thread producerThread = new Thread(new Producer(buffer1,exchanger));
Thread consumerThread = new Thread(new Consumer(buffer2,exchanger));
producerThread.start();
consumerThread.start();
}
} 運作結果：
轉存失敗重新上傳取消
首先生産者Producer、消費者Consumer首先都建立一個緩沖清單，通過Exchanger來同步交換資料。消費中通過調用Exchanger與生産者進行同步來擷取資料，而生産者則通過for循環向緩存隊列存儲資料并使用exchanger對象消費者同步。到消費者從exchanger哪裡得到資料後，他的緩沖清單中有3個資料，而生産者得到的則是一個空的清單。上面的例子充分展示了消費者-生産者是如何利用Exchanger來完成資料交換的。
在Exchanger中，如果一個線程已經到達了exchanger節點時，對于它的夥伴節點的情況有三種：
- 如果它的夥伴節點在該線程到達之前已經調用了exchanger方法，則它會喚醒它的夥伴然後進行資料交換，得到各自資料傳回。
- 如果它的夥伴節點還沒有到達交換點，則該線程将會被挂起，等待它的夥伴節點到達被喚醒，完成資料交換。
- 如果目前線程被中斷了則抛出異常，或者等待逾時了，則抛出逾時異常。
核心交換算法實作
換句話說Exchanger提供的是一個交換服務，允許原子性的交換兩個（多個）對象，但同時隻有一對才會成功。先看一個簡單的執行個體模型。
在上面的模型中，我們假定一個空的棧（Stack），棧頂（Top）當然是沒有元素的。同時我們假定一個資料結構Node，包含一個要交換的元素E和一個要填充的“洞”Node。這時線程T1攜帶節點node1進入棧（cas_push)，當然這是CAS操作，這樣棧頂就不為空了。線程T2攜帶節點node2進入棧，發現棧裡面已經有元素了node1，同時發現node1的hold（Node）為空，于是将自己（node2）填充到node1的hold中（cas_fill）。然後将元素node1從棧中彈出（cas_take）。這樣線程T1就得到了node1.hold.item也就是node2的元素e2，線程T2就得到了node1.item也就是e1，進而達到了交換的目的。
算法描述就是下圖展示的内容。
JDK 5就是采用類似的思想實作的Exchanger。JDK 6以後為了支援多線程多對象同時Exchanger了就進行了改造（為了支援更好的并發），采用ConcurrentHashMap的思想，将Stack分割成很多的片段（或者說插槽Slot），線程Id（Thread.getId()）hash相同的落在同一個Slot上，這樣在預設32個Slot上就有很好的吞吐量。當然會根據機器CPU核心的數量有一定的優化，有興趣的可以去了解下Exchanger的源碼。
實作分析
Exchanger算法的核心是通過一個可交換資料的slot，以及一個可以帶有資料item的參與者。源碼中的描述如下：
for (;;) {
if (slot is empty) { // offer
place item in a Node;
if (can CAS slot from empty to node) {
wait for release;
return matching item in node;
}
}
else if (can CAS slot from node to empty) { // release
get the item in node;
set matching item in node;
release waiting thread;
}
// else retry on CAS failure
} for (;;) {
if (slot is empty) { // offer
place item in a Node;
if (can CAS slot from empty to node) {
wait for release;
return matching item in node;
}
}
else if (can CAS slot from node to empty) { // release
get the item in node;
set matching item in node;
release waiting thread;
}
// else retry on CAS failure
} Exchanger中定義了如下幾個重要的成員變量：
private final Participant participant;
private volatile Node[] arena;
private volatile Node slot; final Participant participant;
private volatile Node[] arena;
private volatile Node slot; participant的作用是為每個線程保留唯一的一個Node節點。
slot為單個槽，arena為數組槽。他們都是Node類型。在這裡可能會感覺到疑惑，slot作為Exchanger交換資料的場景，應該隻需要一個就可以了啊？為何還多了一個Participant 和數組類型的arena呢？一個slot交換場所原則上來說應該是可以的，但實際情況卻不是如此，多個參與者使用同一個交換場所時，會存在嚴重伸縮性問題。既然單個交換場所存在問題，那麼我們就安排多個，也就是數組arena。通過數組arena來安排不同的線程使用不同的slot來降低競争問題，并且可以保證最終一定會成對交換資料。但是Exchanger不是一來就會生成arena數組來降低競争，隻有當産生競争是才會生成arena數組。那麼怎麼将Node與目前線程綁定呢？Participant ，Participant 的作用就是為每個線程保留唯一的一個Node節點，它繼承ThreadLocal，同時在Node節點中記錄在arena中的下标index。
Node定義如下：
@sun.misc.Contended static final class Node {
int index; // Arena index
int bound; // Last recorded value of Exchanger.bound
int collides; // Number of CAS failures at current bound
int hash; // Pseudo-random for spins
Object item; // This thread's current item
volatile Object match; // Item provided by releasing thread
volatile Thread parked; // Set to this thread when parked, else null
} @sun.misc.Contended static final class Node {
int index; // Arena index
int bound; // Last recorded value of Exchanger.bound
int collides; // Number of CAS failures at current bound
int hash; // Pseudo-random for spins
Object item; // This thread's current item
volatile Object match; // Item provided by releasing thread
volatile Thread parked; // Set to this thread when parked, else null
} - index：arena的下标；
- bound：上一次記錄的Exchanger.bound；
- collides：在目前bound下CAS失敗的次數；
- hash：僞随機數，用于自旋；
- item：這個線程的目前項，也就是需要交換的資料；
- match：做releasing操作的線程傳遞的項；
- parked：挂起時設定線程值，其他情況下為null；
在Node定義中有兩個變量值得思考：bound以及collides。前面提到了數組area是為了避免競争而産生的，如果系統不存在競争問題，那麼完全沒有必要開辟一個高效的arena來徒增系統的複雜性。首先通過單個slot的exchanger來交換資料，當探測到競争時将安排不同的位置的slot來儲存線程Node，并且可以確定沒有slot會在同一個緩存行上。如何來判斷會有競争呢？CAS替換slot失敗，如果失敗，則通過記錄沖突次數來擴充arena的尺寸，我們在記錄沖突的過程中會跟蹤“bound”的值，以及會重新計算沖突次數在bound的值被改變時。這裡闡述可能有點兒模糊，不着急，我們先有這個概念，後面在arenaExchange中再次做詳細闡述。
我們直接看exchange()方法
exchange(V x)
exchange(V x)：等待另一個線程到達此交換點（除非目前線程被中斷），然後将給定的對象傳送給該線程，并接收該線程的對象。方法定義如下：
public V exchange(V x) throws InterruptedException {
Object v;
Object item = (x == null) ? NULL_ITEM : x; // translate null args
if ((arena != null ||
(v = slotExchange(item, false, 0L)) == null) &&
((Thread.interrupted() || // disambiguates null return
(v = arenaExchange(item, false, 0L)) == null)))
throw new InterruptedException();
return (v == NULL_ITEM) ? null : (V)v;
} public V exchange(V x) throws InterruptedException {
Object v;
Object item = (x == null) ? NULL_ITEM : x; // translate null args
if ((arena != null ||
(v = slotExchange(item, false, 0L)) == null) &&
((Thread.interrupted() || // disambiguates null return
(v = arenaExchange(item, false, 0L)) == null)))
throw new InterruptedException();
return (v == NULL_ITEM) ? null : (V)v;
} 這個方法比較好了解：arena為數組槽，如果為null，則執行slotExchange()方法，否則判斷線程是否中斷，如果中斷值抛出InterruptedException異常，沒有中斷則執行arenaExchange()方法。整套邏輯就是：如果slotExchange(Object item, boolean timed, long ns)方法執行失敗了就執行arenaExchange(Object item, boolean timed, long ns)方法，最後傳回結果V。
NULL_ITEM 為一個空節點，其實就是一個Object對象而已，slotExchange()為單個slot交換。
slotExchange(Object item, boolean timed, long ns)
private final Object slotExchange(Object item, boolean timed, long ns) {
// 擷取目前線程的節點 p
Node p = participant.get();
// 目前線程
Thread t = Thread.currentThread();
// 線程中斷，直接傳回
if (t.isInterrupted())
return null;
// 自旋
for (Node q;;) {
//slot != null
if ((q = slot) != null) {
//嘗試CAS替換
if (U.compareAndSwapObject(this, SLOT, q, null)) {
Object v = q.item; // 目前線程的項，也就是交換的資料
q.match = item; // 做releasing操作的線程傳遞的項
Thread w = q.parked; // 挂起時設定線程值
// 挂起線程不為null，線程挂起
if (w != null)
U.unpark(w);
return v;
}
//如果失敗了，則建立arena
//bound 則是上次Exchanger.bound
if (NCPU > 1 && bound == 0 &&
U.compareAndSwapInt(this, BOUND, 0, SEQ))
arena = new Node[(FULL + 2) << ASHIFT];
}
//如果arena != null，直接傳回，進入arenaExchange邏輯處理
else if (arena != null)
return null;
else {
p.item = item;
if (U.compareAndSwapObject(this, SLOT, null, p))
break;
p.item = null;
}
}
/*
* 等待 release
* 進入spin+block模式
*/
int h = p.hash;
long end = timed ? System.nanoTime() + ns : 0L;
int spins = (NCPU > 1) ? SPINS : 1;
Object v;
while ((v = p.match) == null) {
if (spins > 0) {
h ^= h << 1; h ^= h >>> 3; h ^= h << 10;
if (h == 0)
h = SPINS | (int)t.getId();
else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0)
Thread.yield();
}
else if (slot != p)
spins = SPINS;
else if (!t.isInterrupted() && arena == null &&
(!timed || (ns = end - System.nanoTime()) > 0L)) {
U.putObject(t, BLOCKER, this);
p.parked = t;
if (slot == p)
U.park(false, ns);
p.parked = null;
U.putObject(t, BLOCKER, null);
}
else if (U.compareAndSwapObject(this, SLOT, p, null)) {
v = timed && ns <= 0L && !t.isInterrupted() ? TIMED_OUT : null;
break;
}
}
U.putOrderedObject(p, MATCH, null);
p.item = null;
p.hash = h;
return v;
} final Object slotExchange(Object item, boolean timed, long ns) {
// 擷取目前線程的節點 p
Node p = participant.get();
// 目前線程
Thread t = Thread.currentThread();
// 線程中斷，直接傳回
if (t.isInterrupted())
return null;
// 自旋
for (Node q;;) {
//slot != null
if ((q = slot) != null) {
//嘗試CAS替換
if (U.compareAndSwapObject(this, SLOT, q, null)) {
Object v = q.item; // 目前線程的項，也就是交換的資料
q.match = item; // 做releasing操作的線程傳遞的項
Thread w = q.parked; // 挂起時設定線程值
// 挂起線程不為null，線程挂起
if (w != null)
U.unpark(w);
return v;
}
//如果失敗了，則建立arena
//bound 則是上次Exchanger.bound
if (NCPU > 1 && bound == 0 &&
U.compareAndSwapInt(this, BOUND, 0, SEQ))
arena = new Node[(FULL + 2) << ASHIFT];
}
//如果arena != null，直接傳回，進入arenaExchange邏輯處理
else if (arena != null)
return null;
else {
p.item = item;
if (U.compareAndSwapObject(this, SLOT, null, p))
break;
p.item = null;
}
}
/*
* 等待 release
* 進入spin+block模式
*/
int h = p.hash;
long end = timed ? System.nanoTime() + ns : 0L;
int spins = (NCPU > 1) ? SPINS : 1;
Object v;
while ((v = p.match) == null) {
if (spins > 0) {
h ^= h << 1; h ^= h >>> 3; h ^= h << 10;
if (h == 0)
h = SPINS | (int)t.getId();
else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0)
Thread.yield();
}
else if (slot != p)
spins = SPINS;
else if (!t.isInterrupted() && arena == null &&
(!timed || (ns = end - System.nanoTime()) > 0L)) {
U.putObject(t, BLOCKER, this);
p.parked = t;
if (slot == p)
U.park(false, ns);
p.parked = null;
U.putObject(t, BLOCKER, null);
}
else if (U.compareAndSwapObject(this, SLOT, p, null)) {
v = timed && ns <= 0L && !t.isInterrupted() ? TIMED_OUT : null;
break;
}
}
U.putOrderedObject(p, MATCH, null);
p.item = null;
p.hash = h;
return v;
} 程式首先通過participant擷取目前線程節點Node。檢測是否中斷，如果中斷return null，等待後續抛出InterruptedException異常。
如果slot不為null，則進行slot消除，成功直接傳回資料V，否則失敗，則建立arena消除數組。
如果slot為null，但arena不為null，則傳回null，進入arenaExchange邏輯。
如果slot為null，且arena也為null，則嘗試占領該slot，失敗重試，成功則跳出循環進入spin+block（自旋+阻塞）模式。
在自旋+阻塞模式中，首先取得結束時間和自旋次數。如果match(做releasing操作的線程傳遞的項)為null，其首先嘗試spins+随機次自旋（改自旋使用目前節點中的hash，并改變之）和退讓。當自旋數為0後，假如slot發生了改變（slot != p）則重置自旋數并重試。否則假如：目前未中斷&arena為null&（目前不是限時版本或者限時版本+目前時間未結束）：阻塞或者限時阻塞。假如：目前中斷或者arena不為null或者目前為限時版本+時間已經結束：不限時版本：置v為null；限時版本：如果時間結束以及未中斷則TIMED_OUT；否則給出null（原因是探測到arena非空或者目前線程中斷）。
match不為空時跳出循環。
整個slotExchange清晰明了。
arenaExchange(Object item, boolean timed, long ns)
private final Object arenaExchange(Object item, boolean timed, long ns) {
Node[] a = arena;
Node p = participant.get();
for (int i = p.index;;) { // access slot at i
int b, m, c; long j; // j is raw array offset
Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE);
if (q != null && U.compareAndSwapObject(a, j, q, null)) {
Object v = q.item; // release
q.match = item;
Thread w = q.parked;
if (w != null)
U.unpark(w);
return v;
}
else if (i <= (m = (b = bound) & MMASK) && q == null) {
p.item = item; // offer
if (U.compareAndSwapObject(a, j, null, p)) {
long end = (timed && m == 0) ? System.nanoTime() + ns : 0L;
Thread t = Thread.currentThread(); // wait
for (int h = p.hash, spins = SPINS;;) {
Object v = p.match;
if (v != null) {
U.putOrderedObject(p, MATCH, null);
p.item = null; // clear for next use
p.hash = h;
return v;
}
else if (spins > 0) {
h ^= h << 1; h ^= h >>> 3; h ^= h << 10; // xorshift
if (h == 0) // initialize hash
h = SPINS | (int)t.getId();
else if (h < 0 && // approx 50% true
(--spins & ((SPINS >>> 1) - 1)) == 0)
Thread.yield(); // two yields per wait
}
else if (U.getObjectVolatile(a, j) != p)
spins = SPINS; // releaser hasn't set match yet
else if (!t.isInterrupted() && m == 0 &&
(!timed ||
(ns = end - System.nanoTime()) > 0L)) {
U.putObject(t, BLOCKER, this); // emulate LockSupport
p.parked = t; // minimize window
if (U.getObjectVolatile(a, j) == p)
U.park(false, ns);
p.parked = null;
U.putObject(t, BLOCKER, null);
}
else if (U.getObjectVolatile(a, j) == p &&
U.compareAndSwapObject(a, j, p, null)) {
if (m != 0) // try to shrink
U.compareAndSwapInt(this, BOUND, b, b + SEQ - 1);
p.item = null;
p.hash = h;
i = p.index >>>= 1; // descend
if (Thread.interrupted())
return null;
if (timed && m == 0 && ns <= 0L)
return TIMED_OUT;
break; // expired; restart
}
}
}
else
p.item = null; // clear offer
}
else {
if (p.bound != b) { // stale; reset
p.bound = b;
p.collides = 0;
i = (i != m || m == 0) ? m : m - 1;
}
else if ((c = p.collides) < m || m == FULL ||
!U.compareAndSwapInt(this, BOUND, b, b + SEQ + 1)) {
p.collides = c + 1;
i = (i == 0) ? m : i - 1; // cyclically traverse
}
else
i = m + 1; // grow
p.index = i;
}
}
} private final Object arenaExchange(Object item, boolean timed, long ns) {
Node[] a = arena;
Node p = participant.get();
for (int i = p.index;;) { // access slot at i
int b, m, c; long j; // j is raw array offset
Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE);
if (q != null && U.compareAndSwapObject(a, j, q, null)) {
Object v = q.item; // release
q.match = item;
Thread w = q.parked;
if (w != null)
U.unpark(w);
return v;
}
else if (i <= (m = (b = bound) & MMASK) && q == null) {
p.item = item; // offer
if (U.compareAndSwapObject(a, j, null, p)) {
long end = (timed && m == 0) ? System.nanoTime() + ns : 0L;
Thread t = Thread.currentThread(); // wait
for (int h = p.hash, spins = SPINS;;) {
Object v = p.match;
if (v != null) {
U.putOrderedObject(p, MATCH, null);
p.item = null; // clear for next use
p.hash = h;
return v;
}
else if (spins > 0) {
h ^= h << 1; h ^= h >>> 3; h ^= h << 10; // xorshift
if (h == 0) // initialize hash
h = SPINS | (int)t.getId();
else if (h < 0 && // approx 50% true
(--spins & ((SPINS >>> 1) - 1)) == 0)
Thread.yield(); // two yields per wait
}
else if (U.getObjectVolatile(a, j) != p)
spins = SPINS; // releaser hasn't set match yet
else if (!t.isInterrupted() && m == 0 &&
(!timed ||
(ns = end - System.nanoTime()) > 0L)) {
U.putObject(t, BLOCKER, this); // emulate LockSupport
p.parked = t; // minimize window
if (U.getObjectVolatile(a, j) == p)
U.park(false, ns);
p.parked = null;
U.putObject(t, BLOCKER, null);
}
else if (U.getObjectVolatile(a, j) == p &&
U.compareAndSwapObject(a, j, p, null)) {
if (m != 0) // try to shrink
U.compareAndSwapInt(this, BOUND, b, b + SEQ - 1);
p.item = null;
p.hash = h;
i = p.index >>>= 1; // descend
if (Thread.interrupted())
return null;
if (timed && m == 0 && ns <= 0L)
return TIMED_OUT;
break; // expired; restart
}
}
}
else
p.item = null; // clear offer
}
else {
if (p.bound != b) { // stale; reset
p.bound = b;
p.collides = 0;
i = (i != m || m == 0) ? m : m - 1;
}
else if ((c = p.collides) < m || m == FULL ||
!U.compareAndSwapInt(this, BOUND, b, b + SEQ + 1)) {
p.collides = c + 1;
i = (i == 0) ? m : i - 1; // cyclically traverse
}
else
i = m + 1; // grow
p.index = i;
}
}
} 首先通過participant取得目前節點Node，然後根據目前節點Node的index去取arena中相對應的節點node。前面提到過arena可以確定不同的slot在arena中是不會相沖突的，那麼是怎麼保證的呢？我們先看arena的建立：
arena = new Node[(FULL + 2) << ASHIFT]; = new Node[(FULL + 2) << ASHIFT]; 這個arena到底有多大呢？我們先看FULL 和ASHIFT的定義：
static final int FULL = (NCPU >= (MMASK << 1)) ? MMASK : NCPU >>> 1;
private static final int ASHIFT = 7;
private static final int NCPU = Runtime.getRuntime().availableProcessors();
private static final int MMASK = 0xff; // 255 final int FULL = (NCPU >= (MMASK << 1)) ? MMASK : NCPU >>> 1;
private static final int ASHIFT = 7;
private static final int NCPU = Runtime.getRuntime().availableProcessors();
private static final int MMASK = 0xff; // 255 假如我的機器NCPU = 8 ，則得到的是768大小的arena數組。然後通過以下代碼取得在arena中的節點：
Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE); Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE); 他仍然是通過右移ASHIFT位來取得Node的，ABASE定義如下：
Class<?> ak = Node[].class;
ABASE = U.arrayBaseOffset(ak) + (1 << ASHIFT); <?> ak = Node[].class;
ABASE = U.arrayBaseOffset(ak) + (1 << ASHIFT); U.arrayBaseOffset擷取對象頭長度，數組元素的大小可以通過unsafe.arrayIndexScale(T[].class) 方法擷取到。這也就是說要通路類型為T的第N個元素的話，你的偏移量offset應該是arrayOffset+N*arrayScale。也就是說BASE = arrayOffset+ 128 。其次我們再看Node節點的定義
@sun.misc.Contended static final class Node{
....
} @sun.misc.Contended static final class Node{
....
} 在Java 8 中我們是可以利用sun.misc.Contended來規避僞共享的。是以說通過 << ASHIFT方式加上sun.misc.Contended，是以使得任意兩個可用Node不會再同一個緩存行中。
關于僞共享請參考如下博文：
僞共享(False Sharing)
[ Java8中用sun.misc.Contended避免僞共享(false sharing)](
http://blog.csdn.net/aigoogle/article/details/41518369)
我們再次回到arenaExchange()。取得arena中的node節點後，如果定位的節點q 不為空，且CAS操作成功，則交換資料，傳回交換的資料，喚醒等待的線程。
如果q等于null且下标在bound & MMASK範圍之内，則嘗試占領該位置，如果成功，則采用自旋 + 阻塞的方式進行等待交換資料。
如果下标不在bound & MMASK範圍之内擷取由于q不為null但是競争失敗的時候：消除p。加入bound 不等于目前節點的bond（b != p.bound），則更新p.bound = b，collides = 0 ，i = m或者m - 1。如果沖突的次數不到m 擷取m 已經為最大值或者修改目前bound的值失敗，則通過增加一次collides以及循環遞減下标i的值；否則更新目前bound的值成功：我們令i為m+1即為此時最大的下标。最後更新目前index的值。
Exchanger使用、原理都比較好了解，但是這個源碼看起來真心有點兒複雜，是真心難看懂，但是這種交換的思路Doug Lea在後續博文中還會提到，例如SynchronousQueue、LinkedTransferQueue。
最後用一個在網上看到的段子結束此篇部落格（http://brokendreams.iteye.com/blog/2253956），部落客對其做了一點點修改，以便更加符合在1.8環境下的Exchanger：
其實就是"我"和"你"(可能有多個"我"，多個"你")在一個叫Slot的地方做交易(一手交錢，一手交貨)，過程分以下步驟：
- 我先到一個叫做Slot的交易場所交易，發現你已經到了，那我就嘗試喊你交易，如果你回應了我，決定和我交易那麼進入第2步；如果别人搶先一步把你喊走了，那我就進入第5步。
- 我拿出錢交給你，你可能會接收我的錢，然後把貨給我，交易結束；也可能嫌我掏錢太慢(逾時)或者接個電話(中斷)，TM的不賣了，走了，那我隻能再找别人買貨了(從頭開始)。
- 我到交易地點的時候，你不在，那我先嘗試把這個交易點給占了(一屁股做凳子上…)，如果我成功搶占了單間(交易點)，那就坐這兒等着你拿貨來交易，進入第4步；如果被别人搶座了，那我隻能在找别的地方兒了，進入第5步。
- 你拿着貨來了，喊我交易，然後完成交易；也可能我等了好長時間你都沒來，我不等了，繼續找别人交易去，走的時候我看了一眼，一共沒多少人，弄了這麼多單間(交易地點Slot)，太TM浪費了，我喊來交易地點管理者：一共也沒幾個人，搞這麼多單間兒幹毛，給哥撤一個！。然後再找别人買貨(從頭開始)；或者我老大給我打了個電話，不讓我買貨了(中斷)。
-
我跑去喊管理者，尼瑪，就一個坑交易個毛啊，然後管理在一個更加開闊的地方開辟了好多個單間，然後我就挨個來看每個單間是否有人。如果有人我就問他是否可以交易，如果回應了我，那我就進入第2步。如果我沒有人，那我就占着這個單間等其他人來交易，進入第4步。
6.如果我嘗試了幾次都沒有成功，我就會認為，是不是我TM選的這個單間風水不好？不行，得換個地兒繼續(從頭開始)；如果我嘗試了多次發現還沒有成功，怒了，把管理者喊來：給哥再開一個單間(Slot)，加一個凳子，這麼多人就這麼幾個破凳子夠誰用！
CountdownLatch基于AQS的共享鎖實作，線程運作完内部調用cas操作修改state - 1，當state為0時阻塞的線程被一起喚醒。
collidebarrier基于lock和condition實作，線程運作到屏障時cas操作修改state - 1，并阻塞在condition上，state= 0 時conditon.signal阻塞的線程。
Semaphore基于AQS共享實作。同上。
exchanger内部保持一個棧（後來改為slot或slot數組），一個線程進入時cas把資料放到棧中，并阻塞，另一個線程進來發現有東西，cas擷取并清空，同時把資料放到夥伴線程的一個結構中，喚醒夥伴線程，運作完畢。
寫到這裡，終于把 AbstractQueuedSynchronizer 基本上說完了，對于 Java 并發，Doug Lea 真的是神一樣的存在。日後我們還會接觸到很多 Doug Lea 的代碼，希望我們大家都可以朝着大神的方向不斷打磨自己的技術，少一些高大上的架構，多一些實實在在的優秀代碼吧。
更多内容請關注微信公衆号【Java技術江湖】
一位阿裡 Java 工程師的技術小站。作者黃小斜，專注 Java 相關技術：SSM、SpringBoot、MySQL、分布式、中間件、叢集、Linux、網絡、多線程，偶爾講點Docker、ELK，同時也分享技術幹貨和學習經驗，緻力于Java全棧開發！（關注公衆号後回複”Java“即可領取 Java基礎、進階、項目和架構師等免費學習資料，更有資料庫、分布式、微服務等熱門技術學習視訊，内容豐富，兼顧原理和實踐，另外也将贈送作者原創的Java學習指南、Java程式員面試指南等幹貨資源）