一、CyclicBarrier

一个同步辅助类&＃xff0c;它允许一组线程互相等待&＃xff0c;直到到达某个公共屏障点 (common barrier point)。

CyclicBarrier循环栅栏&＃xff0c;Cyclic意味循环&＃xff0c;也就是这个计数器可以反复使用。

CyclicBarrier 支持一个可选的 Runnable命令&＃xff0c;在一组线程中的最后一个线程到达之后&＃xff08;但在释放所有线程之前&＃xff09;&＃xff0c;该命令只在每个屏障点运行一次。

CyclicBarrier适用于这样的情况&＃xff1a;创建一组任务&＃xff0c;并行执行工作&＃xff0c;然后在进行下一个步骤之前等待&＃xff0c;直至所有任务都完成&＃xff08;看起来有些像join()&＃xff09;。它使得所有的并行任务都将在栅栏处列队&＃xff0c;因此可以一致地向前移动&＃xff0c;这非常像CountDownLatch,只是CountDownLatch只触发一次的时间&＃xff0c;而 CyclicBarrier可以多次重用

1.1 API

示例用法&＃xff1a;下面是一个在并行分解设计中使用 barrier 的例子(伪代码)&＃xff1a;

class Solver {final int N;final float[][] data;final CyclicBarrier barrier;class Worker implements Runnable {int myRow;Worker(int row) { myRow &＃61; row; }public void run() {while (!done()) {processRow(myRow);try {barrier.await();} catch (InterruptedException ex) {return;} catch (BrokenBarrierException ex) {return;}}}}public Solver(float[][] matrix) {data &＃61; matrix;N &＃61; matrix.length;barrier &＃61; new CyclicBarrier(N,new Runnable() {public void run() {mergeRows(...);}});for (int i &＃61; 0; i }


在这个例子中&＃xff0c;每个worker 线程处理矩阵的一行&＃xff0c;在处理完所有的行之前&＃xff0c;该线程将一直在屏障处等待。处理完所有的行之后&＃xff0c;将执行所提供的 Runnable 屏障操作&＃xff0c;并合并这些行。如果合并者确定已经找到了一个解决方案&＃xff0c;那么done()将返回 true&＃xff0c;所有的worker 线程都将终止。

1.2 案例分析

案例一&＃xff1a;

CyclicBarrier使得每匹马都执行为了向前移动所必需执行的所有工作&＃xff0c;然后必须在栅栏处等待其他所有的马都准备完毕。 当所有的马都向前移动时&＃xff0c;CyclicBarrier将自动调用Runnable栅栏动作任务&＃xff0c;按顺序显示马和终点线的位置。

一旦所有的任务都越过栅栏&＃xff0c;它就会自动地为下一回合比赛做好准备。&＃xff08;可以拷贝运行查看运行结果&＃xff09;

package test;import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;class Horse implements Runnable {private static int counter &＃61; 0;private final int id &＃61; counter&＃43;&＃43;; //定义ID值private int strides &＃61; 0;private static Random random &＃61; new Random(47);private static CyclicBarrier cyclicBarrier;public Horse(CyclicBarrier cyclicBarrier) {this.cyclicBarrier &＃61; cyclicBarrier;}public synchronized int getStrides() {return strides;}&＃64;Overridepublic void run() {try {while (!Thread.interrupted()) {synchronized (this) { //使用this&＃xff0c;表示当前对象strides &＃43;&＃61; random.nextInt(3);//随机获取一个值。表示长度值}cyclicBarrier.await(); //一直等待&＃xff0c;当计数值达到0值时执行一次Runnable中的 run()}} catch (InterruptedException e) {} catch (BrokenBarrierException e) {throw new RuntimeException();}}public String toString() {return "Horse " &＃43; id &＃43; " ";}public String tracks() {StringBuilder stringBuilder &＃61; new StringBuilder();for (int i &＃61; 0; i "*");}stringBuilder.append(id);return stringBuilder.toString();}
}public class HorseRace {static final int FINISH_LINE &＃61; 75; //完成的步长private List horses &＃61; new ArrayList<>();private ExecutorService executorService &＃61; Executors.newCachedThreadPool();private CyclicBarrier cyclicBarrier;//在构造函数public HorseRace(int nHorses, final int pause) {// 在构造函数中给CyclicBarrier赋值&＃xff0c;//在一组线程中的最后一个线程到达之后&＃xff08;但在释放所有线程之前&＃xff09;&＃xff0c;该命令只在每个屏障点运行一次cyclicBarrier &＃61; new CyclicBarrier(nHorses, new Runnable() {&＃64;Overridepublic void run() {StringBuilder stringBuilder &＃61; new StringBuilder();for (int i &＃61; 0; i "&＃61;");}System.out.println(stringBuilder);for (Horse horse : horses) {System.out.println(horse.tracks());}for (Horse horse : horses) {if (horse.getStrides() >&＃61; FINISH_LINE) {System.out.println(horse &＃43; " won");executorService.shutdownNow();return;}}try {TimeUnit.MILLISECONDS.sleep(pause);} catch (InterruptedException e) {System.out.println("barrier-action sleep interruptedException");}}});// 启动比赛for (int i &＃61; 0; i new Horse(cyclicBarrier);horses.add(horse);executorService.execute(horse); //启动线程}}public static void main(String[] args) {int nHorse &＃61; 7;int pause &＃61; 200;new HorseRace(nHorse, pause);//第一个参数设置马的数量&＃xff0c;第二个参数设置暂停时间}
}

可以向CyclicBarrier提供一个动作&＃xff0c;它是一个Runnable,当计数值达到0时&＃xff0c;自动执行&＃xff0c;这是CyclicBarrier和CountDownLatch之间的另外一个区别。;

案例2

其他案例情况。

public class CyclicBarrierTest {public static void main(String[] args) {ExecutorService service &＃61; Executors.newCachedThreadPool();final CyclicBarrier cb &＃61; new CyclicBarrier(3);for(int i&＃61;0;i<3;i&＃43;&＃43;){Runnable runnable &＃61; new Runnable(){public void run(){try {Thread.sleep((long)(Math.random()*10000)); System.out.println("线程" &＃43; Thread.currentThread().getName() &＃43; "即将到达集合地点1&＃xff0c;当前已有" &＃43; (cb.getNumberWaiting()&＃43;1) &＃43; "个已经到达&＃xff0c;" &＃43; (cb.getNumberWaiting()&＃61;&＃61;2?"都到齐了&＃xff0c;继续走啊":"正在等候")); cb.await();Thread.sleep((long)(Math.random()*10000)); System.out.println("线程" &＃43; Thread.currentThread().getName() &＃43; "即将到达集合地点2&＃xff0c;当前已有" &＃43; (cb.getNumberWaiting()&＃43;1) &＃43; "个已经到达&＃xff0c;" &＃43; (cb.getNumberWaiting()&＃61;&＃61;2?"都到齐了&＃xff0c;继续走啊":"正在等候"));cb.await(); Thread.sleep((long)(Math.random()*10000)); System.out.println("线程" &＃43; Thread.currentThread().getName() &＃43; "即将到达集合地点3&＃xff0c;当前已有" &＃43; (cb.getNumberWaiting() &＃43; 1) &＃43; "个已经到达&＃xff0c;" &＃43; (cb.getNumberWaiting()&＃61;&＃61;2?"都到齐了&＃xff0c;继续走啊":"正在等候")); cb.await(); } catch (Exception e) {e.printStackTrace();} }};service.execute(runnable);}service.shutdown();}
}


运行结果

线程pool-1-thread-2即将到达集合地点1&＃xff0c;当前已有1个已经到达&＃xff0c;正在等候
线程pool-1-thread-1即将到达集合地点1&＃xff0c;当前已有2个已经到达&＃xff0c;正在等候
线程pool-1-thread-3即将到达集合地点1&＃xff0c;当前已有3个已经到达&＃xff0c;都到齐了&＃xff0c;继续走啊
线程pool-1-thread-3即将到达集合地点2&＃xff0c;当前已有1个已经到达&＃xff0c;正在等候
线程pool-1-thread-2即将到达集合地点2&＃xff0c;当前已有2个已经到达&＃xff0c;正在等候
线程pool-1-thread-1即将到达集合地点2&＃xff0c;当前已有3个已经到达&＃xff0c;都到齐了&＃xff0c;继续走啊
线程pool-1-thread-3即将到达集合地点3&＃xff0c;当前已有1个已经到达&＃xff0c;正在等候
线程pool-1-thread-2即将到达集合地点3&＃xff0c;当前已有2个已经到达&＃xff0c;正在等候
线程pool-1-thread-1即将到达集合地点3&＃xff0c;当前已有3个已经到达&＃xff0c;都到齐了&＃xff0c;继续走啊

1.2.1 CyclicBarrier 异常问题

Cyclicbarrier.await()方法可能会抛出两个异常&＃xff0c;一个是InterruptedException,也就是等待过程中&＃xff0c;线程被中断&＃xff0c;这是很常见的异常。 大部分迫使线程等待的方法都可能会抛出这个异常&＃xff0c;使得线程在等待时依然可以响应外部紧急事件。

另外一个异常则是CyclicBarrier特有的BrokenbarrierException。一旦遇到这个异常&＃xff0c;则表示当前的CyclicBarrier已经破损&＃xff0c;可能系统已经没有办法等待所有线程都到齐了,所以就撤销所有线程。

参考

1. 张孝祥-Java多线程与并发库高级应用
2. 《java编程思想》
3. 《实战Java高并发程序设计》