Java的CyclicBarrier循环屏障解析

 更新时间:2023年12月16日 09:57:51   作者:niulx111  
这篇文章主要介绍了Java的CyclicBarrier循环屏障解析,CyclicBarrier和CountDownLatch一样,是一个同步工具类,它允许一组线程相互等待直到达到某个common barrier point,在程序中CyclicBarrier是非常有用的,它适用于一组线程必须互相等待的情况,需要的朋友可以参考下

CyclicBarrier循环屏障

CyclicBarrier和CountDownLatch一样,是一个同步工具类,它允许一组线程相互等待直到达到某个common barrier point。

在程序中CyclicBarrier是非常有用的,它适用于一组线程必须互相等待的情况。barrier被称为周期是因为等待的线程在释放后可以重用。

构造函数

    public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }
    public CyclicBarrier(int parties) {
        this(parties, null);
    }

CyclicBarrier有两种构造方法,默认是指定一个线程数量,并把它赋给等待状态的线程count,还有一个barrierAction任务,它是在满足barrier时,所有的线程准备好之前,才执行这个runnable,这是最后一个线程来执行的。parties是表示必须有几个线程要达到barrier,count是表示当前未达到barrier的线程数量,只有count=0时,线程才能继续执行。

看一下它的示例。

public static void main(String[] args) {
		CyclicBarrier c = new CyclicBarrier(5,new Runnable() {
			@Override
			public void run() {
				System.out.println("开始");
			}
		});
		ExecutorService es = Executors.newCachedThreadPool();
		for (int i = 0; i < 5; i++) {
			es.execute(new Task(i, c));
		}
		es.shutdown();
	}
	static public class Task implements Runnable {
		private int i;
		private CyclicBarrier cyclicBarrier;
		public Task(int i,CyclicBarrier cyclicBarrier) {
			this.i = i;
			this.cyclicBarrier = cyclicBarrier;
		}
		@Override
		public void run() {
			System.out.println(i+"准备完毕");
			try {
				cyclicBarrier.await();
				Thread.sleep(100);
				System.out.println(i+"加入了");
			} catch (InterruptedException e) {	
				e.printStackTrace();
			} catch (BrokenBarrierException e) {
				e.printStackTrace();
			}
		}
	}

执行结果

0准备完毕
4准备完毕
3准备完毕
1准备完毕
2准备完毕
开始
4加入了
1加入了
2加入了
3加入了
0加入了

看一下它的阻塞await方法。

    /**
     * 如果当前线程不是最后一个达到的它就会休眠直到出现下面的任意一种情况:
     *  1. 最后一个线程达到
     *  2. 一些其他的线程中断了当前线程
     *  3. 一些其他的线程中断了正在等待的线程
     *  4. 一些其他线程在等待过程中超时了
     *  5. 一些其他的线程在barrier被重置
     *  当前线程在等待的过程中如果被中断直接会抛出异常并且当前线程的中断状态会被清除
     *  如果在等待的过程中一些线程被中断,其他的等待线程就会抛出异常,barrier就会被设置成broken为true的状态
     *  如果当前线程是最后一个达到的话,如果构造方法中有runnable任务,直接由最后一个线程去执行,其他线程继续等待
     *  在barrier action中,如果出现异常会由当前线程传播并且barrier被设置成broken为true的状态
     */
    public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // 永远不会进入catch
        }
    }

主要调用了dowait方法,看一下这个方法,方法比较长。

    /**
     * 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;
            if (index == 0) {  // tripped
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }
            // loop until tripped, broken, interrupted, or timed out
            for (;;) {
                try {
                    if (!timed)
                        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();
        }
    }

首先会创建一个静态内部类Generation,这个类中只有一个属性,broken,表示当前屏障是否被中断,默认为false,未中断.

如果当前broken为true,就会传播这个异常.

如果当前线程被中断,broken为中断状态,调用breakBarrier方法,把broken设为true,重置count.唤醒其他的线程.

    private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();
    }

如果前面的条件都不成立,那就count自减,表示当前线程已经完成了任务,当count=0的时候,表示所有的任务都已经完成,此时在判断是否构造函数中有传入的runnable,如果command不为空,总是最后一个线程去执行,如果在执行这个command的过程中出现异常,再次调用breakBarrier方法重置count.

执行成功,调用nextGeneration方法,唤醒所有在等待队列上的线程,重置count.再次把broken设为false.

    private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll();
        // set up next generation
        count = parties;
        generation = new Generation();
    }

如果当前count不为0,那就进行自旋,没有超时直接释放锁进入等待队列,如果线程在释放锁进入等待队列的过程中被中断直接重置count,抛出异常.

在自旋过程中,如果发生了broken状态的变化或者超时,直接抛出异常,并做相应的处理,最后释放锁.

总结

CyclicBarrier的count是可以重用的,这是与CountDownLatch最大的区别.

barrier中任务是由最有一个线程执行的,并且会执行构造方法中传入的任务.这个任务总是在到达barrier时执行.

到此这篇关于Java的CyclicBarrier循环屏障解析的文章就介绍到这了,更多相关CyclicBarrier循环屏障内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • 如何在java中正确使用注释

    如何在java中正确使用注释

    在编写程序时,经常需要添加一些注释,用以描述某段代码的作用。 一般来说,对于一份规范的程序源代码而言,注释应该占到源代码的 1/3 以上。下面我们来详细了解一下吧
    2019-06-06
  • 基于java实现简单发红包功能

    基于java实现简单发红包功能

    这篇文章主要为大家详细介绍了基于java实现简单发红包功能,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2020-11-11
  • 详解在springboot中使用Mybatis Generator的两种方式

    详解在springboot中使用Mybatis Generator的两种方式

    这篇文章主要介绍了详解在springboot中使用Mybatis Generator的两种方式,本文将介绍到在springboot的项目中如何去配置和使用MBG以及MBG生成代码的两种方式,非常具有实用价值,需要的朋友可以参考下
    2018-11-11
  • SpringBoot开发案例 分布式集群共享Session详解

    SpringBoot开发案例 分布式集群共享Session详解

    这篇文章主要介绍了SpringBoot开发案例 分布式集群共享Session详解,在分布式系统中,为了提升系统性能,通常会对单体项目进行拆分,分解成多个基于功能的微服务,可能还会对单个微服务进行水平扩展,保证服务高可用,需要的朋友可以参考下
    2019-07-07
  • spring-boot-maven-plugin 配置有啥用

    spring-boot-maven-plugin 配置有啥用

    这篇文章主要介绍了spring-boot-maven-plugin 配置是干啥的,这个是SpringBoot的Maven插件,主要用来打包的,通常打包成jar或者war文件,本文通过示例代码给大家介绍的非常详细,需要的朋友可以参考下
    2022-08-08
  • Mybatis往Mapper.xml文件中传递多个参数问题

    Mybatis往Mapper.xml文件中传递多个参数问题

    这篇文章主要介绍了Mybatis往Mapper.xml文件中传递多个参数问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2024-05-05
  • 两分钟解决IntelliJ IDEA中文乱码问题(推荐)

    两分钟解决IntelliJ IDEA中文乱码问题(推荐)

    这篇文章主要介绍了两分钟解决IntelliJ IDEA中文乱码问题,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2021-02-02
  • SpringBoot整合RocketMQ的详细过程

    SpringBoot整合RocketMQ的详细过程

    这篇文章主要介绍了SpringBoot整合RocketMQ的详细过程,本文分为三部分,第一部分实现SpringBoot与RocketMQ的整合,第二部分解决在使用RocketMQ过程中可能遇到的一些问题并解决他们,第三部分介绍如何封装RocketMQ以便更好地使用,需要的朋友可以参考下
    2023-04-04
  • java多线程-同步块实例讲解

    java多线程-同步块实例讲解

    本文主要介绍java多线程-同步块的知识,这里整理了相关的详细资料及简单示例代码,有兴趣的小伙伴可以参考下
    2016-09-09

最新评论