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

程序员文章站 2022-06-09 12:56:02

CyclicBarrier是什么？ CyclicBarrier具有什么特性？ CyclicBarrier与CountDownLatch的对比？ ......

问题

（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()方法处，它们会加入到条件队列中，等待被通知，当它们唤醒的时候已经更新换“代”了，这时候返回。

图解

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

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

总结

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

（2）cyclicbarrier不是直接使用aqs实现的一个同步器；

（3）cyclicbarrier基于reentrantlock及其condition实现整个同步逻辑；

彩蛋

cyclicbarrier与countdownlatch的异同？

（1）两者都能实现阻塞一组线程等待被唤醒；

（2）前者是最后一个线程到达时自动唤醒；

（3）后者是通过显式地调用countdown()实现的；

（4）前者是通过重入锁及其条件锁实现的，后者是直接基于aqs实现的；

（5）前者具有“代”的概念，可以重复使用，后者只能使用一次；

（6）前者只能实现多个线程到达栅栏处一起运行；

（7）后者不仅可以实现多个线程等待一个线程条件成立，还能实现一个线程等待多个线程条件成立（详见countdownlatch那章使用案例）；

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

问题

简介

使用方法

源码分析

主要内部类

主要属性

构造方法

await()方法

图解

总结

彩蛋

推荐阅读

死磕 java同步系列之Phaser源码解析