生产环境中,存在需要等待多个线程都达到某种状态后,才继续运行的情景。并发工具CyclicBarrier就能够完成这种功能。本篇从源码方面,简要分析CyclicBarrier的实现原理。
使用示例
public class CyclicBarrierTest { public static void main(String[] args) { //屏障,阻拦3个线程 CyclicBarrier cyclicBarrier = new CyclicBarrier(3); new Thread(new Runnable() { @Override public void run() { System.out.println("线程1正在执行"); try { // 等待 cyclicBarrier.await(); } catch (Exception e) { e.printStackTrace(); } System.out.println("线程1运行结束,时间: " + System.currentTimeMillis()); } }).start(); new Thread(new Runnable() { @Override public void run() { System.out.println("线程2正在执行"); try { // 等待 cyclicBarrier.await(); } catch (Exception e) { e.printStackTrace(); } System.out.println("线程2运行结束,时间: " + System.currentTimeMillis()); } }).start(); new Thread(new Runnable() { @Override public void run() { System.out.println("线程3正在执行"); try { //线程3阻塞2秒,测试效果 Thread.sleep(2000); // 等待 cyclicBarrier.await(); } catch (Exception e) { e.printStackTrace(); } System.out.println("线程3运行结束,时间: " + System.currentTimeMillis()); } }).start(); } }
执行结果如下:
线程1正在执行 线程2正在执行 线程3正在执行 线程1运行结束,时间: 1550324116837 线程3运行结束,时间: 1550324116837 线程2运行结束,时间: 1550324116837
可以看到线程1,2,3在同一个时间结束。
源码分析
主要成员:
private final ReentrantLock lock = new ReentrantLock(); private final Condition trip = lock.newCondition(); private int count;
CyclicBarrier主要借助重入锁ReentrantLock和Condition实现。count初始值等于CyclicBarrier实例化指明的等待线程数量,用于等待线程计数。
主要方法await()
public int await() throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } } private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { final ReentrantLock lock = this.lock; lock.lock(); // 1 try { final Generation g = generation; if (g.broken) throw new BrokenBarrierException(); if (Thread.interrupted()) { breakBarrier(); throw new InterruptedException(); } int index = --count; // 2 if (index == 0) { // 3 boolean ranAction = false; try { final Runnable command = barrierCommand; if (command != null) command.run(); ranAction = true; nextGeneration(); // 4 return 0; } finally { if (!ranAction) breakBarrier(); // 5 } } // loop until tripped, broken, interrupted, or timed out for (;;) { try { if (!timed) trip.await(); // 6 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 { lock.unlock(); // 7 } }
nextGeneration()的代码如下:
private void nextGeneration() { // signal completion of last generation trip.signalAll(); // set up next generation count = parties; generation = new Generation(); }
使用Condition的signalAll()方法,唤醒全部等待线程
说完CyclicBarrier的原理之后,再对本篇的使用示例做一下描述:
参与评论
手机查看
返回顶部