CyclicBarrier(循环屏障)源码解读与使用

2024-04-23 23:12

本文主要是介绍CyclicBarrier(循环屏障)源码解读与使用,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

🏷️个人主页:牵着猫散步的鼠鼠 

🏷️系列专栏:Java全栈-专栏

🏷️个人学习笔记,若有缺误,欢迎评论区指正 

 

目录

1. 前言

2. 什么是CyclicBarrier?

3. CyclicBarrier与CountDownLatch的区别

4. CyclicBarrier底层原理实现

5. CyclicBarrier的使用

6. 总结 


1. 前言

我们在前面的文章讲解了AQS的底层原理,以及CountDownLatch(倒计时器),Semaphore(信号量)等同步工具类的底层原理与使用,甚至基于AQS手搓了一个简易的线程同步器,今天我们就继续来学习一个新的线程同步器CyclicBarrier(循环屏障),他与前两者相比其内部没有使用AQS。

2. 什么是CyclicBarrier?

CyclicBarrier 是 Java 中的一个线程同步工具类,它允许一组线程互相等待,直到所有线程都达到某个屏障点(同步点)后才继续执行。这个屏障被称为循环屏障,因为它可以在释放等待线程后重新使用。

当线程调用 CyclicBarrier 的 await() 方法时,该线程会被阻塞,直到所有线程都调用了 await() 方法。一旦最后一个线程调用了 await() 方法,所有等待的线程将被释放,并且可以选择性地执行一个预先定义好的 Runnable 任务。这个机制特别适合用于并行计算中,例如,当多个线程需要处理任务的多个部分,并且必须等待所有部分都完成才能进行下一步操作时。

CyclicBarrier 的一个特点是它可以循环使用。一旦所有线程都通过了屏障,CyclicBarrier 可以重置并再次使用。

3. CyclicBarrier与CountDownLatch的区别

有同学可能会感觉CyclicBarrier和CountDownLatch功能很相像,都是让线程等待其他线程,但他们之间有如下区别:

  1. 计数器是否可重置:

    • CyclicBarrier 的计数器可以重置,因此它可以被多次使用。一旦所有的线程都到达了屏障点,计数器会自动重置,CyclicBarrier 可以再次使用。
    • CountDownLatch 的计数器只能被使用一次,一旦计数器到达零,就不能再重置或重新使用。
  2. 使用场景:

    • CyclicBarrier 通常用于一组线程必须相互等待,直到所有线程都达到某个屏障点的情况。它适合用于模拟并发问题的场景,例如多线程计算,每个线程处理一部分数据,然后等待所有线程完成再进行汇总。
    • CountDownLatch 通常用于一个或多个线程等待其他线程完成某些操作的情况。它适用主线程等待一系列的子任务线程完成这种场景。

以及底层实现不同,CountDownLatch内部是通过AQS和CAS操作实现的,而CyclicBarrier的实现原理我们接下来会讲解。

4. CyclicBarrier底层原理实现

在CyclicBarrier有两个成员变量分别为parties,count,前者代表每次拦截的线程数量,后者是计数器,每有一个线程执行到同步点时,count减1,当count值变为0时说明所有线程都走到了同步点,这时就可以尝试执行我们在构造方法中设计的任务啦。

//每次拦截的线程数
private final int parties;
//计数器
private int count;//一个参数的构造
public CyclicBarrier(int parties) {this(parties, null);
}
//多参构造,parties为拦截线程数,barrierAction这个 Runnable会在 
// CyclicBarrier 的计数器为 0 的时候执行,也就是所有线程都完成任务后执行
public CyclicBarrier(int parties, Runnable barrierAction) {if (parties <= 0) throw new IllegalArgumentException();this.parties = parties;this.count = parties;this.barrierCommand = barrierAction;
}

每个线程通过调用await方法告诉CyclicBarrier已经到达屏障,然后进行阻塞等待,知道count等于0,所有线程都到达了屏障,因此,我们跟入await方法的源码中去看一下,在dowait(boolean timed, long nanos),可以通过时间参数来设置阻塞的时间,默认为false。

public int await() throws InterruptedException, BrokenBarrierException {try {// false参数代表不设置阻塞超时时间return dowait(false, 0L);} catch (TimeoutException toe) {throw new Error(toe); // cannot happen}
}
//await方法内部,继续调用dowait方法实现功能
private int dowait(boolean timed, long nanos)throws InterruptedException, BrokenBarrierException,TimeoutException {final ReentrantLock lock = this.lock;// 1.使用重入锁对操作上锁lock.lock();try {final Generation g = generation;if (g.broken)throw new BrokenBarrierException();// 2.如果线程中断了,抛出异常,唤醒其他线程if (Thread.interrupted()) {//打破屏障,唤醒全部等待的线程breakBarrier();throw new InterruptedException();}// 3.cout减1int index = --count;// 4.当 count 数量减为 0 之后说明最后一个线程已经到达栅栏了,// 这时执行我们预先定义好的 Runnable 任务,唤醒全部的等待线程,重置count值为partiesif (index == 0) {  // trippedboolean ranAction = false;try {final Runnable command = barrierCommand;if (command != null)command.run();ranAction = true;// 将 count 重置为 parties 属性的初始化值// 唤醒之前等待的线程// 下一波执行开始nextGeneration();return 0;} finally {if (!ranAction)breakBarrier();}}// 超时时间判断,如果没有配置超时时间就会之间等待for (;;) {try {if (!timed)// 使用lock接口提供的Condition信号量让当前线程阻塞等待trip.await();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();}
}

这么长的源码,其实原理很简单,我给大家总结下总体逻辑(可结合代码注解阅读)

内部通过使用重入锁上锁来保证对count修改的原子性,每次线程调用await后,都会进行--count操作,直到 count 为0时,会去执行我们预先定义好的 Runnable 任务command,然后唤醒线程继续向下执行,并且重置当前计数器,所以它能处理循环使用的场景。

简言之:CyclicBarrier内部通过维护一个int类型的count变量代表还需要等待到达的线程数,通过ReentrantLock保证了内部修改操作的原子性,通过ReentrantLock提供的Condition实现了线程的等待与唤醒,每次调用await都会让count-1,直到count变为0,就唤醒所有等待中的线程,并执行预设的Runnable任务,并重置同步器中count等数据,让该同步器可以被循环利用

5. CyclicBarrier的使用

大致的了解了CyclicBarrier的原理之后,我们写个小demo测试一下它如何使用

public class Test {public static void main(String[] args) {int numberOfThreads = 3; // 线程数量CyclicBarrier barrier = new CyclicBarrier(numberOfThreads, () -> {// 当所有线程都到达障碍点时执行的操作System.out.println("所有线程都已到达屏障,进入下一阶段");});for (int i = 0; i < numberOfThreads; i++) {new Thread(new Task(barrier), "Thread " + (i + 1)).start();}}static class Task implements Runnable {private final CyclicBarrier barrier;public Task(CyclicBarrier barrier) {this.barrier = barrier;}@Overridepublic void run() {try {System.out.println(Thread.currentThread().getName() + " 正在屏障处等待");barrier.await(); // 等待所有线程到达障碍点System.out.println(Thread.currentThread().getName() + " 已越过屏障.");} catch (Exception e) {e.printStackTrace();}}}
}

执行结果:

Thread 2 正在屏障处等待
Thread 1 正在屏障处等待
Thread 3 正在屏障处等待
所有线程都已到达屏障,进入下一阶段
Thread 3 已越过屏障.
Thread 1 已越过屏障.
Thread 2 已越过屏障.

6. 总结 

CyclicBarrier可以让一组线程互相等待,直到最后一个线程也准备就绪后,它们才能继续运行。就好比几个朋友约好一起吃晚餐,必须等到所有人到齐后才能入座就餐。CyclicBarrier实现了这种"集体出发"的功能,每次所有线程就位后,它们可以执行一个预先指定的任务,然后继续向前推进。

有趣的是,CyclicBarrier与CountDownLatch不同,它可以重复使用。这也从侧面体现了两者在设计理念上的差异。

借鉴了AQS的精髓,CyclicBarrier内部通过ReentrantLock和Condition实现了高效、安全的线程同步和通信。count变量作为"计数器",当减至0时,所有线程被唤醒,重置计数,使其可以被循环利用。

肝文章真的很累,特别是阅读源码,如果有帮助的话就多多点赞支持我吧~ 

这篇关于CyclicBarrier(循环屏障)源码解读与使用的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/930146

相关文章

Java中流式并行操作parallelStream的原理和使用方法

《Java中流式并行操作parallelStream的原理和使用方法》本文详细介绍了Java中的并行流(parallelStream)的原理、正确使用方法以及在实际业务中的应用案例,并指出在使用并行流... 目录Java中流式并行操作parallelStream0. 问题的产生1. 什么是parallelS

Linux join命令的使用及说明

《Linuxjoin命令的使用及说明》`join`命令用于在Linux中按字段将两个文件进行连接,类似于SQL的JOIN,它需要两个文件按用于匹配的字段排序,并且第一个文件的换行符必须是LF,`jo... 目录一. 基本语法二. 数据准备三. 指定文件的连接key四.-a输出指定文件的所有行五.-o指定输出

Linux jq命令的使用解读

《Linuxjq命令的使用解读》jq是一个强大的命令行工具,用于处理JSON数据,它可以用来查看、过滤、修改、格式化JSON数据,通过使用各种选项和过滤器,可以实现复杂的JSON处理任务... 目录一. 简介二. 选项2.1.2.2-c2.3-r2.4-R三. 字段提取3.1 普通字段3.2 数组字段四.

Linux kill正在执行的后台任务 kill进程组使用详解

《Linuxkill正在执行的后台任务kill进程组使用详解》文章介绍了两个脚本的功能和区别,以及执行这些脚本时遇到的进程管理问题,通过查看进程树、使用`kill`命令和`lsof`命令,分析了子... 目录零. 用到的命令一. 待执行的脚本二. 执行含子进程的脚本,并kill2.1 进程查看2.2 遇到的

详解SpringBoot+Ehcache使用示例

《详解SpringBoot+Ehcache使用示例》本文介绍了SpringBoot中配置Ehcache、自定义get/set方式,并实际使用缓存的过程,文中通过示例代码介绍的非常详细,对大家的学习或者... 目录摘要概念内存与磁盘持久化存储:配置灵活性:编码示例引入依赖:配置ehcache.XML文件:配置

Java 虚拟线程的创建与使用深度解析

《Java虚拟线程的创建与使用深度解析》虚拟线程是Java19中以预览特性形式引入,Java21起正式发布的轻量级线程,本文给大家介绍Java虚拟线程的创建与使用,感兴趣的朋友一起看看吧... 目录一、虚拟线程简介1.1 什么是虚拟线程?1.2 为什么需要虚拟线程?二、虚拟线程与平台线程对比代码对比示例:三

k8s按需创建PV和使用PVC详解

《k8s按需创建PV和使用PVC详解》Kubernetes中,PV和PVC用于管理持久存储,StorageClass实现动态PV分配,PVC声明存储需求并绑定PV,通过kubectl验证状态,注意回收... 目录1.按需创建 PV(使用 StorageClass)创建 StorageClass2.创建 PV

Redis 基本数据类型和使用详解

《Redis基本数据类型和使用详解》String是Redis最基本的数据类型,一个键对应一个值,它的功能十分强大,可以存储字符串、整数、浮点数等多种数据格式,本文给大家介绍Redis基本数据类型和... 目录一、Redis 入门介绍二、Redis 的五大基本数据类型2.1 String 类型2.2 Hash

Redis中Hash从使用过程到原理说明

《Redis中Hash从使用过程到原理说明》RedisHash结构用于存储字段-值对,适合对象数据,支持HSET、HGET等命令,采用ziplist或hashtable编码,通过渐进式rehash优化... 目录一、开篇:Hash就像超市的货架二、Hash的基本使用1. 常用命令示例2. Java操作示例三

Linux创建服务使用systemctl管理详解

《Linux创建服务使用systemctl管理详解》文章指导在Linux中创建systemd服务,设置文件权限为所有者读写、其他只读,重新加载配置,启动服务并检查状态,确保服务正常运行,关键步骤包括权... 目录创建服务 /usr/lib/systemd/system/设置服务文件权限:所有者读写js,其他