当前位置: 首页 > news >正文

【同步工具类:CyclicBarrier】

同步工具类:CyclicBarrier

  • 介绍
  • 源码分析
    • CyclicBarrier 基于ReetrantLock + Condition实现。
    • 构造函数
    • await() 函数
  • 业务场景
    • 方案一:
      • 代码实现
      • 测试截图
    • 方案二
      • 代码实现
      • 测试打印
  • 总结

介绍

官方介绍:
一种同步辅助工具,允许一组线程都等待对方到达共同的障碍点。CyclicBarrier在涉及固定大小的线程组的程序中非常有用,这些线程组偶尔必须彼此等待。该屏障被称为循环屏障,因为它可以在释放等待线程后重新使用。
CyclicBarrier支持可选的Runnable命令,该命令在参与方中的最后一个线程到达后,但在释放任何线程之前,在每个障碍点运行一次。此屏障动作对于在任何一方继续之前更新共享状态都很有用。
通俗理解:
它可以协同多个线程,让多个线程在这个栅栏前等待,直到所有线程都达到了这个栅栏时,再一起继续执行后面的动作.
举个例子,你和朋友约定在公交站汇合,去公园玩。这个公交站相当于栅栏。只有你们都到了公交站,才一起去公园。

源码分析

CyclicBarrier 基于ReetrantLock + Condition实现。

    /** The lock for guarding barrier entry *///用于线程之间互相唤醒private final ReentrantLock lock = new ReentrantLock();/** Condition to wait on until tripped */private final Condition trip = lock.newCondition();//总线程数private final int parties;

构造函数

可以看到,不仅可以传入 参与方的总数量(即 parties)。还可以传入一个回调函数,当所有的线程被唤醒时,barrierAction 被执行,该参数可以为空。

    /*** Creates a new {@code CyclicBarrier} that will trip when the* given number of parties (threads) are waiting upon it, and which* will execute the given barrier action when the barrier is tripped,* performed by the last thread entering the barrier.** @param parties the number of threads that must invoke {@link #await}*        before the barrier is tripped* @param barrierAction the command to execute when the barrier is*        tripped, or {@code null} if there is no action* @throws IllegalArgumentException if {@code parties} is less than 1*/public CyclicBarrier(int parties, Runnable barrierAction) {if (parties <= 0) throw new IllegalArgumentException();this.parties = parties;this.count = parties;this.barrierCommand = barrierAction;}

await() 函数

1.CyclicBarrier 是可以被重用的。
2.CyclicBarrier 会响应中断,N 个线程还没有到齐,如果有线程收到了中断信号,所有阻塞的线程也会被唤醒。也就是 breakBarrier函数。然后count 被重置为初始值(parties),重新开始
3.构造函数传入的回调函数,barrierAction 只会被最后一个线程执行一次。

 public int await() throws InterruptedException, BrokenBarrierException {try {return dowait(false, 0L);} catch (TimeoutException toe) {throw new Error(toe); // cannot happen}}
    /*** Main barrier code, covering the various policies.*/private int dowait(boolean timed, long nanos)throws InterruptedException, BrokenBarrierException,TimeoutException {final ReentrantLock lock = this.lock;lock.lock();try {final Generation g = generation;if (g.broken)throw new BrokenBarrierException();if (Thread.interrupted()) {  //响应中断breakBarrier();  //唤醒所有阻塞的线程throw new InterruptedException();}int index = --count;  //每个线程调用一次await(). count 减一,当count==0时,则唤醒其他的所有线程if (index == 0) {  // trippedboolean ranAction = false;try {final Runnable command = barrierCommand;if (command != null)// 一起唤醒之和,如果回调函数不为空,还需要执行回调函数command.run();ranAction = true;nextGeneration();//唤醒其他所有线程,并将count值复原。//用于下一次的CyclicBarrier.这是可以复用的原因return 0;} finally {if (!ranAction)breakBarrier();}}// loop until tripped, broken, interrupted, or timed out//当count>0,说明 人没有到齐,需要阻塞自己for (;;) {try {if (!timed)trip.await();//当阻塞自己的时候,await方法会释放锁,这样其他线程调用await方法时会执行--countelse 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.//如果不是响应的中断,说明是被 sigalAll唤醒。则自己唤醒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();}}
     private void nextGeneration() {// signal completion of last generation// 唤醒所有阻塞的线程trip.signalAll();// set up next generation// 设置初始值,开始下一个轮回count = parties;generation = new Generation();}

业务场景

10 个求职者一起来公司应聘,招聘方式为笔试和面试。首先,需要等10个人到期后,开始笔试,笔试结束之后,再一起参加面试。把10个人看作10个线程。如图所示:
在这里插入图片描述

方案一:

采用一个CyclicBarrier.重复实现两次等待

代码实现

class Solver {public static void main(String[] args) {CyclicBarrier barrier=new CyclicBarrier(10);for (int i=0;i<10;i++){//开启10个线程模拟10个求职者new Thread(new JobHunt(barrier)).start();}}
}class JobHunt implements Runnable {private CyclicBarrier cyclicBarrier;public JobHunt(CyclicBarrier cyclicBarrier) {this.cyclicBarrier = cyclicBarrier;}@Overridepublic void run() {//赶来公司路上doOnTheWay();//到公司后,看人是否到齐,如果没有到齐,就阻塞,// 到齐了就开始笔试try {System.out.println(Thread.currentThread().getName()+" 已经来公司了...");cyclicBarrier.await();doWriteExam();System.out.println(Thread.currentThread().getName()+" 笔试做完了....");cyclicBarrier.await();doInterview();System.out.println(Thread.currentThread().getName()+"  面试完啦.....");} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}}/*** 模拟在路上方法*/public void doOnTheWay(){doCostTime(2000);}/*** 模拟笔试过程*/public void doWriteExam(){doCostTime(3000);}/*** 模拟面试过程*/public void doInterview(){doCostTime(5000);}private void doCostTime(int time){Random random=new Random();try {//随机休眠时间int count=random.nextInt(time);// System.out.println(count);Thread.sleep(count);} catch (InterruptedException e) {e.printStackTrace();}}}

测试截图

从截图中我们可以看出,CyclicBarrier 实现了大家一起等待,直至人到齐了再去一起做笔试或者面试。
在这里插入图片描述

方案二

由于两次等待结束后,打印的消息不一样。所以我们采用两个 CyclicBarrier。分别传入不同的 barrierAction,来实现自定义的 等待结束后的打印事件。

代码实现

class Solver {public static void main(String[] args) {//将笔试等待的回调函数传入CyclicBarrier barrierOnWriteExam=new CyclicBarrier(10,new BarrierActionOnWriteExam());//将面试等待的回调函数传入CyclicBarrier barrierOnInterview=new CyclicBarrier(10,new BarrierActionOnInterview());for (int i=0;i<10;i++){//开启10个线程模拟10个求职者new Thread(new JobHunt(barrierOnWriteExam,barrierOnInterview)).start();}}
}class JobHunt implements Runnable {private CyclicBarrier cyclicBarrierOnWriteExam;private CyclicBarrier cyclicBarrierOnInterview;public JobHunt(CyclicBarrier cyclicBarrierOnWriteExam,CyclicBarrier cyclicBarrierOnInterview) {this.cyclicBarrierOnWriteExam = cyclicBarrierOnWriteExam;this.cyclicBarrierOnInterview=  cyclicBarrierOnInterview;}@Overridepublic void run() {//赶来公司路上doOnTheWay();//到公司后,看人是否到齐,如果没有到齐,就阻塞,// 到齐了就开始笔试try {System.out.println(Thread.currentThread().getName()+" 已经来公司了...");cyclicBarrierOnWriteExam.await();doWriteExam();System.out.println(Thread.currentThread().getName()+" 笔试做完了....");cyclicBarrierOnInterview.await();doInterview();System.out.println(Thread.currentThread().getName()+"  面试完啦.....");} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}}/*** 模拟在路上方法*/public void doOnTheWay(){doCostTime(2000);}/*** 模拟笔试过程*/public void doWriteExam(){doCostTime(3000);}/*** 模拟面试过程*/public void doInterview(){doCostTime(5000);}private void doCostTime(int time){Random random=new Random();try {//随机休眠时间int count=random.nextInt(time);// System.out.println(count);Thread.sleep(count);} catch (InterruptedException e) {e.printStackTrace();}}}class BarrierActionOnWriteExam implements Runnable{@Overridepublic void run() {//自定义等待完成后的回调函数System.out.println("大家人到齐了,开始笔试吧");}}class BarrierActionOnInterview implements Runnable{@Overridepublic void run() {//自定义等待完成后的回调函数System.out.println("大家人到齐了,开始面试吧");}
}

测试打印

通过打印结果可以看到,首先是能正确实现效果。其次 是通过传入 回调事件参数给 CyclicBarrier,可以很方便实现 自己的业务逻辑。
在这里插入图片描述

总结

虽然 CountDownLatch 和CyclicBarrier 都能实现多个线程一起等待然后一起做某些事情。
CountDownLatch 更多的是 一个主线程等待 分支线程完成。然后主线程去做其他事情。
CyclicBarrier 是 大家分别做某些事情,等每个人都做完后,大家再一起去做另外一件事情。
并且两者实现的 原理完全不同。
希望通过本文大家能对 CyclicBarrier 有个更加理性的认识。多敲敲小demo。看能否有优化的地方。这样才能更好的理解。
CountDownLatch 学习的地址:
https://blog.csdn.net/echohuangshihuxue/article/details/129280219

http://www.lryc.cn/news/25814.html

相关文章:

  • Android 12.0 Settings 去掉打开开发者模式和USB调试模式的广播
  • OSI七层网络模型和TCP/IP四层网络模型的异同
  • 接口测试必备技能 - 加密和签名
  • JVM虚拟机概述(1)
  • 学习.NET MAUI Blazor(七)、实现一个真正的ChatGPT聊天应用
  • Django框架学习
  • JavaSE21-集合1-set
  • Web版和客户端哪种SQL工具更好?ChatGPT有话要说
  • 从客户端的角度来看移动端IM即时通讯的消息可靠性和送达机制
  • 2023年java春招面试题及答案
  • Django学习——基础篇(上)
  • 研报精选230302
  • Unity心得
  • TryHackMe-Binex
  • 外贸人如何写出优秀的开发信?附详细思路
  • python自学之《21天学通Python》(18)——第21章 案例2 Python搞定大数据
  • 面试问题【数据库】
  • Allegro如何输出钻孔表操作指导
  • 消息队列 面试题 整理
  • 【Java】对象比较大小
  • 发票自动OCR识别并录入模板 3分钟免费配置
  • Dubbo 配置说明
  • 英飞凌TCxxx实战系列01_Alarm处理
  • 飞桨全量支持业内AI科学计算工具——DeepXDE!
  • 【c++基础】
  • 语音识别技术对比分析
  • Idea git 回滚远程仓库版本
  • vscode C++配置
  • 【微电网_储能】基于启发式状态机策略和线性程序策略优化方法的微电网中的储能研究【给定系统约束和定价的情况下】(Matlab代码实现)
  • rk3288-android8-IR-mouse