生产环境中,存在需要等待多个线程都达到某种状态后,才继续运行的情景。并发工具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的原理之后,再对本篇的使用示例做一下描述:
如对本文有疑问, 点击进行留言回复!!
动态内存申请(malloc, calloc, new)之分配虚拟内存空间和物理内存空间
Ubuntu 16.04 离线升级Openssl(1.1.1g) 和 Openssh(8.3p1)
网友评论