【JUC源码解析】CyclicBarrier
简介
CyclicBarrier,一个同步器,允许多个线程相互等待,直到达到一个公共屏障点。
概述
CyclicBarrier支持一个可选的 Runnable 命令,在一组线程中的最后一个线程到达之后,释放所有线程之前,该命令只在屏障点运行一次。
应用
描述
有一个矩阵,每一行数据交给一个线程去处理,处理内容是,将这行数据的每一个值相加,结果存入第一个元素中,每个线程处理完成后,会在屏障点相互等待,直到最后一个线程也到达屏障点,最后将各个线程处理的数据汇总,具体是将每一行汇总的数据(存在每行的第一个元素中的数据)再相加,结果存在第一行中的第一个元素。
具体如下代码所示。
代码
public class Solver { final int N; final float[][] data; // 待处理的数据 final CyclicBarrier barrier; // 屏障 class Worker implements Runnable { // 工作者 int myRow; Worker(int row) { myRow = row; } public void run() { System.out.println("Matrix[" + myRow + "]数据准备处理"); processRow(myRow); // 处理每一行数据,把各个数据相加,并且存入第一个元素 System.out.println("Matrix[" + myRow + "]数据处理完成"); try { barrier.await(); // 处理完成后,在此等待,直到最后一个线程也完成任务 } catch (InterruptedException ex) { return; } catch (BrokenBarrierException ex) { return; } } } public Solver(float[][] matrix) throws InterruptedException { data = matrix; N = matrix.length; System.out.println("开始处理数据,最初矩阵如下所示"); displayData(); // 数据展示 Runnable barrierAction = new Runnable() { // 操作完成后,由最后一个到达屏障点的线程执行此操作,整体只执行一次 public void run() { // 数据汇总,把第一列每行的和再相加,结果存入第一行第一个元素 float tmp = 0.0f; for (int i = 0; i < N; i++) { tmp += data[i][0]; } data[0][0] = tmp; System.out.println("Matrix[0]数据汇总完成,结果是" + tmp + ",存在了Matrix[0][0]"); } }; barrier = new CyclicBarrier(N, barrierAction); List<Thread> threads = new ArrayList<Thread>(N); for (int i = 0; i < N; i++) { Thread thread = new Thread(new Worker(i)); threads.add(thread); thread.start(); // 启动各个工作线程 } for (Thread thread : threads) { // 等待所有线程执行完成 thread.join(); } System.out.println("所以数据都已经处理完成,最终矩阵如下所示"); displayData(); } private void processRow(int myRow) { // 处理每一行数据,把各个数据相加,并且存入第一个元素 float[] row = data[myRow]; float tmp = 0.0f; int length = row.length; for (int i = 0; i < length; i++) { tmp += row[i]; } String msg = Arrays.toString(row); row[0] = tmp; System.out.println("Matrix[" + myRow + "]数据" + msg + "的和是:" + tmp + ", 且存入了Matrix[" + myRow + "][0]"); } private void displayData() { // 数据展示 TableInfo info = new TableInfo(N + 1); // TableInfo见下面代码,仅仅为了展示数据,可忽略,也可以用Arrays.toString(array); String[] header = new String[N + 1]; header[0] = String.valueOf(" "); for (int i = 0; i < N; i++) { header[i + 1] = String.valueOf(i); } info.addHeader(header); for (int i = 0; i < N; i++) { String[] tmp = new String[N + 1]; tmp[0] = String.valueOf(i); for (int j = 0; j < data[i].length; j++) { tmp[j + 1] = String.valueOf(data[i][j]); } info.addRecode(tmp); } System.out.println(info.getInfo()); } public static void main(String[] args) throws InterruptedException { Random r = new Random(47); float[][] matrix = new float[5][5]; for (int i = 0; i < 5; i++) { for (int j = 0; j < 5; j++) { matrix[i][j] = r.nextFloat(); } } new Solver(matrix); } } class TableInfo { private int count; private int[] maxLens; private String[] columns; private List<String[]> records; private StringBuilder builder; public TableInfo(int count) { this.count = count; maxLens = new int[count]; columns = new String[count]; builder = new StringBuilder(); records = new ArrayList<String[]>(); } private boolean isValid(String... args) { return null == args || args.length == 0; } private boolean isOutOfIndex(String... args) { return args.length > count; } public boolean addHeader(String... record) { if (isValid(record) || isOutOfIndex(record)) { return false; } copy(columns, record); copy(maxLens, record); return false; } public boolean addRecode(String... record) { if (isValid(record) || isOutOfIndex(record)) { return false; } copy(maxLens, record); records.add(record); return false; } public String getInfo() { buildLine(); buildHeader(); buildLine(); buildBody(); buildLine(); return builder.toString(); } private void buildHeader() { builder.append("|"); for (int i = 0; i < columns.length; i++) { builder.append(columns[i]); builder.append(getEmpty(maxLens[i] - columns[i].length())); builder.append("|"); } builder.append("\r\n"); } private void buildBody() { for (String[] recode : records) { builder.append("|"); for (int i = 0; i < recode.length; i++) { builder.append(recode[i]); builder.append(getEmpty(maxLens[i] - recode[i].length())); builder.append("|"); } builder.append("\r\n"); } } private void buildLine() { builder.append("+"); for (int i = 0; i < maxLens.length; i++) { builder.append(getLine(maxLens[i])); builder.append("+"); } builder.append("\r\n"); } private String getLine(int count) { StringBuilder builder = new StringBuilder(); for (int i = 0; i < count; i++) { builder.append("-"); } return builder.toString(); } private String getEmpty(int count) { StringBuilder builder = new StringBuilder(); for (int i = 0; i < count; i++) { builder.append(" "); } return builder.toString(); } private void copy(String[] args1, String... args2) { for (int i = 0; i < args2.length; i++) { args1[i] = args2[i]; } } private void copy(int[] args1, String... args2) { for (int i = 0; i < args2.length; i++) { args1[i] = swap(args1[i], args2[i].length()); } } private int swap(int max, int length) { if (max > length) { return max; } return length; }
输出
开始处理数据,最初矩阵如下所示 +----+----------+----------+----------+----------+----------+ | |0 |1 |2 |3 |4 | +----+----------+----------+----------+----------+----------+ |0 |0.72711575|0.39982635|0.5309454 |0.0534122 |0.16020656| |1 |0.57799757|0.18847865|0.4170137 |0.51660204|0.73734957| |2 |0.2678662 |0.9510573 |0.261361 |0.11435455|0.05086732| |3 |0.5466897 |0.8037155 |0.20143336|0.76206654|0.55373144| |4 |0.5304296 |0.15709275|0.5295954 |0.39661872|0.48718303| +----+----------+----------+----------+----------+----------+ Matrix[0]数据准备处理 Matrix[1]数据准备处理 Matrix[0]数据[0.72711575, 0.39982635, 0.5309454, 0.0534122, 0.16020656]的和是:1.8715063, 且存入了Matrix[0][0] Matrix[0]数据处理完成 Matrix[2]数据准备处理 Matrix[2]数据[0.2678662, 0.9510573, 0.261361, 0.11435455, 0.05086732]的和是:1.6455064, 且存入了Matrix[2][0] Matrix[2]数据处理完成 Matrix[1]数据[0.57799757, 0.18847865, 0.4170137, 0.51660204, 0.73734957]的和是:2.4374416, 且存入了Matrix[1][0] Matrix[1]数据处理完成 Matrix[4]数据准备处理 Matrix[3]数据准备处理 Matrix[3]数据[0.5466897, 0.8037155, 0.20143336, 0.76206654, 0.55373144]的和是:2.8676367, 且存入了Matrix[3][0] Matrix[3]数据处理完成 Matrix[4]数据[0.5304296, 0.15709275, 0.5295954, 0.39661872, 0.48718303]的和是:2.1009195, 且存入了Matrix[4][0] Matrix[4]数据处理完成 Matrix[0]数据汇总完成,结果是10.923011,存在了Matrix[0][0] 所以数据都已经处理完成,最终矩阵如下所示 +----+---------+----------+----------+----------+----------+ | |0 |1 |2 |3 |4 | +----+---------+----------+----------+----------+----------+ |0 |10.923011|0.39982635|0.5309454 |0.0534122 |0.16020656| |1 |2.4374416|0.18847865|0.4170137 |0.51660204|0.73734957| |2 |1.6455064|0.9510573 |0.261361 |0.11435455|0.05086732| |3 |2.8676367|0.8037155 |0.20143336|0.76206654|0.55373144| |4 |2.1009195|0.15709275|0.5295954 |0.39661872|0.48718303| +----+---------+----------+----------+----------+----------+
源码解析
关于静态内部类Generation,其属性broken描述CyclicBarrier是否被打破。每次到达屏障点,或者重置时,都会新生下一代(创建Generation实例,记录下一批线程的屏障状态)。
一组线程,执行一批任务,调用CyclicBarrier的await方法,其内部调用的是Condition的await方法,实质上是调用LockSupport.park方法,入队等待。也就是说,这组线程一调用await方法,就会再此阻塞,而最后一个线程调用await方法时,不在走Condition的await的代码分支,而是先执行barrierCommand任务,然后调用nextGeneration方法,在该方法内部,会调用Condition的signalAll方法,即是唤醒阻塞在Condition上的线程,重置count,并重新创建Generation实例,一遍下次使用。
reset方法,会先打破原有的屏障,即是Generation的broken设置为true,提前调用Condition的signalAll方法,释放阻塞着的线程。并且接着调用nextGeneration,开启一个新的Generation实例。
问题:为什么要多次创建Generation实例呢?重用一个实例不是更好吗?毕竟只有一个broken属性,加一个set方法就好了?
回答:之所以创建新的Generation,是因为,每个Generation实例对应一批冲向屏障的线程。假如,只有一个Generation实例,想像这样一个场景,某个线程调用了reset方法,本意是使另外一组线程能重新使用CyclicBarrier,并唤醒与它同一批阻塞在Generation上的线程,被唤醒的老一批线程需要记录屏障打破记录(reset方法会打破屏障状态,即是broken为true),如果是同一个Generation对象,且broken为true,而新线程又要求broken为false,因为是全新的,对它们来说,屏障没有被打破。因此,无法满足。
以下是源码:
属性
private static class Generation { // 分代,对应每一批冲向屏障的线程 boolean broken = false; // 记录屏障是否被打破 } private final ReentrantLock lock = new ReentrantLock(); // 可重入锁 private final Condition trip = lock.newCondition(); // 条件 private final int parties; // 记录线程数量 private final Runnable barrierCommand; // 最后一个到达屏障的线程需要执行的任务 private Generation generation = new Generation(); // 初代 private int count; // 还未到达屏障的线程个数,会再次重置为parties
创建下一代
private void nextGeneration() { // 创建下一代 trip.signalAll(); // 唤醒阻塞在该代屏障上的线程 count = parties; // 重置count generation = new Generation(); // 创建新的Generation实例 }
打破屏障
private void breakBarrier() { // 打破屏障 generation.broken = true; // 设置broken count = parties; // 重置count trip.signalAll(); // 唤醒阻塞的线程 }
关键的dowait
private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { final ReentrantLock lock = this.lock; // 可重入锁 lock.lock(); // 加锁 try { final Generation g = generation; // 对应该代线程的分代器,记录屏障的打破状态 if (g.broken) // 如果被打破,抛出异常 throw new BrokenBarrierException(); if (Thread.interrupted()) { // 如果线程中断了 breakBarrier(); // 打破屏障,并抛出异常 throw new InterruptedException(); } int index = --count; // 来一个线程,count减1 if (index == 0) { // 即将通过屏障 boolean ranAction = false; // 记录是否正常完成 try { final Runnable command = barrierCommand; // 屏障点任务,由最后一个到达的线程负责执行 if (command != null) command.run(); // 执行任务 ranAction = true; // 设置为正常完成 nextGeneration(); // 创建新的Generation对象 return 0; // 返回 } finally { if (!ranAction) // 如果没有正常完成,打破屏障 breakBarrier(); } } for (;;) { try { if (!timed) // 如果没有设置超时 trip.await(); // 调用await方法 else if (nanos > 0L) // 否则,调用awaitNanos方法 nanos = trip.awaitNanos(nanos); } catch (InterruptedException ie) { // 如果是中断唤醒的,则看是否换代 if (g == generation && !g.broken) { // 如果还是同一代,并且屏障没有被打破,那么打破屏障,并抛出异常 breakBarrier(); throw ie; } else { // 如果换代了,或者屏障已经打破了,什么都不作,仅仅重新设置中断标记 Thread.currentThread().interrupt(); } } if (g.broken) // 如果是正常唤醒的,并且屏障已经打破,抛出异常 throw new BrokenBarrierException(); if (g != generation) // 超时,换代了,返回未到达屏障的线程数目 return index; if (timed && nanos <= 0L) { // 超时, 没换代 breakBarrier(); // 打破屏障 throw new TimeoutException(); // 抛出超时异常 } } } finally { lock.unlock(); // 解锁 } }
行文至此结束。
尊重他人的劳动,转载请注明出处:http://www.cnblogs.com/aniao/p/aniao_cyclicbarrier.html
- 【JUC源码解析】ConcurrentLinkedQueue
- 【JUC源码解析】PriorityBlockingQueue
- 【JUC源码解析】CountDownLatch
- 【JUC源码解析】Semaphore
- Java - "JUC"之Condition源码解析
- 【JUC源码解析】ScheduledThreadPoolExecutor
- 【JUC源码解析】CompletableFuture
- JUC.Condition学习笔记[附详细源码解析]
- 【JUC源码解析】LinkedBlockingQueue
- 【JUC源码解析】DelayQueue
- 【JUC源码解析】ForkJoinPool
- JUC.Lock(锁机制)学习笔记[附详细源码解析]
- Jdk1.6 JUC源码解析(7)-locks-ReentrantLock
- 【JUC源码解析】SynchronousQueue
- 【JUC源码解析】FutureTask
- 【JUC源码解析】Exchanger
- JUC源码解析(6)-locks-AbstractQueuedSynchronizer
- Jdk1.6 JUC源码解析(12)-ArrayBlockingQueue
- 【JUC源码解析】ConcurrentHashMap
- 【JUC源码解析】ReentrantReadWriteLock