1. 首页
  2. 死磕java容器

第49篇:死磕 java同步系列之CyclicBarrier源码解析——有图有真相

问题

(1)CyclicBarrier是什么?

(2)CyclicBarrier具有什么特性?

(3)CyclicBarrier与CountDownLatch的对比?

简介

CyclicBarrier,回环栅栏,它会阻塞一组线程直到这些线程同时达到某个条件才继续执行。它与CountDownLatch很类似,但又不同,CountDownLatch需要调用countDown()方法触发事件,而CyclicBarrier不需要,它就像一个栅栏一样,当一组线程都到达了栅栏处才继续往下走。

使用方法

  public class CyclicBarrierTest {
        public static void main(String[] args) {
            CyclicBarrier cyclicBarrier = new CyclicBarrier(3);

            for (int i = 0; i < 3; i++) {
                new Thread(()->{
                    System.out.println("before");
                    try {
                        cyclicBarrier.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                    System.out.println("after");
                }).start();
            }
        }
    }    

这段方法很简单,使用一个CyclicBarrier使得三个线程保持同步,当三个线程同时到达cyclicBarrier.await();处大家再一起往下运行。

源码分析

主要内部类

  private static class Generation {
        boolean broken = false;
    }

Generation,中文翻译为代,一代人的代,用于控制CyclicBarrier的循环使用。

比如,上面示例中的三个线程完成后进入下一代,继续等待三个线程达到栅栏处再一起执行,而CountDownLatch则做不到这一点,CountDownLatch是一次性的,无法重置其次数。

主要属性

  // 重入锁
    private final ReentrantLock lock = new ReentrantLock();
    // 条件锁,名称为trip,绊倒的意思,可能是指线程来了先绊倒,等达到一定数量了再唤醒
    private final Condition trip = lock.newCondition();
    // 需要等待的线程数量
    private final int parties;
    // 当唤醒的时候执行的命令
    private final Runnable barrierCommand;
    // 代
    private Generation generation = new Generation();
    // 当前这一代还需要等待的线程数
    private int count;

通过属性可以看到,CyclicBarrier内部是通过重入锁的条件锁来实现的,那么你可以脑补一下这个场景吗?

彤哥来脑补一下:假如初始时count = parties = 3,当第一个线程到达栅栏处,count减1,然后把它加入到Condition的队列中,第二个线程到达栅栏处也是如此,第三个线程到达栅栏处,count减为0,调用Condition的signalAll()通知另外两个线程,然后把它们加入到AQS的队列中,等待当前线程运行完毕,调用lock.unlock()的时候依次从AQS的队列中唤醒一个线程继续运行,也就是说实际上三个线程先依次(排队)到达栅栏处,再依次往下运行。

以上纯属彤哥脑补的内容,真实情况是不是如此呢,且往后看。

构造方法

  public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        // 初始化parties
        this.parties = parties;
        // 初始化count等于parties
        this.count = parties;
        // 初始化都到达栅栏处执行的命令
        this.barrierCommand = barrierAction;
    }

    public CyclicBarrier(int parties) {
        this(parties, null);
    }

构造方法需要传入一个parties变量,也就是需要等待的线程数。

await()方法

每个需要在栅栏处等待的线程都需要显式地调用await()方法等待其它线程的到来。

  public int await() throws InterruptedException, BrokenBarrierException {
        try {
            // 调用dowait方法,不需要超时
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }

    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();
            }

            // count的值减1
            int index = --count;
            // 如果数量减到0了,走这段逻辑(最后一个线程走这里)
            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();
                }
            }

            // 这个循环只有非最后一个线程可以走到
            for (;;) {
                try {
                    if (!timed)
                        // 调用condition的await()方法
                        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();

                // 正常来说这里肯定不相等
                // 因为上面打破栅栏的时候调用nextGeneration()方法时generation的引用已经变化了
                if (g != generation)
                    return index;

                // 超时检查
                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }
    private void nextGeneration() {
        // 调用condition的signalAll()将其队列中的等待者全部转移到AQS的队列中
        trip.signalAll();
        // 重置count
        count = parties;
        // 进入下一代
        generation = new Generation();
    }

dowait()方法里的整个逻辑分成两部分:

(1)最后一个线程走上面的逻辑,当count减为0的时候,打破栅栏,它调用nextGeneration()方法通知条件队列中的等待线程转移到AQS的队列中等待被唤醒,并进入下一代。

(2)非最后一个线程走下面的for循环逻辑,这些线程会阻塞在condition的await()方法处,它们会加入到条件队列中,等待被通知,当它们唤醒的时候已经更新换“代”了,这时候返回。

图解

nmajiexiyoutuyouzhenxiang_1.png

学习过前面的章节,看这个图很简单了,看不懂的同学还需要把推荐的内容好好看看哦^^

总结

(1)CyclicBarrier会使一组线程阻塞在await()处,当最后一个线程到达时唤醒(只是从条件队列转移到AQS队列中)前面的线程大家再继续往下走;

(2)CyclicBarrier不是直接使用AQS实现的一个同步器;

(3)CyclicBarrier基于ReentrantLock及其Condition实现整个同步逻辑;

彩蛋

CyclicBarrier与CountDownLatch的异同?

(1)两者都能实现阻塞一组线程等待被唤醒;

(2)前者是最后一个线程到达时自动唤醒;

(3)后者是通过显式地调用countDown()实现的;

(4)前者是通过重入锁及其条件锁实现的,后者是直接基于AQS实现的;

(5)前者具有“代”的概念,可以重复使用,后者只能使用一次;

(6)前者只能实现多个线程到达栅栏处一起运行;

(7)后者不仅可以实现多个线程等待一个线程条件成立,还能实现一个线程等待多个线程条件成立(详见CountDownLatch那章使用案例);

作者:彤哥读源码

来源:https://www.cnblogs.com/tong-yuan/p/CyclicBarrier.html


看完两件小事

如果你觉得这篇文章对你挺有启发,我想请你帮我两个小忙:

  1. 关注我们的 GitHub 博客,让我们成为长期关系
  2. 把这篇文章分享给你的朋友 / 交流群,让更多的人看到,一起进步,一起成长!
  3. 关注公众号 「方志朋」,公众号后台回复「666」 免费领取我精心整理的进阶资源教程
  4. JS中文网,Javascriptc中文网是中国领先的新一代开发者社区和专业的技术媒体,一个帮助开发者成长的社区,是给开发者用的 Hacker News,技术文章由为你筛选出最优质的干货,其中包括:Android、iOS、前端、后端等方面的内容。目前已经覆盖和服务了超过 300 万开发者,你每天都可以在这里找到技术世界的头条内容。

    本文著作权归作者所有,如若转载,请注明出处

    转载请注明:文章转载自「 Java极客技术学习 」https://www.javajike.com

    标题:第49篇:死磕 java同步系列之CyclicBarrier源码解析——有图有真相

    链接:https://www.javajike.com/article/1966.html

« 第50篇:死磕 java同步系列之StampedLock源码解析
第48篇:死磕 java同步系列之Phaser源码解析»

相关推荐

QR code