CyclicBarrier笔记 CyclicBarrier笔记

CyclicBarrier 定义
jdk源码对CyclicBarrier的定义是这样的：

A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point.

翻译下来的意思即：允许一组线程都等待对方都到达一个共同屏障点的同步辅助工具。另外，Cyclic表示它是一个可重复使用的工具，在每次同步协作后，只要是正常完成任务或者调用reset函数，它都能够重置状态并准备好在下次同步协助时被使用。
使用场景
在一些场景下，我需要可能会将一些复杂的任务划为多个子任务来处理，在一些特殊的情况下，我们需要子任务执行到某一个位置（我们称之为屏障点），停下来等待所有其他子任务也执行到某一个位置，然后所有子任务再继续往下处理；或者是所有子任务都处理到某个屏障点时，去触发某个任务或者指令。这里边的同步等待顺序如果要开发者自己实现，可能需要开发者对多线程并发协作有一个比较好的基础，而jdk针对这个多线程同步协作的多线程并发模型，封装为CyclicBarrier同步辅助工具，使得用户可以比较简单地直接使用。
官方举了一个合并矩阵的应用场景：
有一个多行矩阵，要求在每一行处理完成后进行合并操作，利用CyclicBarrier这个辅助工具就可以简单的进行实现。

class Solver { final int N; final float[][] data; final CyclicBarrier barrier; // 每一行的处理线程，简称 worker class Worker implements Runnable { int myRow; Worker(int row) { myRow = row; } public void run() { while (!done()) { // 处理行 processRow(myRow); try { // 处理完成后，同步等待公共屏障, 这一步实际是等待所有Worker线程都执行到barrier.await() // 实际上前所有处理行的子任务都到达这屏障点后，就会自动处理合并行的任务（见Solver类中定义的barrierAction）。 barrier.await(); } catch (InterruptedException ex) { return; } catch (BrokenBarrierException ex) { return; } } } } // 处理者，进行任务分派 public Solver(float[][] matrix) { data = https://www.it610.com/article/matrix; N = matrix.length; // 定义到达公共屏幕后需要处理的行为指令，这里为合并行操作。 Runnable barrierAction = new Runnable() { public void run() { mergeRows(...); }}; barrier = new CyclicBarrier(N, barrierAction); List threads = new ArrayList(N); for (int i = 0; i < N; i++) { // 启动多个子任务，每个任务都是负责处理矩阵其中一行 Thread thread = new Thread(new Worker(i)); threads.add(thread); thread.start(); } // wait until done for (Thread thread : threads) thread.join(); } }}

底层实现
构造函数 【CyclicBarrier笔记】CyclicBarrier类中提供了两个构造函数，分别为CyclicBarrier(int parties)和CyclicBarrier(int parties, Runnable barrierAction)，如下

public CyclicBarrier(int parties) { this(parties, null); }public CyclicBarrier(int parties, Runnable barrierAction) { if (parties <= 0) throw new IllegalArgumentException(); this.parties = parties; this.count = parties; this.barrierCommand = barrierAction; }

parties参数表示需要到达屏障点的次数，通常与划分出来的子任务的数量一致。调用多少cyclicBarrier对象的await()函数都会认为一个任务到达屏障点，所以这个参数实际上代表的涵义是：需要调用多少次cyclicBarrier对象的await()函数才会认为所有任务都到达了屏障点并触发任务指令。
barrierAction表示任务指令，即所有进行协助的线程都到达屏障点后需要执行的任务操作。这个参数是可选的。

await CyclicBarrier类中提供了两个await函数，分别是await()和await(long timeout, TimeUnit unit)，如下

public int await() throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } } public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException { return dowait(true, unit.toNanos(timeout)); }

从代码可知，这两个函数主要区别在于：一个是无时间限制的同步等待，另一个是有时间限制的同步等待。它们的主要逻辑还是通过dowai函数来实现。
dowait(核心实现) 在了解dowait之间，我们需要了解一个名词：
Generation：一代，在cycylicBarrier的实现逻辑中，以『代』来区分每一次的任务协作，一个任务被完成后，那么就被创建一个新的代。
dowait函数是cyclicBarrier的核心，它主要的处理逻辑顺序是这样的：

使用ReentrantLock获取锁对象
检查当前代是否被破坏或者线程中断，如果满足一个条件，则抛出对应的异常。
paries的副本count进行自减，如果自减等于0，说明当前所有协助任务线程都都已经到达屏幕点，这时候如果barrierCommand不null，就会被触发(PS：barrierCommand由最后一个到达屏障点的线程来执行。如果barrierCommand执行错误，那么就会调用breakBarrier函数标识break为true，并唤醒其他正常阻塞的协作任务线程。否则，调用nextGeneration函数进行重置状态和创建新的代并唤醒其他正常阻塞的协作任务线程，同时为下一次协作做好准备。
paries的副本count如果自减后不等于0，那么表示尚有其他协作线程并到到达屏障点，通过for无限循环进行等待，直至超时或者被其他线程唤醒。唤醒后有两种不同的结果：
- 异常：表示当前这一代的协作任务被标识为break，可能原因为线程中断，barrierCommand执行错误、其他线程超时。
- 正常：当前代已经被更新，说明上一代的任务协作已经正常完成。

具体的细节可以看下边的源码：

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(); }// count自减(count是parties的一个副本) int index = --count; // 判断count自减后是否等于0 if (index == 0) {// tripped boolean ranAction = false; try { // 如果为0，表示所有线程都到达屏障点，那么触发屏障指令任务。 final Runnable command = barrierCommand; if (command != null) command.run(); ranAction = true; // 执行nextGeneration，重置状态，准备下次使用; nextGeneration(); return 0; } finally { if (!ranAction) breakBarrier(); } }// 循环直至以下条件：1、所有线程到达屏幕点，2、当前这一代broken，3、中断、4、超时 // loop until tripped, broken, interrupted, or timed out for (; ; ) { try { // 同步阻塞 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(); } }// 如果这一代被标为broken，那么抛出BrokenBarrierException if (g.broken) throw new BrokenBarrierException(); // 如果发现代发生了变化，那么表示当前这一代已经完成，返回index， // index表示当前为第parties-index个到达屏障点的线程。 if (g != generation) return index; // 如果超时，调用breakBarrier并返回TimeoutException if (timed && nanos <= 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { lock.unlock(); } }

从dowait的实现逻辑来看，我们可以看出

内部线程间的通讯采用ReentrantLock和Condiction来实现线程音的消息传递。
在协作任务正常返回前，线程中断、BarrierCommand执行异常和线程触发breakBarrier函数都能使CyclicBarrier的dowait提前退出。

barrier的内部状态在前一个小节中，我们发现dowait的逻辑中会调用nextGentration和breakBarrier函数，这两个函数是cycilcBarrier中更新内部状态和唤醒协作线程的函数，这一节我们主要讲一下barrier内部状态的一个变化过程。
nextGenration nextGenration函数仅这一代的协作任务正常完成时被调用，通过它来唤醒其他协作线程，并且对barrier进行代的更新。一旦调用nextGenration函数后，generation的引用将被更新，这时候它又可以被重新利用。

private void nextGeneration() { // signal completion of last generation trip.signalAll(); // set up next generation count = parties; generation = new Generation(); }

breakBarrier breakBarrier函数被调用发生在检测到当前线程中断、等待超时、BarrierCommand执行异常。它的具体逻辑为设置broken标识为true，重置count，然后唤醒其他所有等待线程。

private void breakBarrier() { generation.broken = true; count = parties; trip.signalAll(); }

从实现来看，breakBarrier函数并没有更新generation，结合breakBarrier和dowait的逻辑来看，breakBarrier函数一时被调用，那么barrier就进入了一个broken的状态，它已经不具备重复循环复用的条件，一旦有线程调用dowait，那么它将直接抛出BrokenBarrierException。那么有什么方法让它重新可用呢，那就是调用reset函数。
reset reset可以重置CyclicBarrier的状态，从来的实现来看，它会先调用breakBarrier，这意味着如果当前barrier有协作任务正在处理，那么将会影响它们的处理结果。在breakBarrier后，会调用nextGeneration来重置内部状态并更新generation，这时候它就变得可重复复用了。

public void reset() { final ReentrantLock lock = this.lock; lock.lock(); try { breakBarrier(); // break the current generation nextGeneration(); // start a new generation } finally { lock.unlock(); } }

总结

CyclicBarrier是一个可重复使用的多线程协作的同步辅助工具，允许一组线程都等待对方都到达一个共同屏障点后执行指定的任务指令。
仅当前一代任务协作正常结束或调用reset方法后，CyclicBarrier才可被再次使用；线程中断、超时以及其他线程调用reset将导致当前CyclicBarrier这一代的状态被破坏。
内部主维护parties的副本count，每有一个线程到达屏障点就自减，通过count是否等于0来判断是否这一代的任务协助是否完成，否则进入等待。
多线程协作过程中，主要通过ReentranLock和Condition来实现线程间的状态传递。