CyclicBarrier是循环栅栏的意思,循环的等待多个线程执行任务;
<1> 示例代码如下:
public class CyclicBarrierTest { public static CyclicBarrier cb = new CyclicBarrier(3, () -> System.out.println("-------开始点名-------")); public static void main(String[] args) { System.out.println("三人组团出游,门口等待,准备点名"); for (int i = 0; i < 3 ; i++) { new Thread(() -> { System.out.println(Thread.currentThread().getName() + "到达集合点"); try { cb.await(); }catch (Exception e){} System.out.println(Thread.currentThread().getName() + "到车上"); try { cb.await(); }catch (Exception e){} System.out.println(Thread.currentThread().getName() + "到景区"); try { cb.await(); }catch (Exception e){} }, "线程" + i).start(); } }
}
执行结果:
<2> CyclicBarrier 源码简析:
- 构造方法
/** * CyclicBarrier JDK1.8分析 * @param parties 等待的线程个数 * @param barrierAction 所有等待的线程就绪后要执行任务 */ public CyclicBarrier(int parties, Runnable barrierAction) { if (parties <= 0) throw new IllegalArgumentException(); this.parties = parties; this.count = parties; this.barrierCommand = barrierAction; }
- await() 方法
/** * 每一个线程在某个点等待其他线程,执行await方法 * @return * @throws InterruptedException * @throws BrokenBarrierException */ public int await() throws InterruptedException, BrokenBarrierException { try { //具体的功能方法 return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } }
-
dowait()栅栏的核心实现
/** * Main barrier code, covering the various policies. * * 栅栏的核心代码,包含各种条件 * * ReentrantLock lock - CyclicBarrier是用Lock锁实现的,这里用来控制线程的执行 * * CyclicBarrier.Generation 是一个内部类,只是用来进行状态标识 * */ private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { final ReentrantLock lock = this.lock; //每个线程执行await方法获取锁 lock.lock(); try { //CyclicBarrier在初始化的时候,创建的g赋值给局部变量,在后面使用 final CyclicBarrier.Generation g = generation; //g.broken 默认是false,如果是true表示栅栏出问题了,抛出异常 if (g.broken) throw new BrokenBarrierException(); //这里判断当前执行await()的线程状态,如果是中断,表明程序出问题了 if (Thread.interrupted()) { //初始化栅栏数据,并将g.broken设置为true breakBarrier(); throw new InterruptedException(); } //count用来记录,还需要几个线程执行await()方法,可以冲破栅栏 int index = --count; //index == 0 表示等待的线程数量够了 if (index == 0) { // tripped boolean ranAction = false; try { final Runnable command = barrierCommand; if (command != null) //执行构造函数里面传进来的任务,这里是由最后一个达到的线程执行的 command.run(); //表明这次等待任务完成,准备执行下一次栅栏任务 ranAction = true; //该方法里用:trip.signalAll(); 唤醒进入方法里的其他线程,重新初始化 count 和 generation //为下一次执行栅栏做准备 nextGeneration(); return 0; } finally { if (!ranAction) breakBarrier(); } } // loop until tripped, broken, interrupted, or timed out for (;;) { try { if (!timed) //执行await()方法获取到锁执行到此,调用condition.await()方法进行挂起,并释放锁,让其他线程获取锁 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(); } } if (g.broken) throw new BrokenBarrierException(); if (g != generation) return index; if (timed && nanos <= 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { //释放锁,之前进来的线程,都在调condition.await()方法的时候已经释放锁了,所以这里只有最后一个进来的线程有效果 lock.unlock(); } }
(笔记整理,待完善)