CyclicBarrier源码分析

概述

CyclicBarrier是一个同步辅助类,它允许一组线程相互等待,直到达到某个公共屏障点。并且在释放等待线程之后,CyclicBarrier是可以重复使用的。

简单使用

下面这段代码利用了CyclicBarrier来使得线程创建后相互等待,直到所有的线程都准备好,以此来使多个线程同时执行。

 public class CyclicBarrierTest {

    public static void main(String[] args) {
        CyclicBarrierTest cyclicBarrierTest=new CyclicBarrierTest();
        cyclicBarrierTest.runThread();
    }
    //有10个线程需要相互等待
    CyclicBarrier cyclicBarrier=new CyclicBarrier(10);

    /**
     * 创建一个线程
     * @return
     */
    private Thread createThread(int i){
        Thread thread=new Thread(new Runnable() {
            @Override
            public void run() {
                try {

                //线程在此相互等待,直到所有线程都准备好
                    cyclicBarrier.await();
                    System.out.println("thread"+Thread.currentThread().getName()+"准备完毕"+System.currentTimeMillis());
                }catch (InterruptedException e){
                    e.printStackTrace();
                }catch (BrokenBarrierException e){
                    e.printStackTrace();
                }

            }
        });
        thread.setName("thread-"+i);
        return  thread;
    }

    public void runThread(){
        ExecutorService executorService= Executors.newFixedThreadPool(10);

        try {
            for(int i=0;i<10;i++){
                Thread.sleep(100);
                executorService.submit(createThread(i));
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }

    }
}

源码分析

核心属性

    private static class Generation {
        //是否被销毁
        boolean broken = false;//false代表没被销毁
    }

    /** The lock for guarding barrier entry ,守护入口的锁*/
    private final ReentrantLock lock = new ReentrantLock();
    /** Condition to wait on until tripped,等待条件 */
    private final Condition trip = lock.newCondition();
    /** The number of parties,要屏障的线程数 */
    private final int parties;
    /* The command to run when tripped ,当线程都到待barrier,需要运行的内容*/
    private final Runnable barrierCommand;
    /** The current generation ,记录当前barrier状态的对象*/
    private Generation generation = new Generation();

    /**
     * Number of parties still waiting. Counts down from parties to 0
     * on each generation.  It is reset to parties on each new
     * generation or when broken.
     *当前等待barrier到达的线程的数量
     */
    private int count;

重要方法分析

await方法

int await()方法的具体实现如下:

    public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }

从这个方法可以看出,实际上起作用的就是dowait(false, 0L);.
那我们来看一下dowait(false, 0L);的具体实现:

private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        final ReentrantLock lock = this.lock;
        //获取锁,整段代码都使用该锁进行同步
        lock.lock();
        try {
            //获取当前的generation
            final Generation g = generation;

            if (g.broken)
                throw new BrokenBarrierException();

            if (Thread.interrupted()) {
                breakBarrier(); //如果线程被中断,就会终止Barrier,唤醒所有的等待线程
                throw new InterruptedException();
            }

            //count就是我们实例化CyclicBarrier时传入的值
            //此时index代表当前是最后几个等待的线程
            int index = --count;
            if (index == 0) {  // tripped
                //如过当前线程是最后一个等待的线程
                //它都已经调用await,说明所有线程都已经到达
                //屏障点了,可以唤醒所有线程了
                boolean ranAction = false;
                try {
                //如果有barrierCommand,就运行它
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    //更新Barrier状态,并唤醒所有线程
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }

            // loop until tripped, broken, interrupted, or timed out
            //自旋等待,直到所有线程都到达屏障点
            //或者发生中断
            //或者generation被销毁
            //或者超时
            for (;;) {
                try {
                    if (!timed)
                        trip.await(); //在此利用lock的Condition阻塞,当前线程
                    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(); //超时就销毁当前Barrier
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

在这个方法中,还有几个比较重要的方法。
用于销毁Barrier的void breakBarrier()方法

    private void breakBarrier() {
    //将当前gengeration标记为弃用状态
        generation.broken = true;
        count = parties;//将等待barrier的线程数量,恢复到之前的值
        trip.signalAll(); //唤醒锁上的Condition上等待的所有线程
    }

用于重置CyclicBarrier和唤醒所有等待线程的void nextGeneration()方法实现如下:

    private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll(); //唤醒所有的等待线程
        // set up next generation
        count = parties;//将等待线程数复原,以便CyclicBarrier下次重复使用
        generation = new Generation(); //复原generation
    }

整体看下来,这个CyclicBarrier的实现还是比较简单,我们在实例化CyclicBarrier的时候就指定了一个需要相互等待的线程数。每当一个线程调用await方法的时候,都会去判断,自己是不是最后一个线程,如果自己是最后一个线程,那么说明其它线程都在阻塞等待自己,那么就去唤醒所有等待的线程。如果自己不是最后一个线程,那么就需要去等待其它的线程,那么就去自旋,或者阻塞。

在整个源码中比较重要的一点就是CyclicBarrier内部利用了一个ReentrantLock利用它来对代码块加锁,让线程在它的Condition上阻塞。
每个CyclicBarrier内部都维护了一个Generation对象,它主要是记录当前CyclicBarrier的状态,即是否被弃用。因为CyclicBarrier是可以重复使用的,因此在所有线程都到达屏障点的时候,会调用nextGeneration()来重置整个CyclicBarrier,方便下次使用。

还需要注意的是,CyclicBarrier是会响应中断,一旦发生中断,就会重置CyclicBarrier,并唤醒等待的线程。

原文地址:https://www.cnblogs.com/zofun/p/12206913.html

时间: 2024-10-27 05:27:00

CyclicBarrier源码分析的相关文章

concurrent(四)同步屏障 CyclicBarrier &amp; 源码分析

参考文档:Java多线程系列--"JUC锁"10之 CyclicBarrier原理和示例:https://www.cnblogs.com/skywang12345/p/3533995.html简介CyclicBarrier是一个同步辅助类,允许一组线程互相等待,直到到达某个公共屏障点 (common barrier point).因为该 barrier 在释放等待线程后可以重用,所以称它为循环的 barrier.基于ReentrantLock实现举个栗子 /** * 简单模拟一下对战平

【JUC】JDK1.8源码分析之CyclicBarrier(四)

一.前言 有了前面分析的基础,现在,接着分析CyclicBarrier源码,CyclicBarrier类在进行多线程编程时使用很多,比如,你希望创建一组任务,它们并行执行工作,然后在进行下一个步骤之前等待,直至所有的任务都完成,和join很类似,下面,开始分析源码. 二.CyclicBarrier数据结构 分析源码可以知道,CyclicBarrier底层是基于ReentrantLock和AbstractQueuedSynchronizer来实现的,所以,CyclicBarrier的数据结构也依托

【JDK源码分析】通过源码分析CyclicBarrier

前言 CyclicBarrier它是什么?一个同步辅助类,它允许一组线程互相等待,直到到达某个公共屏障点.类似于朋友之间联系要在中午聚个会,几个朋友全部到齐后才开始喝酒吃菜. 源码 CyclicBarrier属性和构造器 public class CyclicBarrier { // 互斥锁 private final ReentrantLock lock = new ReentrantLock(); // 条件等待 private final Condition trip = lock.new

并发编程(七)——AbstractQueuedSynchronizer 之 CountDownLatch、CyclicBarrier、Semaphore 源码分析

这篇,我们的关注点是 AQS 最后的部分,共享模式的使用.本文先用 CountDownLatch 将共享模式说清楚,然后顺着把其他 AQS 相关的类 CyclicBarrier.Semaphore 的源码一起过一下. CountDownLatch CountDownLatch 这个类是比较典型的 AQS 的共享模式的使用,这是一个高频使用的类.使用方法在前面一篇文章中有介绍 并发编程(二)—— CountDownLatch.CyclicBarrier和Semaphore 使用例子 我们看下 Do

CountDownLatch &amp; CyclicBarrier源码Android版实现解析

CountDownLatch CountDownLatch允许一条或者多条线程等待直至其它线程完成以系列的操作的辅助同步器. 用一个指定的count值对CountDownLatch进行初始化.await方法会阻塞,直至因为调用countDown方法把当前的count降为0,在这以后,所有的等待线程会被释放,并且在这以后的await调用将会立即返回.这是一个一次性行为--count不能被重置.如果你需要一个可以重置count的版本,考虑使用CyclicBarrier. 其实本类实现非常简单,和Re

JDK源码分析之concurrent包(一) -- Executor架构

Java5新出的concurrent包中的API,是一些并发编程中实用的的工具类.在高并发场景下的使用非常广泛.笔者在这做了一个针对concurrent包中部分常用类的源码分析系列.本系列针对的读者是已经对并发包中的Executor框架和工具类有所了解并懂得如何使用的人群,如果对并发包还不了解的朋友,请先做些了解.网上对这方面的讲述有丰富的资源. 本篇博文是第一期,首先对Executor架构做一个概述.这里只简单介绍接口和类的继承.使用关系. 盗用一张类图来描述结构: 解析: Executor是

【JUC】JDK1.8源码分析之CountDownLatch(五)

一.前言 分析完了CyclicBarrier后,下面分析CountDownLatch,CountDownLatch用于同步一个或多个任务,强制他们等待由其他任务执行的一组操作完成.CountDownLatch典型的用法是将一个程序分为n个互相独立的可解决任务,并创建值为n的CountDownLatch.当每一个任务完成时,都会在这个锁存器上调用countDown,等待问题被解决的任务调用这个锁存器的await,将他们自己拦住,直至锁存器计数结束.下面开始分析源码. 二.CountDownLatc

【JUC】JDK1.8源码分析之Semaphore(六)

一.前言 分析了CountDownLatch源码后,下面接着分析Semaphore的源码.Semaphore称为计数信号量,它允许n个任务同时访问某个资源,可以将信号量看做是在向外分发使用资源的许可证,只有成功获取许可证,才能使用资源.下面开始分析Semaphore的源码. 二.Semaphore的数据结构 分析源码可以知道,Semaphore底层是基于AbstractQueuedSynchronizer来实现的,所以,Semaphore的数据结构也依托于AQS的数据结构,在前面对AQS的分析中

jdk源码分析总览

今天看到了一个源码分析按照重要性排序的例子, 这里拿过来用了,之后按照这个顺序不断的完善源码的内容. 引用的出处忘记了(对作者说声抱歉) 很多java开发的小伙伴都会阅读jdk源码,然而确不知道应该从哪读起.以下为小编整理的通常所需阅读的源码范围. 标题为包名,后面序号为优先级1-4,优先级递减 1.java.lang 1) Object 12) String 13) AbstractStringBuilder 14) StringBuffer 15) StringBuilder 16) Boo