導(dǎo)讀：這篇文章介紹的是java并發(fā)組件aqs之CyclicBarrier
Cyclicbarrier概念：
-
Cyclicbarrier是一個同步輔助類皂冰，它允許一組線程相互等待直到到達(dá)某個公共的屏障點鞠眉，通過它可以完成多個線程之間相互等待，直有當(dāng)每個線程都準(zhǔn)備就緒后才能各自繼續(xù)往下執(zhí)行后面的操作固耘，和countdownlatch有相似的地方，都是通過計數(shù)器來實現(xiàn)的，當(dāng)某個線程調(diào)用了await()方法之后該線程就進(jìn)入了等待狀態(tài)而且計數(shù)器執(zhí)行的是加一的操作文判，當(dāng)計數(shù)器的值達(dá)到我們設(shè)置的初始值的時候胧谈，因為調(diào)用了await()方法進(jìn)入等待狀態(tài)的線程會被喚醒繼續(xù)執(zhí)行它們后續(xù)的操作忆肾。由于CyclicBarrier釋放等待線程后可以重用所以我們又稱為循環(huán)屏障。
cyclicbarrier
Cyclicbarrier使用場景：
-
Cyclicbarrier的使用場景跟Countdownlatch很相似菱肖，可以用于多線程計算數(shù)據(jù)最后合并計算結(jié)果的應(yīng)用場景客冈。比如我們需要統(tǒng)計多個Excel中的數(shù)據(jù)，然后等到一個總結(jié)果稳强。我們可以通過多線程處理每一個Excel场仲，執(zhí)行完成后得到相應(yīng)的結(jié)果，最后通過Barrieraction來計算這些線程的計算結(jié)果退疫，得到所有Excel的總和渠缕。
CyclicBarrier中的方法
下面通過實例來展示下Cyclicbarrier的使用
- 1、示列一
@Slf4j
public class CyclicBarrierExample1 {
private static CyclicBarrier barrier = new CyclicBarrier(5); //定義5個線程同步等待
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
barrier.await(); //線程調(diào)用await()告訴當(dāng)前有個線程ok了
log.info("{} continue", threadNum); //到達(dá)定義的數(shù)目時褒繁，await()方法之后的操作就可以執(zhí)行
}
}
返回結(jié)果：可以看出通過下面的返回結(jié)果亦鳞。代碼中定義了5個線程同步等待，當(dāng)rece()中每個線程調(diào)用await()告訴當(dāng)前線程ok了，當(dāng)累計到我們之前定義的線程數(shù)時蚜迅，await()方法后的操作開始執(zhí)行舵匾。
- 2、示列二
@Slf4j
public class CyclicBarrierExample2 {
private static CyclicBarrier barrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
try {
// 由于狀態(tài)可能會改變谁不，所以會拋出BarrierException異常坐梯，如果想繼續(xù)往下執(zhí)行，需要加上try-catch
barrier.await(2000, TimeUnit.MILLISECONDS); //用于做超時處理
} catch (Exception e) {
log.warn("BarrierException", e);
}
log.info("{} continue", threadNum);
}
}
返回結(jié)果：注意查看下返回結(jié)果的時間刹帕，示例二和示例一的代碼差不多吵血，我們加了個await的超時處理，
這個過程中如果超時狀態(tài)可能會改變偷溺，所以會拋出BarrierException異常蹋辅，這時候如果要讓后續(xù)的操作繼續(xù)執(zhí)行，得捕捉異常挫掏。
- 3侦另、示列三
@Slf4j
public class CyclicBarrierExample3 {
private static CyclicBarrier barrier = new CyclicBarrier(5, () -> {
// 當(dāng)線程全部到達(dá)屏障時，優(yōu)先執(zhí)行這里的runnable
log.info("callback is running");
});
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
barrier.await();
log.info("{} continue", threadNum);
}
}
返回結(jié)果：這里我們定義CyclicBarrier(int parties, Runnable barrierAction)尉共，指定了一個runnable就是在線程到達(dá)屏障的時候優(yōu)先執(zhí)行runnable褒傅。
總結(jié)：CyclicBarrier的內(nèi)部是使用重入鎖ReentrantLock和Condition，在CyclicBarrier中最重要的方法莫過于await()方法袄友，表示在此的所有線程都已經(jīng)在此barrier上調(diào)用了await()方法之前殿托，將一直等待。同時await()方法內(nèi)部調(diào)用了dowait(boolean timed, long nanos)方法剧蚣，更多詳細(xì)的可以看CyclicBarrier的類源碼(目前先不記錄了)支竹，await()的處理邏輯還是比較簡單的：如果該線程不是到達(dá)的最后一個線程，則他會一直處于等待狀態(tài)鸠按，除非發(fā)生以下情況:
1礼搁、最后一個線程到達(dá)，即index == 0
2待诅、超出了指定時間（超時等待）
3叹坦、其他的某個線程中斷當(dāng)前線程
4、其他的某個線程中斷另一個等待的線程
5卑雁、其他的某個線程在等待barrier超時
6募书、其他的某個線程在此barrier調(diào)用reset()方法。reset()方法用于將屏障重置為初始狀態(tài)测蹲。
