CyclicBarrier

文章目录

简介用法类属性await()总结

简介

CyclicBarrier可以阻塞一组线程直到这些线程同时达到某个条件才继续执行。 当线程到达栅栏位置时将调用await方法，这个方法将阻塞当前线程直到所有线程都到达栅栏位置。如果所有线程都到达栅栏位置，那么栅栏将打开，此时所有的线程都将被释放，而栅栏将被被重置以便下次使用。

用法

public static void main(String[] args) { CyclicBarrier c = new CyclicBarrier(5); for (int i = 0; i < 5; i++) { new Thread(()->{ try { System.out.println(Thread.currentThread().getName() + "...inti"); // 1 c.await(); // 2 System.out.println(Thread.currentThread().getName() + "...end"); // 3 } catch (Exception e){ e.printStackTrace(); } }).start(); } }

上面的代码表示，当5个线程都运行到位置3时，其它4个线程才回继续运行，否则其它4个线程将会一直阻塞。

类属性

// 内部类，用于控制CyclicBarrier的循环使用 private static class Generation { // broken 代表栅栏是否被破坏 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; // 构造方法 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; }

await()

// CyclicBarrier.await() public int await() throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } } // CyclicBarrier.dowait(boolean timed, long nanos) private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { final ReentrantLock lock = this.lock; // 获取锁 lock.lock(); try { final Generation g = generation; // g.broken 初始化为false if (g.broken) throw new BrokenBarrierException(); // 中断检查 if (Thread.interrupted()) { // 如果当前线程被中断 // 将损坏状态设置为true // 并通知其他阻塞在此栅栏上的线程 breakBarrier(); throw new InterruptedException(); } int index = --count; if (index == 0) { // tripped // count == 0，说明已经是最后一个线程了 boolean ranAction = false; try { // barrierCommand 为实例化时传入代码，当count为0的时候就执行它 final Runnable command = barrierCommand; if (command != null) command.run(); ranAction = true; // 调用下一代，该方法主要实现唤醒正在阻塞的线程 nextGeneration(); return 0; } finally { if (!ranAction) // 如果执行栅栏任务的时候失败了，就将损坏状态设置为true breakBarrier(); } } // 从上面的代码可以知道，最后一个线程是不会执行到这里的，除了最后一个线程，其它线程都将被阻塞在这里 for (;;) { try { // 如果我们传入进来的timed为false if (!timed) trip.await(); // 释放锁，当前线程被构造成节点加入到等待队列中，并阻塞自己 else if (nanos > 0L) 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(); // 如果程序运行正常，那么这里肯定不会像等了 // 因为在nextGeneration方法中，被重新实例化了generation if (g != generation) return index; if (timed && nanos <= 0L) { // 如果有时间限制，且时间小于等于0，销毁栅栏并抛出异常 breakBarrier(); throw new TimeoutException(); } } } finally { // 释放锁 lock.unlock(); } } // CyclicBarrier.breakBarrier() private void breakBarrier() { generation.broken = true; count = parties; // 唤醒正在阻塞的线程，把等待队列中的线程转移到AQS同步队列中 trip.signalAll(); } // CyclicBarrier.nextGeneration() private void nextGeneration() { // 唤醒正在阻塞的线程，把等待队列中的线程转移到AQS同步队列中 trip.signalAll(); // 重新设置下一代需要等待的线程数 count = parties; generation = new Generation(); }

总结

CyclicBarrier能够实现一组线程阻塞在await()里，直到最后一个线程运行到await()方法时，才会去唤醒前面被阻塞的线程，让前面被阻塞的线程继续运行。

CyclicBarrier与CountDownLatch的异区别：

Countdownlatch是一次性的，计数器的值只能在构造方法中初始化一次，之后没有任何机制再次对其设置值，当CountDownLatch使用完毕后，他不能再次被使用 。CyclicBarrier只能现多个线程到达栅栏处一起运行 。CountDownLatch 不仅能实现一个线程等待多个线程条件成立，还能实现多个线程等待一个线程条件成立。