同步器-CyclicBarrier

Dear 丶 2022-05-24 03:14 140阅读 0赞

java.util.concurrent 包包含了几个可以帮助我们管理线程间相互合作的集合类， 这些机制具有为线程之间的共用集结点模式 （ common rendezvous patterns ) 提供的 “ 预置功能 ”( canned functionality )

CyclicBarrier允许线程集等待直至其中预定数目的线程到达一个公共障栅 （ barrier )

， 然后可以选择执行一个处理障栅的动作

当大量的线程需要在它们的结果可用之前完成时

CyclicBarrier 类实现了一个集结点 （ rendezvous ) 称为障栅 （ barrier ) ； 考虑大量线程运行在一次计算的不同部分的情形。 当所有部分都准备好时 ， 需要把结果组合在一起 。 当一个线程完成了它的那部分任务后 ， 我们让它运行到障栅处 。 一旦所有的线程都到达了这个障栅，障栅就撤销 ， 线程就可以继续运行。

/\*\*  
 \*   
 \* @Date 2018年5月19日 下午12:57:39  
 \* @version 1.0  
 \*/  
public class CyclicBarrierTest \{  
public static CyclicBarrier c = new CyclicBarrier(2, new Runnable() \{  
@Override  
public void run() \{  
System.out.println("innner here !");  
\}  
\});  
  
  
public static void main(String\[\] args) \{  
for(int i = 0;i<2;i++) \{  
final int j = i;  
new Thread(new Runnable() \{  
@Override  
public void run() \{  
System.out.println(j + " " + "start!");  
try \{  
c.await();  
\} catch (Exception e) \{  
System.out.println(e.getMessage() + "错误信息");  
\}  
System.out.println("end" + " " + j);  
\}  
\}).start();  
\}  
System.out.println(2 + " " + "main end");  
\}  
\}

流程 ： 首先 ，构造一个障栅 ， 并给出参与的线程数 ：

CyclicBarrier barrier = new CydicBarrier ( nthreads ) ;

每一个线程做一些工作， 完成后在障栅上调用 await :  
public void runO  
\{doWork（）;barrier . await（）；\}

await 方法有一个可选的超时参数 ：barrier . await（100， TineUnit . MILLISECONDS ) ;

如果任何一个在障栅上等待的线程离开了障栅 ， 那么障栅就被破坏了 （ 线程可能离开是因为它调用 await 时设置了超时 ， 或者因为它被中断了 ）。 在这种情况下 ， 所有其他线程的await 方法抛出 BrokenBarrierException 异常 。 那些已经在等待的线程立即终止 await 的调用 。可以提供一个可选的障栅动作 （ barrier action ) , 当所有线程到达障栅的时候就会执行这一动作。

Runnable barrierAction = . .  
CyclicBarrier barrier = new CyclicBarrier ( nthreads , barrierAction ) ;  
该动作可以收集那些单个线程的运行结果。  
障栅被称为是循环的 （ cyclic ) , 因为可以在所有等待线程被释放后被重用。 在这一点上 ，有别于 CountDownLatch , CountDownLatch 只能被使用一次。

/\*\*  
 \* 等待超时，抛出异常  
 \* @Date 2018年5月19日 下午12:57:39  
 \* @version 1.0  
 \*/  
public class CyclicBarrierTest \{  
public static CyclicBarrier c = new CyclicBarrier(2, new Runnable() \{  
@Override  
public void run() \{  
System.out.println("innner here !");  
\}  
\});  
  
  
public static void main(String\[\] args) \{  
for(int i = 0;i<2;i++) \{  
final int j = i;  
new Thread(new Runnable() \{  
@Override  
public void run() \{  
System.out.println(j + " " + "start!");  
  
try \{  
if(j==0)\{  
Thread.currentThread().sleep(10000);  
c.await();  
\}else \{  
c.await(100,TimeUnit.MILLISECONDS);  
\}  
\} catch (Exception e) \{  
e.printStackTrace();  
System.out.println(e.getMessage() + "错误信息");  
\}  
System.out.println("end" + " " + j);  
\}  
\}).start();  
\}  
System.out.println(2 + " " + "main end");  
\}  
\}