前言
如果我們希望所有的線程都到達同一個地方才能繼續(xù)往下執(zhí)行，那么CyclicBarrier就是一個不錯的選擇。
開始
在下面一個例子中，我希望10個線程都要到達同一個地方才可以往下走。
public static void main(String[] args) {
int num = 10;
CyclicBarrier cb = new CyclicBarrier(num);
for (int i = 0; i < num; i++) {
Thread t = new Thread(new Task(cb, i + 1));
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
t.start();
}
}
static class Task implements Runnable {
private final CyclicBarrier cb;
private final int id;
Task(CyclicBarrier cb, int id) {
this.cb = cb;
this.id = id;
}
@Override
public void run() {
try {
System.out.println("Task#" + id + " is waiting...");
cb.await();
System.out.println("Task#" + id + " is finished...");
} catch (Exception e) {
e.printStackTrace();
}
}
}
結(jié)果如下所示：
Task#1 is waiting...
Task#2 is waiting...
Task#3 is waiting...
Task#4 is waiting...
Task#5 is waiting...
Task#6 is waiting...
Task#7 is waiting...
Task#8 is waiting...
Task#9 is waiting...
Task#10 is waiting...
Task#10 is finished...
Task#1 is finished...
Task#3 is finished...
Task#4 is finished...
Task#5 is finished...
Task#6 is finished...
Task#2 is finished...
Task#8 is finished...
Task#9 is finished...
Task#7 is finished...
可以看見，線程一個接一個啟動了，一旦第十個線程到達了前九個線程到達的地方，那么每個線程都啟動了，啟動后的輸出就沒有什么順序了。這是CycicBarrier一個比較簡單的例子。再利用jstack可以看到線程等待的狀態(tài)：
"Thread-0" #12 prio=5 os_prio=0 tid=0x00007f7438136800 nid=0x3dc4 waiting on condition [0x00007f7411de2000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x00000000d6d7ce38> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:234)
at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:362)
at com.jdk.CyclicBarrierDemo$Task.run(CyclicBarrierDemo.java:39)
at java.lang.Thread.run(Thread.java:745)
這就很清楚了，從調(diào)用深度高到低排序，LockSupport的park方法，AQS里的ConditionObject的await方法，CyclicBarrier的dowait方法，找它們錯不了。
CyclicBarrier的dowait方法
dowait方法這個描述很形象，一下子就告訴了我們，我就是讓線程抱團停下的方法，有什么事找我吧。
// 這是CyclicBarrier提供給我們調(diào)用的API
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
/**
* Main barrier code, covering the various policies.
*/
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
//畢竟是臨界區(qū)，安全起見，還是要上鎖了，
//每個new出來的CyclicBarrier對象就有一把唯一（final）的重入鎖
final ReentrantLock lock = this.lock;
lock.lock();
try {
//CyclicBarrier是可以重用的，它的重用機制就是通過設(shè)置一個Generation實現(xiàn)
//當(dāng)我們希望重用這個CyclicBarrier對象，reset()調(diào)用以后
//generation屬性就被重新創(chuàng)建。
final Generation g = generation;
//這段代碼是想說，所有的線程還沒到達同一個地方，
//CyclicBarrier就被“無效”了(broken)，那是不正常的，要拋異常
if (g.broken)
throw new BrokenBarrierException();
//線程被設(shè)置了中斷標(biāo)簽，這是符合突破這個CyclicBarrier的條件了，
//所以，調(diào)用breakBarrier()沒毛病，但是還是要拋出異常
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
//每次拿到鎖的線程走到這里都要對計數(shù)count--
//雖然count--有三個步驟要走，但是畢竟上鎖了，who cares
int index = --count;
if (index == 0) { // tripped
//走到這里，CyclicBarrier就真的被翻越了
boolean ranAction = false;
try {
//按照文檔的說法，我們可以設(shè)置一個Runnable
//在CyclicBarrier被翻越的時候執(zhí)行
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
//設(shè)置新的Generation已提供再用。
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
// loop until tripped, broken, interrupted, or timed out
for (;;) {
try {
//沒有超時，調(diào)用ConditionObject的await或者awaitNanos方法，當(dāng)前線程被阻塞
//至于拿到的鎖，會在await或者awaitNanos方法里面釋放
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
// We're about to finish waiting even if we had not
// been interrupted, so this interrupt is deemed to
// "belong" to subsequent execution.
Thread.currentThread().interrupt();
}
}
//以下代碼和上面說的一樣，總結(jié)一下就是
//如果Thread的中斷狀態(tài)位被設(shè)置了就拋出InterruptedException
//或者在還有線程等待的時候CyclicBarrier被翻越BrokenBarrierException
if (g.broken)
throw new BrokenBarrierException();
if (g != generation)
return index;
//等待超時，也要拋異常，這次是TimeoutException
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
//解鎖好習(xí)慣
lock.unlock();
}
}
AQS里的ConditionObject的await方法
//這段代碼還是在CyclicBarrier里面
//線程等待的入口就是通過這個實現(xiàn)了Condition接口的trip來完成的
//是通過ReentrantLock獲取的，這也是它能夠釋放鎖的伏筆了。
private final Condition trip = lock.newCondition();
//以下這個方法才是主角
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
//先加入一個waitStatus=CONDITION的Node
//這個Node就是AQS里面使用的CLH隊列的Node
//加入的是條件（condition）隊列，而不是同步隊列（sync）
Node node = addConditionWaiter();
//釋放鎖的地方就在這里了，要知道，ReebtrantLock繼承了AQS，
//這實際上是在一個類里面操作，還要返回這個線程上了幾次鎖，
//也就是AQS里state屬性的值，保存在savedState，
//給突破CyclicBarrier以后，爭取鎖使用
int savedState = fullyRelease(node);
int interruptMode = 0;
//只要不在同步隊列中，當(dāng)前線程就要調(diào)用LockSupport的park方法
//調(diào)用park方法的線程會阻塞
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
//等于0表示沒有cancel，還繼續(xù)要等待
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
//嘗試把node加入阻塞隊列，加入阻塞隊列就是獲取公平鎖的機制了
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
//Condition隊列用的是nextWaiter指針，屬于單向鏈表
//在這里要清理已經(jīng)取消等待的線程
//unlinkCancelledWaiters使用的思路就是我們在單向鏈表刪除節(jié)點的思路，
//從頭遍歷，改變nextWaiter指針。
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
//不等于0，確實是需要中斷，給線程設(shè)置中斷標(biāo)志位，或者拋出異常
//這可以在reportInterruptAfterWait方法立細(xì)究
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
喚醒線程
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}
所以，現(xiàn)在就主要看ConditionObject的signalAll的方法了。
ConditionObject的signalAll方法
public final void signalAll() {
//如果不是異己鎖，拋出異常
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignalAll(first);
}
private void doSignalAll(Node first) {
lastWaiter = firstWaiter = null;
do {
//將節(jié)點脫離條件隊列，就是將nextWaiter指針置空
Node next = first.nextWaiter;
first.nextWaiter = null;
//把脫離條件隊列的一個接一個節(jié)點加入同步隊列，并喚醒
transferForSignal(first);
first = next;
} while (first != null);
}
final boolean transferForSignal(Node node) {
/*
* 把節(jié)點的狀態(tài)值從CONDITION設(shè)置為0
* 0的意義就是，他不屬于CONDITION，CANCELED，SIGNAL這三種狀態(tài)，屬于等待狀態(tài)。
* 如果設(shè)置失敗，就只能返回false了
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
// 這就是前面說的，加入同步隊列的地方
Node p = enq(node);
int ws = p.waitStatus;
// !compareAndSetWaitStatus(p, ws, Node.SIGNAL)這句話的意思是說，重設(shè)waitStatus可能會暫時無害的錯誤
// ws > 0代表Node的狀態(tài)是CANCELLED
// 整體來說就是，如果Node狀態(tài)是CANCELLED或者把Node的狀態(tài)設(shè)為SIGNAL失敗，
// 就把線程喚醒，也讓他去爭取鎖，源碼上寫這是暫時的，無害的錯誤
// 在AQS的acquireQueued方法里面，沒有獲取到鎖會一直卡在那個方法里面
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
后記
總結(jié)起來就是，要想讓規(guī)定數(shù)量的線程都達到同一個點才開始執(zhí)行，就得讓線程等待，計數(shù)。
AQS的條件隊列和同步隊列的設(shè)計正是用的這種思想，線程等待，就加入條件隊列，要釋放線程，因為還要獲取鎖才能越過CyclicBarrier的await方法，所以要加入同步隊列獲取鎖。
