CyclicBarrier使用和源码分析

CyclicBarrier使用和源码分析

一、CyclicBarrier用法

字面意思回环栅栏，通过它可以实现让一组线程等待至某个状态之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后，CyclicBarrier可以被重用。我们暂且把这个状态就叫做barrier，当调用await()方法之后，线程就处于barrier了。

　　CyclicBarrier类位于java.util.concurrent包下，CyclicBarrier提供2个构造器：

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指让多少个线程或者任务等待至barrier状态；参数barrierAction为当这些线程都达到barrier状态时会执行的内容。

　 然后CyclicBarrier中最重要的方法就是await方法，它有2个重载版本：

public int await(long timeout, TimeUnit unit)
       throws InterruptedException,
              BrokenBarrierException,
              TimeoutException {
       return dowait(true, unit.toNanos(timeout));
   }

public int await() throws InterruptedException, BrokenBarrierException {
       try {
           return dowait(false, 0L);
       } catch (TimeoutException toe) {
           throw new Error(toe); // cannot happen
       }
   }

　　第二个版本比较常用，用来挂起当前线程，直至所有线程都到达barrier状态再同时执行后续任务；

　　第一个版本是让这些线程等待至一定的时间，如果还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务。

　

二、源码分析

CyclicBarrier的核心方法就是await，先看无参的await方法

public int await() throws InterruptedException, BrokenBarrierException {
       try {
           return dowait(false, 0L);
       } catch (TimeoutException toe) {
           throw new Error(toe); // cannot happen
       }
   }

无参的await方法就是 就是挂起当前线程，等所有线程都到达barrier状态在同行执行后续任务。具体的实现在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();
//如果当前线处在中断状态，则放开栅栏，让其他已经在等待的线程同步运行，breakBarrier()方法
//就是放开当前所有在等待的线程，并重新设置generation，这样使栅栏可重用
           if (Thread.interrupted()) {
               breakBarrier();
               throw new InterruptedException();
           }

//count 未到barrier的线程个数，当前线程调用dowait方法时，count减一
//如果index=0，说明所有线程都到了barrier状态，则所有线程放行。
           int index = --count;
           if (index == 0) {  // tripped
               boolean ranAction = false;
               try {
                   final Runnable command = barrierCommand;
                   if (command != null)
                       command.run();
                   ranAction = true;
                   nextGeneration();
                   return 0;
               } finally {
                   if (!ranAction)
                       breakBarrier();
               }
           }
​
           // loop until tripped, broken, interrupted, or timed out
           //如果还有线程没有到达barrier状态，则等待直到所有线程到达barrier状态、或者栅栏broken、线程
           //中断、线程所设置的time期限已到，
           for (;;) {
               try {
               //若没有设置time，则直接等待
                   if (!timed)
                       trip.await();
                //若有时间等待期限，则等待一段时间
                   else if (nanos > 0L)
                       nanos = trip.awaitNanos(nanos);
               } catch (InterruptedException ie) {
               //过程中线程中断，先判断是不是当前generation和栅栏是否已经broken，如没有，则全部放行
                   if (g == generation && ! g.broken) {
                       breakBarrier();
                       throw ie;
                   } else {
                   //如不是当前generation或者generation已经broken，则中断当前线程
                       // 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();
                   }
               }
​
               if (g.broken)
                   throw new BrokenBarrierException();
​
               if (g != generation)
                   return index;
​
               if (timed && nanos <= 0L) {
                   breakBarrier();
                   throw new TimeoutException();
               }
           }
       } finally {
           lock.unlock();
       }
   }

无参的await方法没有设置time期限，则先判断有没有所有线程到达barrier状态，没有全部到达barrier状态，则等待，若都达到了barrier状态，则放行。

有参的await方法这是了time期限，也就是说此线程在等待的时候有一个time期限，若期限已到、栅栏意外broken、线程中断，同时此次generation是当前generation，则放行所有线程，若不是当前generation，则直接设置线程状态为中断状态。

总结

如果线程在等待的过程中发生中断异常、时间期限到了，只要还是在目前的generation中，就会发生breakBarrier，也就是放行。

若不是在目前的generation，就会中断线程。

参考：https://www.cnblogs.com/dolphin0520/p/3920397.html